test(15-03): add AnalyzeProcessor independence tests

- 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 <noreply@anthropic.com>
This commit is contained in:
2026-03-23 23:03:12 +08:00
parent 9e33df7154
commit 7371c6e8c0
+96 -12
View File
@@ -6,8 +6,9 @@ Tests to ensure that configuration changes don't break existing functionality:
- Existing API calls remain compatible - Existing API calls remain compatible
""" """
import json
import pytest import pytest
from unittest.mock import Mock, patch, MagicMock from unittest.mock import Mock, patch, MagicMock, AsyncMock
from pathlib import Path from pathlib import Path
from kb_workflow.srt.parser import Segment from kb_workflow.srt.parser import Segment
from kb_workflow.srt.segmenter import SemanticSegmenter from kb_workflow.srt.segmenter import SemanticSegmenter
@@ -164,22 +165,105 @@ class TestBackwardCompatibility:
# ============================================================================= # =============================================================================
class TestAnalyzeProcessorIndependence: 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): def test_analyze_processor_unaffected_by_config(self):
"""Test: AnalyzeProcessor 不受 KBSettings.enable_semantic_segmentation 影响""" """Test: AnalyzeProcessor 不受 KBSettings.enable_semantic_segmentation 影响"""
# This test will be implemented in Task 2 from kb_workflow.pipeline.processors.analyze import AnalyzeProcessor
pass
# 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): def test_analyze_processor_normal_initialization(self):
"""Test: AnalyzeProcessor 正常初始化和处理""" """Test: AnalyzeProcessor 正常初始化和处理"""
# This test will be implemented in Task 2 from kb_workflow.pipeline.processors.analyze import AnalyzeProcessor
pass from kb_workflow.pipeline.schemas import AnalyzeInput
@pytest.mark.skipif(True, reason="Not implemented yet - will implement in Task 2") # Create processor
def test_analyze_processor_output_unchanged(self): 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 输出格式未改变""" """Test: AnalyzeProcessor 输出格式未改变"""
# This test will be implemented in Task 2 from kb_workflow.pipeline.processors.analyze import AnalyzeProcessor
pass 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