From be783f97fa3283498887d922ea7d56845b28a65c Mon Sep 17 00:00:00 2001 From: starmen1265 Date: Mon, 16 Mar 2026 23:24:17 +0800 Subject: [PATCH] test(05-03): add ArchiveProcessor unit tests - Test archive copies to correct platform directory - Test filename format {YYYYMMDD}_{content_id}_analysis.md - Test skip when step already completed - Test detect existing archive by content_id - Test handle missing analysis file with error Co-Authored-By: Claude Opus 4.6 --- tests/test_processors/test_archive.py | 222 ++++++++++++++++++++++++++ 1 file changed, 222 insertions(+) create mode 100644 tests/test_processors/test_archive.py diff --git a/tests/test_processors/test_archive.py b/tests/test_processors/test_archive.py new file mode 100644 index 0000000..0af7e12 --- /dev/null +++ b/tests/test_processors/test_archive.py @@ -0,0 +1,222 @@ +"""Tests for ArchiveProcessor + +Tests that ArchiveProcessor: +- Copies md file to correct platform directory +- Uses correct filename format {YYYYMMDD}_{content_id}_analysis.md +- Skips when step already completed +- Detects existing archive by content_id +- Handles missing analysis file with error +""" + +import pytest +from datetime import datetime +from pathlib import Path +from unittest.mock import patch, MagicMock + +from kb_pipeline.processors.archive import ArchiveProcessor +from kb_pipeline.schemas import ArchiveInput, ArchiveOutput +from kb_pipeline.state import WorkflowStateManager + + +# ============================================================================= +# Fixtures +# ============================================================================= + + +@pytest.fixture +def archive_test_env(tmp_path): + """Create test environment for archive tests""" + # Create intermediate content directory + content_dir = tmp_path / "intermediate" / "bilibili_BV1test" + content_dir.mkdir(parents=True) + + # Create analysis file + analysis_file = content_dir / "analysis.md" + analysis_file.write_text("# Test Analysis\n\nContent here.") + + # Create archive directory + archive_dir = tmp_path / "archive" + archive_dir.mkdir(parents=True) + + # Create state manager + state_manager = WorkflowStateManager(content_dir) + + return { + "content_dir": content_dir, + "analysis_file": analysis_file, + "archive_dir": archive_dir, + "state_manager": state_manager, + } + + +# ============================================================================= +# ArchiveProcessor Tests +# ============================================================================= + + +class TestArchiveProcessorCopyToPlatformDir: + """Test that archive file is copied to correct platform directory""" + + @pytest.mark.asyncio + async def test_copies_to_platform_dir(self, archive_test_env): + """Test that archive file is copied to correct platform directory""" + mock_settings = MagicMock() + mock_settings.archive_path = str(archive_test_env["archive_dir"]) + + with patch('kb_pipeline.processors.archive.get_settings') as mock_get_settings: + mock_get_settings.return_value = mock_settings + + processor = ArchiveProcessor() + # Mock _ask_overwrite to return True (user says yes) in case of collision + processor._ask_overwrite = lambda f: True + + input_data = ArchiveInput( + content_id="BV1test", + platform="bilibili", + content_dir=str(archive_test_env["content_dir"]), + analysis_file=str(archive_test_env["analysis_file"]), + content_type="video" + ) + + result = await processor.process(input_data) + + assert result.success is True + assert result.skipped is False + assert "bilibili" in result.archive_file + assert result.archive_file.endswith("_analysis.md") + + +class TestArchiveProcessorFilenameFormat: + """Test that archive filename follows correct format""" + + @pytest.mark.asyncio + async def test_filename_format(self, archive_test_env): + """Test that archive filename follows {YYYYMMDD}_{content_id}_analysis.md format""" + mock_settings = MagicMock() + mock_settings.archive_path = str(archive_test_env["archive_dir"]) + + with patch('kb_pipeline.processors.archive.get_settings') as mock_get_settings: + mock_get_settings.return_value = mock_settings + + processor = ArchiveProcessor() + # Mock _ask_overwrite to return True (user says yes) in case of collision + processor._ask_overwrite = lambda f: True + + input_data = ArchiveInput( + content_id="BV1test", + platform="bilibili", + content_dir=str(archive_test_env["content_dir"]), + analysis_file=str(archive_test_env["analysis_file"]), + content_type="video" + ) + + result = await processor.process(input_data) + + # Check filename format + archive_path = Path(result.archive_file) + filename = archive_path.name + today = datetime.now().strftime("%Y%m%d") + expected = f"{today}_BV1test_analysis.md" + assert filename == expected + + +class TestArchiveProcessorSkipsCompleted: + """Test that archive step is skipped when already completed""" + + @pytest.mark.asyncio + async def test_skips_completed(self, archive_test_env): + """Test that archive step is skipped when already completed""" + mock_settings = MagicMock() + mock_settings.archive_path = str(archive_test_env["archive_dir"]) + + # Create state with archive already completed + state = archive_test_env["state_manager"].get_or_create( + content_id="BV1test", + platform="bilibili", + source_url="https://test.com" + ) + state.mark_step_completed("archive") + archive_test_env["state_manager"].save(state) + + with patch('kb_pipeline.processors.archive.get_settings') as mock_get_settings: + mock_get_settings.return_value = mock_settings + + processor = ArchiveProcessor(state_manager=archive_test_env["state_manager"]) + input_data = ArchiveInput( + content_id="BV1test", + platform="bilibili", + content_dir=str(archive_test_env["content_dir"]), + analysis_file=str(archive_test_env["analysis_file"]), + content_type="video" + ) + + result = await processor.process(input_data) + + assert result.success is True + assert result.skipped is True + + +class TestArchiveProcessorDetectsExisting: + """Test that archive processor detects existing archive by content_id""" + + @pytest.mark.asyncio + async def test_detects_existing_archive(self, archive_test_env): + """Test that archive processor detects existing archive by content_id""" + mock_settings = MagicMock() + mock_settings.archive_path = str(archive_test_env["archive_dir"]) + + # Create existing archive with different timestamp + platform_dir = archive_test_env["archive_dir"] / "bilibili" + platform_dir.mkdir(parents=True, exist_ok=True) + existing = platform_dir / "20260101_BV1test_analysis.md" + existing.write_text("Old content") + + with patch('kb_pipeline.processors.archive.get_settings') as mock_get_settings: + mock_get_settings.return_value = mock_settings + + processor = ArchiveProcessor() + # Mock _ask_overwrite to return False (user says no) + processor._ask_overwrite = lambda f: False + + input_data = ArchiveInput( + content_id="BV1test", + platform="bilibili", + content_dir=str(archive_test_env["content_dir"]), + analysis_file=str(archive_test_env["analysis_file"]), + content_type="video" + ) + + result = await processor.process(input_data) + + assert result.success is True + assert result.skipped is True + + +class TestArchiveProcessorHandlesMissingFile: + """Test that archive processor handles missing analysis file""" + + @pytest.mark.asyncio + async def test_handles_missing_file(self, archive_test_env): + """Test that archive processor handles missing analysis file""" + mock_settings = MagicMock() + mock_settings.archive_path = str(archive_test_env["archive_dir"]) + + with patch('kb_pipeline.processors.archive.get_settings') as mock_get_settings: + mock_get_settings.return_value = mock_settings + + processor = ArchiveProcessor() + # Mock _ask_overwrite to avoid stdin issues + processor._ask_overwrite = lambda f: True + + input_data = ArchiveInput( + content_id="BV1test", + platform="bilibili", + content_dir=str(archive_test_env["content_dir"]), + analysis_file="/nonexistent/analysis.md", # Missing file + content_type="video" + ) + + result = await processor.process(input_data) + + assert result.success is False + assert "not found" in result.error_message.lower()