diff --git a/tests/test_processors/test_analyze.py b/tests/test_processors/test_analyze.py new file mode 100644 index 0000000..2dc247e --- /dev/null +++ b/tests/test_processors/test_analyze.py @@ -0,0 +1,417 @@ +"""Tests for AnalyzeProcessor and AnalyzeInput/AnalyzeOutput schemas. + +Test coverage: +- AnalyzeInput schema validation +- AnalyzeOutput schema validation +- AnalyzeProcessor DeepAnalyzer integration +- AnalyzeProcessor state recording +""" + +import pytest +import json +from datetime import datetime +from unittest.mock import AsyncMock, MagicMock, patch +from pathlib import Path + +from kb_pipeline.schemas import AnalyzeInput, AnalyzeOutput + + +class TestAnalyzeInputSchema: + """Tests for AnalyzeInput schema validation.""" + + def test_schema_analyze_input_validates_required_fields(self): + """Test 1: AnalyzeInput validates required fields""" + input_data = AnalyzeInput( + content_id="BV1xx", + platform="bilibili", + metadata_file="/tmp/content/bilibili_BV1xx_metadata.json", + content_file="/tmp/content/bilibili_BV1xx.json", + content_type="video", + output_file="/tmp/content/bilibili_BV1xx_analysis.md" + ) + assert input_data.content_id == "BV1xx" + assert input_data.platform == "bilibili" + assert input_data.metadata_file == "/tmp/content/bilibili_BV1xx_metadata.json" + assert input_data.content_file == "/tmp/content/bilibili_BV1xx.json" + assert input_data.content_type == "video" + assert input_data.output_file == "/tmp/content/bilibili_BV1xx_analysis.md" + + def test_schema_analyze_input_requires_metadata_file(self): + """AnalyzeInput should require metadata_file""" + with pytest.raises(Exception): # ValidationError from pydantic + AnalyzeInput( + content_id="BV1xx", + platform="bilibili", + content_file="/tmp/content/bilibili_BV1xx.json", + content_type="video", + output_file="/tmp/content/bilibili_BV1xx_analysis.md" + # Missing metadata_file + ) + + def test_schema_analyze_input_requires_content_file(self): + """AnalyzeInput should require content_file""" + with pytest.raises(Exception): # ValidationError from pydantic + AnalyzeInput( + content_id="BV1xx", + platform="bilibili", + metadata_file="/tmp/content/bilibili_BV1xx_metadata.json", + content_type="video", + output_file="/tmp/content/bilibili_BV1xx_analysis.md" + # Missing content_file + ) + + def test_schema_analyze_input_requires_content_type(self): + """AnalyzeInput should require content_type""" + with pytest.raises(Exception): # ValidationError from pydantic + AnalyzeInput( + content_id="BV1xx", + platform="bilibili", + metadata_file="/tmp/content/bilibili_BV1xx_metadata.json", + content_file="/tmp/content/bilibili_BV1xx.json", + output_file="/tmp/content/bilibili_BV1xx_analysis.md" + # Missing content_type + ) + + def test_schema_analyze_input_requires_output_file(self): + """AnalyzeInput should require output_file""" + with pytest.raises(Exception): # ValidationError from pydantic + AnalyzeInput( + content_id="BV1xx", + platform="bilibili", + metadata_file="/tmp/content/bilibili_BV1xx_metadata.json", + content_file="/tmp/content/bilibili_BV1xx.json", + content_type="video" + # Missing output_file + ) + + +class TestAnalyzeOutputSchema: + """Tests for AnalyzeOutput schema validation.""" + + def test_schema_analyze_output_validates_required_fields(self): + """Test 2: AnalyzeOutput validates required fields""" + output = AnalyzeOutput( + success=True, + analysis_file="/tmp/content/bilibili_BV1xx_analysis.md", + summary="This is a test summary", + main_topics=["Topic 1", "Topic 2"], + key_insights=["Insight 1", "Insight 2"], + tags=["tag1", "tag2"], + stats={"segments_count": 10} + ) + assert output.success is True + assert output.analysis_file == "/tmp/content/bilibili_BV1xx_analysis.md" + assert output.summary == "This is a test summary" + assert output.main_topics == ["Topic 1", "Topic 2"] + assert output.key_insights == ["Insight 1", "Insight 2"] + assert output.tags == ["tag1", "tag2"] + assert output.stats == {"segments_count": 10} + + def test_schema_analyze_output_handles_video_content_type(self): + """Test 3: AnalyzeOutput handles video content type""" + output = AnalyzeOutput( + success=True, + analysis_file="/tmp/content/bilibili_BV1xx_analysis.md", + summary="Video summary", + main_topics=["Topic 1"], + key_insights=["Insight 1"], + tags=["video", "test"], + stats={"duration": 300, "segments": 15} + ) + assert output.success is True + assert "duration" in output.stats + assert output.stats["segments"] == 15 + + def test_schema_analyze_output_handles_article_content_type(self): + """Test 3: AnalyzeOutput handles article content type""" + output = AnalyzeOutput( + success=True, + analysis_file="/tmp/content/zhihu_123_analysis.md", + summary="Article summary", + main_topics=["Topic 1"], + key_insights=[], + tags=["article", "zhihu"], + stats={"word_count": 2000} + ) + assert output.success is True + assert "word_count" in output.stats + + def test_schema_analyze_output_handles_failure(self): + """AnalyzeOutput should handle failure case""" + output = AnalyzeOutput( + success=False, + error_message="AI API error", + analysis_file="", + summary="", + main_topics=[], + key_insights=[], + tags=[], + stats={} + ) + assert output.success is False + assert output.error_message == "AI API error" + + +class TestAnalyzeProcessorIntegration: + """Tests for AnalyzeProcessor DeepAnalyzer integration.""" + + @pytest.fixture + def mock_deep_analyzer(self): + """Create a mock DeepAnalyzer result.""" + from dataclasses import dataclass, field + from typing import List + + @dataclass + class AnalysisResult: + summary: str = "Test summary" + main_topics: List[str] = field(default_factory=lambda: ["Topic 1", "Topic 2"]) + key_insights: List[str] = field(default_factory=lambda: ["Insight 1"]) + recommended_tags: List[str] = field(default_factory=lambda: ["tag1", "tag2"]) + full_analysis: str = "## Summary\n\nTest analysis content" + original_text: str = "" + + return AnalysisResult() + + @pytest.fixture + def mock_state_manager(self): + """Create a mock WorkflowStateManager.""" + manager = MagicMock() + manager.load = MagicMock(return_value=None) + manager.save = MagicMock() + return manager + + @pytest.fixture + def sample_metadata(self, tmp_path): + """Create sample metadata file.""" + metadata = { + "title": "Test Video Title", + "author": "Test Author", + "pubtime": "2024-01-01", + "duration": 300, + "bvid": "BV1xx", + "platform": "bilibili", + "url": "https://www.bilibili.com/video/BV1xx" + } + metadata_file = tmp_path / "bilibili_BV1xx_metadata.json" + metadata_file.write_text(json.dumps(metadata, ensure_ascii=False), encoding="utf-8") + return str(metadata_file) + + @pytest.fixture + def sample_content(self, tmp_path): + """Create sample content file.""" + content = { + "text": "This is the full transcript of the video." + } + content_file = tmp_path / "bilibili_BV1xx.json" + content_file.write_text(json.dumps(content, ensure_ascii=False), encoding="utf-8") + return str(content_file) + + @pytest.mark.asyncio + async def test_analyze_loads_metadata_and_content_from_files( + self, mock_deep_analyzer, mock_state_manager, sample_metadata, sample_content, tmp_path + ): + """Test 1: AnalyzeProcessor loads metadata and content from files""" + from kb_pipeline.processors.analyze import AnalyzeProcessor + from kb_pipeline.schemas import AnalyzeInput + + output_file = str(tmp_path / "bilibili_BV1xx_analysis.md") + + input_data = AnalyzeInput( + content_id="BV1xx", + platform="bilibili", + metadata_file=sample_metadata, + content_file=sample_content, + content_type="video", + output_file=output_file + ) + + with patch("kb_pipeline.processors.base.get_settings") as mock_settings: + mock_settings.return_value = MagicMock() + with patch("kb_pipeline.processors.analyze.DeepAnalyzer") as MockDeepAnalyzer: + mock_analyzer_instance = MagicMock() + mock_analyzer_instance.analyze = MagicMock(return_value=mock_deep_analyzer) + MockDeepAnalyzer.return_value = mock_analyzer_instance + + processor = AnalyzeProcessor(state_manager=mock_state_manager) + result = await processor.process(input_data) + + assert result.success is True + # Verify DeepAnalyzer.analyze was called with loaded data + mock_analyzer_instance.analyze.assert_called_once() + call_args = mock_analyzer_instance.analyze.call_args + assert "title" in call_args[0][0] # metadata dict + assert "text" in call_args[0][1] # content dict + + @pytest.mark.asyncio + async def test_analyze_calls_deep_analyzer_with_correct_parameters( + self, mock_deep_analyzer, mock_state_manager, sample_metadata, sample_content, tmp_path + ): + """Test 2: AnalyzeProcessor calls DeepAnalyzer.analyze() with correct parameters""" + from kb_pipeline.processors.analyze import AnalyzeProcessor + from kb_pipeline.schemas import AnalyzeInput + + output_file = str(tmp_path / "bilibili_BV1xx_analysis.md") + + input_data = AnalyzeInput( + content_id="BV1xx", + platform="bilibili", + metadata_file=sample_metadata, + content_file=sample_content, + content_type="video", + output_file=output_file + ) + + with patch("kb_pipeline.processors.base.get_settings") as mock_settings: + mock_settings.return_value = MagicMock() + with patch("kb_pipeline.processors.analyze.DeepAnalyzer") as MockDeepAnalyzer: + mock_analyzer_instance = MagicMock() + mock_analyzer_instance.analyze = MagicMock(return_value=mock_deep_analyzer) + MockDeepAnalyzer.return_value = mock_analyzer_instance + + processor = AnalyzeProcessor(state_manager=mock_state_manager) + result = await processor.process(input_data) + + assert result.success is True + # Verify content_type was passed correctly + call_kwargs = mock_analyzer_instance.analyze.call_args + assert call_kwargs[0][2] == "video" # content_type parameter + + @pytest.mark.asyncio + async def test_analyze_saves_analysis_report_as_markdown( + self, mock_deep_analyzer, mock_state_manager, sample_metadata, sample_content, tmp_path + ): + """Test 3: AnalyzeProcessor saves analysis report as Markdown""" + from kb_pipeline.processors.analyze import AnalyzeProcessor + from kb_pipeline.schemas import AnalyzeInput + + output_file = str(tmp_path / "bilibili_BV1xx_analysis.md") + + input_data = AnalyzeInput( + content_id="BV1xx", + platform="bilibili", + metadata_file=sample_metadata, + content_file=sample_content, + content_type="video", + output_file=output_file + ) + + with patch("kb_pipeline.processors.base.get_settings") as mock_settings: + mock_settings.return_value = MagicMock() + with patch("kb_pipeline.processors.analyze.DeepAnalyzer") as MockDeepAnalyzer: + mock_analyzer_instance = MagicMock() + mock_analyzer_instance.analyze = MagicMock(return_value=mock_deep_analyzer) + MockDeepAnalyzer.return_value = mock_analyzer_instance + + processor = AnalyzeProcessor(state_manager=mock_state_manager) + result = await processor.process(input_data) + + assert result.success is True + assert result.analysis_file == output_file + + # Verify Markdown file was created + output_path = Path(output_file) + assert output_path.exists() + + # Verify content includes frontmatter and analysis + content = output_path.read_text(encoding="utf-8") + assert content.startswith("---") # YAML frontmatter + assert "##" in content # Markdown headers + + @pytest.mark.asyncio + async def test_analyze_records_state_on_success( + self, mock_deep_analyzer, mock_state_manager, sample_metadata, sample_content, tmp_path + ): + """Test 4: AnalyzeProcessor records state on success""" + from kb_pipeline.processors.analyze import AnalyzeProcessor + from kb_pipeline.schemas import AnalyzeInput, WorkflowState + + # Create a mock state that will be updated + mock_state = MagicMock(spec=WorkflowState) + mock_state.is_step_completed = MagicMock(return_value=False) + mock_state_manager.load.return_value = mock_state + + output_file = str(tmp_path / "bilibili_BV1xx_analysis.md") + + input_data = AnalyzeInput( + content_id="BV1xx", + platform="bilibili", + metadata_file=sample_metadata, + content_file=sample_content, + content_type="video", + output_file=output_file + ) + + with patch("kb_pipeline.processors.base.get_settings") as mock_settings: + mock_settings.return_value = MagicMock() + with patch("kb_pipeline.processors.analyze.DeepAnalyzer") as MockDeepAnalyzer: + mock_analyzer_instance = MagicMock() + mock_analyzer_instance.analyze = MagicMock(return_value=mock_deep_analyzer) + MockDeepAnalyzer.return_value = mock_analyzer_instance + + processor = AnalyzeProcessor(state_manager=mock_state_manager) + result = await processor.process(input_data) + + assert result.success is True + + # Verify state was recorded + mock_state.mark_step_completed.assert_called_once() + call_args = mock_state.mark_step_completed.call_args + assert call_args[0][0] == "analyze" # step_name + mock_state_manager.save.assert_called_once_with(mock_state) + + @pytest.mark.asyncio + async def test_analyze_records_state_on_failure( + self, mock_state_manager, sample_metadata, sample_content, tmp_path + ): + """Test 5: AnalyzeProcessor records state on failure""" + from kb_pipeline.processors.analyze import AnalyzeProcessor + from kb_pipeline.schemas import AnalyzeInput, WorkflowState + + # Create a mock state that will be updated + mock_state = MagicMock(spec=WorkflowState) + mock_state.is_step_completed = MagicMock(return_value=False) + mock_state_manager.load.return_value = mock_state + + output_file = str(tmp_path / "bilibili_BV1xx_analysis.md") + + input_data = AnalyzeInput( + content_id="BV1xx", + platform="bilibili", + metadata_file=sample_metadata, + content_file=sample_content, + content_type="video", + output_file=output_file + ) + + with patch("kb_pipeline.processors.base.get_settings") as mock_settings: + mock_settings.return_value = MagicMock() + with patch("kb_pipeline.processors.analyze.DeepAnalyzer") as MockDeepAnalyzer: + mock_analyzer_instance = MagicMock() + mock_analyzer_instance.analyze = MagicMock( + side_effect=ValueError("AI API error") + ) + MockDeepAnalyzer.return_value = mock_analyzer_instance + + processor = AnalyzeProcessor(state_manager=mock_state_manager) + result = await processor.process(input_data) + + assert result.success is False + assert "AI API error" in result.error_message + + # Verify failure was recorded + mock_state.mark_step_failed.assert_called_once() + call_args = mock_state.mark_step_failed.call_args + assert call_args[0][0] == "analyze" # step_name + mock_state_manager.save.assert_called_once_with(mock_state) + + +class TestAnalyzeProcessorModuleExport: + """Tests for AnalyzeProcessor module export.""" + + def test_analyze_processor_importable_from_processors(self): + """Test: AnalyzeProcessor can be imported from kb_pipeline.processors""" + from kb_pipeline.processors import AnalyzeProcessor + + assert AnalyzeProcessor is not None + assert hasattr(AnalyzeProcessor, 'STEP_NAME') + assert AnalyzeProcessor.STEP_NAME == "analyze"