From 7371c6e8c075d1f10688fb89306e64e9356b9b5b Mon Sep 17 00:00:00 2001 From: starmen1265 Date: Mon, 23 Mar 2026 23:03:12 +0800 Subject: [PATCH] test(15-03): add AnalyzeProcessor independence tests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - test_analyze_processor_unaffected_by_config: 验证 AnalyzeProcessor 不受 enable_semantic_segmentation 配置影响 - test_analyze_processor_normal_initialization: 验证 AnalyzeProcessor 正常初始化 - test_analyze_processor_output_unchanged: 验证 AnalyzeProcessor 输出格式未改变 Task 2 完成：AnalyzeProcessor 独立性验证（CONFIG-03） 现有 test_analyze.py 测试全部通过，未被配置变更破坏 Co-Authored-By: Claude Opus 4.6 --- tests/test_backward_compat.py | 108 ++++++++++++++++++++++++++++++---- 1 file changed, 96 insertions(+), 12 deletions(-) diff --git a/tests/test_backward_compat.py b/tests/test_backward_compat.py index de5e546..a0af0b8 100644 --- a/tests/test_backward_compat.py +++ b/tests/test_backward_compat.py @@ -6,8 +6,9 @@ Tests to ensure that configuration changes don't break existing functionality: - Existing API calls remain compatible """ +import json import pytest -from unittest.mock import Mock, patch, MagicMock +from unittest.mock import Mock, patch, MagicMock, AsyncMock from pathlib import Path from kb_workflow.srt.parser import Segment from kb_workflow.srt.segmenter import SemanticSegmenter @@ -164,22 +165,105 @@ class TestBackwardCompatibility: # ============================================================================= class TestAnalyzeProcessorIndependence: - """Test that AnalyzeProcessor is independent of SRTProcessor configuration""" + """Test that AnalyzeProcessor is independent of SRTProcessor configuration (CONFIG-03)""" - @pytest.mark.skipif(True, reason="Not implemented yet - will implement in Task 2") def test_analyze_processor_unaffected_by_config(self): """Test: AnalyzeProcessor 不受 KBSettings.enable_semantic_segmentation 影响""" - # This test will be implemented in Task 2 - pass + from kb_workflow.pipeline.processors.analyze import AnalyzeProcessor + + # Verify AnalyzeProcessor can be initialized normally + processor = AnalyzeProcessor(state_manager=None) + + # Verify it's a valid processor + assert processor is not None + assert processor.STEP_NAME == "analyze" + + # Verify AnalyzeProcessor does NOT have enable_semantic_segmentation attribute + # (this is SRTProcessor-specific config) + assert not hasattr(processor, 'enable_semantic_segmentation'), \ + "AnalyzeProcessor should NOT have enable_semantic_segmentation attribute" - @pytest.mark.skipif(True, reason="Not implemented yet - will implement in Task 2") def test_analyze_processor_normal_initialization(self): """Test: AnalyzeProcessor 正常初始化和处理""" - # This test will be implemented in Task 2 - pass + from kb_workflow.pipeline.processors.analyze import AnalyzeProcessor + from kb_workflow.pipeline.schemas import AnalyzeInput - @pytest.mark.skipif(True, reason="Not implemented yet - will implement in Task 2") - def test_analyze_processor_output_unchanged(self): + # Create processor + processor = AnalyzeProcessor(state_manager=None) + + # Verify it's properly initialized + assert processor is not None + assert hasattr(processor, 'process') + assert hasattr(processor, '_create_ai_client') + + # Verify it can create AI client (needed for analysis) + ai_client = processor._create_ai_client() + assert ai_client is not None + # AI client may or may not have api_key depending on env setup + + @pytest.mark.asyncio + async def test_analyze_processor_output_unchanged(self, tmp_path): """Test: AnalyzeProcessor 输出格式未改变""" - # This test will be implemented in Task 2 - pass + from kb_workflow.pipeline.processors.analyze import AnalyzeProcessor + from kb_workflow.pipeline.schemas import AnalyzeInput + + # Create temporary files + metadata_file = tmp_path / "metadata.json" + content_file = tmp_path / "content.json" + output_file = tmp_path / "output.md" + + # Create test metadata + metadata = { + "title": "Test Title", + "author": "Test Author", + "platform": "zhihu" # Use valid platform value + } + metadata_file.write_text(json.dumps(metadata, ensure_ascii=False)) + + # Create test content + content = { + "text": "This is test content for analysis." + } + content_file.write_text(json.dumps(content, ensure_ascii=False)) + + # Create input + input_data = AnalyzeInput( + content_id="test123", + platform="zhihu", # Use valid platform value + metadata_file=str(metadata_file), + content_file=str(content_file), + content_type="article", + output_file=str(output_file) + ) + + # Create a mock result object + mock_result = Mock() + mock_result.summary = "Test summary" + mock_result.main_topics = ["Topic 1"] + mock_result.key_insights = ["Insight 1"] + mock_result.recommended_tags = ["tag1"] + mock_result.full_analysis = "## Analysis\n\nTest analysis content" + mock_result.original_text = "This is test content for analysis." + + # Mock the _run_analysis method directly + with patch.object(AnalyzeProcessor, '_run_analysis', new_callable=AsyncMock) as mock_run: + mock_run.return_value = mock_result + + # Create processor and process + processor = AnalyzeProcessor(state_manager=None) + result = await processor.process(input_data) + + # Verify output structure is unchanged + assert result.success is True + assert result.analysis_file == str(output_file) + assert isinstance(result.main_topics, list) + assert isinstance(result.key_insights, list) + assert isinstance(result.tags, list) + + # Verify output file was created with expected format + assert output_file.exists() + content = output_file.read_text(encoding='utf-8') + # Should contain YAML frontmatter + assert content.startswith('---') + # Should contain analysis content + assert '## Analysis' in content or 'Test analysis content' in content