From 360aee200f993245af3d76558c32a42c42a31532 Mon Sep 17 00:00:00 2001 From: starmen1265 Date: Mon, 16 Mar 2026 23:26:14 +0800 Subject: [PATCH] test(05-03): add Pipeline archive integration tests - Test pipeline includes archive step - Test archive uses platform directory structure - Test no date-based subdirectories in archive - Update default steps assertion to include archive Co-Authored-By: Claude Opus 4.6 --- tests/test_pipeline/test_pipeline.py | 194 ++++++++++++++++++++++++++- 1 file changed, 193 insertions(+), 1 deletion(-) diff --git a/tests/test_pipeline/test_pipeline.py b/tests/test_pipeline/test_pipeline.py index a6c74d9..988a02a 100644 --- a/tests/test_pipeline/test_pipeline.py +++ b/tests/test_pipeline/test_pipeline.py @@ -122,7 +122,7 @@ class TestPipelineConfig: config = PipelineConfig() assert hasattr(config, "steps") assert isinstance(config.steps, list) - assert config.steps == ["parse", "download", "asr", "analyze"] + assert config.steps == ["parse", "download", "asr", "analyze", "archive"] def test_schema_pipeline_config_allows_custom_steps(self): """PipelineConfig allows custom steps list""" @@ -303,3 +303,195 @@ class TestPipelineParseUrl: assert result.platform == "bilibili" assert result.content_type == "video" assert result.content_id == "BV1xx411c7mD" + + +# ============================================================================= +# Archive Integration Tests +# ============================================================================= + + +@pytest.fixture +def mock_processors_with_archive(): + """Create mock processors including archive""" + processors = [] + + for step_name in ["parse", "download", "asr", "analyze", "archive"]: + processor = MagicMock() + processor.STEP_NAME = step_name + processor.process = AsyncMock() + + if step_name == "parse": + processor.process.return_value = MagicMock( + success=True, + platform="bilibili", + content_type="video", + content_id="BV1test", + resolved_url="https://www.bilibili.com/video/BV1test", + original_url="https://www.bilibili.com/video/BV1test" + ) + elif step_name == "download": + processor.process.return_value = MagicMock( + success=True, + metadata_file="/tmp/metadata.json", + content_file="/tmp/content.json", + title="Test Video", + author="Test Author" + ) + elif step_name == "asr": + processor.process.return_value = MagicMock( + success=True, + tool="whisper", + has_subtitle=False, + output_files={"json": "/tmp/subtitle.json"} + ) + elif step_name == "analyze": + processor.process.return_value = MagicMock( + success=True, + analysis_file="/tmp/analysis.md", + summary="Test summary", + main_topics=["topic1"], + key_insights=["insight1"], + tags=["tag1"], + stats={} + ) + elif step_name == "archive": + processor.process.return_value = MagicMock( + success=True, + skipped=False, + platform="bilibili", + archive_file="/archive/bilibili/20260316_BV1test_analysis.md" + ) + + processors.append(processor) + + return processors + + +class TestPipelineArchiveStep: + """Test archive step integration in pipeline""" + + @pytest.mark.asyncio + async def test_pipeline_includes_archive_step(self, mock_processors_with_archive, tmp_path): + """Test that pipeline includes archive as final step""" + from kb_pipeline.pipeline import Pipeline + from kb_pipeline.schemas import PipelineConfig + + config = PipelineConfig( + steps=["parse", "download", "asr", "analyze", "archive"], + output_dir=tmp_path, + skip_completed=False + ) + + with patch("kb_pipeline.pipeline.get_settings") as mock_settings: + mock_settings.return_value = MagicMock(intermediate_path=str(tmp_path)) + + pipeline = Pipeline( + processors=mock_processors_with_archive, + config=config, + output_dir=tmp_path + ) + + # Mock _parse_url and _get_content_dir + with patch.object(pipeline, '_parse_url') as mock_parse, \ + patch.object(pipeline, '_get_content_dir') as mock_dir: + + from kb_pipeline.link_parser import ParsedLink + mock_parse.return_value = ParsedLink( + platform="bilibili", + content_type="video", + content_id="BV1test", + original_url="https://www.bilibili.com/video/BV1test", + resolved_url="https://www.bilibili.com/video/BV1test" + ) + mock_dir.return_value = tmp_path / "bilibili_BV1test" + + result = await pipeline.run("https://www.bilibili.com/video/BV1test") + + assert result.success is True + # Verify archive processor was called + archive_processor = next(p for p in mock_processors_with_archive if p.STEP_NAME == "archive") + archive_processor.process.assert_called_once() + + +class TestArchivePlatformDirectory: + """Test that archive files go to correct platform directory""" + + @pytest.mark.asyncio + async def test_archive_uses_platform_directory(self, tmp_path): + """Test that archive files go to {archive_path}/{platform}/ directory""" + from kb_pipeline.processors.archive import ArchiveProcessor + from kb_pipeline.schemas import ArchiveInput + + # Setup + archive_dir = tmp_path / "archive" + content_dir = tmp_path / "intermediate" / "bilibili_BV1test" + content_dir.mkdir(parents=True) + analysis_file = content_dir / "analysis.md" + analysis_file.write_text("# Test") + + # Mock settings - patch at base module level where get_settings is called in __init__ + with patch('kb_pipeline.processors.base.get_settings') as mock_get_settings: + mock_settings = MagicMock() + mock_settings.archive_path = str(archive_dir) + 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(content_dir), + analysis_file=str(analysis_file), + content_type="video" + ) + + result = await processor.process(input_data) + + assert result.success is True + # Verify platform subdirectory + archive_path = Path(result.archive_file) + assert archive_path.parent.name == "bilibili" + + +class TestArchiveNoDateSubdirectories: + """Test that archive does not create date-based subdirectories""" + + @pytest.mark.asyncio + async def test_archive_no_date_subdirectories(self, tmp_path): + """Test that archive does not create date-based subdirectories""" + from kb_pipeline.processors.archive import ArchiveProcessor + from kb_pipeline.schemas import ArchiveInput + + archive_dir = tmp_path / "archive" + content_dir = tmp_path / "intermediate" / "bilibili_BV1test" + content_dir.mkdir(parents=True) + analysis_file = content_dir / "analysis.md" + analysis_file.write_text("# Test") + + with patch('kb_pipeline.processors.base.get_settings') as mock_get_settings: + mock_settings = MagicMock() + mock_settings.archive_path = str(archive_dir) + 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(content_dir), + analysis_file=str(analysis_file), + content_type="video" + ) + + result = await processor.process(input_data) + + # Verify no date-based subdirectories + archive_path = Path(result.archive_file) + # Parent should be platform dir, grandparent should be archive base + assert archive_path.parent.parent == archive_dir + # Check that parent is just "bilibili", not a date + assert archive_path.parent.name == "bilibili"