"""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"