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 <noreply@anthropic.com>
This commit is contained in:
@@ -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"
|
||||
|
||||
Reference in New Issue
Block a user