diff --git a/tests/test_semantic_robustness.py b/tests/test_semantic_robustness.py index 3d78cd9..b9a25ba 100644 --- a/tests/test_semantic_robustness.py +++ b/tests/test_semantic_robustness.py @@ -5,9 +5,17 @@ - 缓存读写 - 版本控制 - 错误处理 + +测试解析失败的重试和降级逻辑: +- JSONDecodeError 重试 +- ValidationError 直接降级 +- 日志记录 """ +import json +import logging from pathlib import Path +from unittest.mock import Mock, MagicMock, patch import pytest import tempfile import shutil @@ -16,7 +24,12 @@ from kb_workflow.srt.cache.semantic_cache import ( SemanticCache, compute_content_hash ) -from kb_workflow.srt.semantic_engine import SegmentationResult, VideoSegment +from kb_workflow.srt.semantic_engine import ( + SegmentationResult, + VideoSegment, + UnifiedSegmentationStrategy, + extract_json_from_llm_response, +) class TestComputeContentHash: @@ -305,3 +318,373 @@ class TestSemanticCache: assert cached_result.segments[1].title == "第二部分" assert cached_result.segments[2].title is None assert cached_result.actual_segment_count == 3 + + +# ============================================================================ +# Test: 解析失败重试 (Task 14-02-01) +# ============================================================================ + +class TestParseFailureRetry: + """测试解析失败时的重试逻辑""" + + @pytest.fixture + def sample_segments(self): + """样本字幕分段""" + segments = [] + for i in range(10): + seg = Mock() + seg.text = f"这是第{i+1}段字幕内容。" + segments.append(seg) + return segments + + @pytest.fixture + def sample_metadata(self): + """样本视频元数据""" + return { + "title": "测试视频", + "duration": 600, + "category": "知识类" + } + + @pytest.fixture + def mock_ai_client(self): + """模拟 AI 客户端""" + client = Mock() + client.api_key = "test-key" + return client + + @pytest.fixture + def valid_llm_response(self): + """有效的 LLM 响应""" + return json.dumps({ + "video_type": "知识类", + "segments": [ + { + "title": "机器学习概述", + "content": "机器学习是人工智能的一个分支。", + "start_index": 0, + "end_index": 20 + }, + { + "title": None, + "content": "深度学习是机器学习的一个子集。", + "start_index": 21, + "end_index": 40 + } + ], + "actual_segment_count": 2 + }, ensure_ascii=False) + + @pytest.fixture + def invalid_json_response(self): + """无效的 JSON 响应""" + return "这不是一个有效的 JSON 格式。" + + @pytest.fixture + def pydantic_invalid_response(self): + """Pydantic 验证失败的响应(缺少必需字段)""" + return json.dumps({ + "video_type": "知识类", + "segments": [ + { + "title": "测试", + # 缺少 content, start_index, end_index + } + ], + "actual_segment_count": 1 + }, ensure_ascii=False) + + def test_json_decode_error_triggers_retry( + self, + mock_ai_client, + sample_segments, + sample_metadata, + invalid_json_response, + caplog + ): + """测试 JSONDecodeError 触发重试 1 次,仍失败则降级""" + # 第一次返回无效 JSON,第二次也返回无效 JSON + mock_ai_client.generate.side_effect = [ + invalid_json_response, + invalid_json_response + ] + + strategy = UnifiedSegmentationStrategy( + ai_client=mock_ai_client, + fallback_segmenter=None + ) + # 手动设置 fallback_segmenter + from kb_workflow.srt.segmenter import SemanticSegmenter + strategy.fallback_segmenter = SemanticSegmenter() + + with caplog.at_level(logging.WARNING): + result = strategy.segment(sample_segments, sample_metadata) + + # 应该降级到规则分段 + assert isinstance(result, list) + assert len(result) > 0 + assert all(isinstance(seg, VideoSegment) for seg in result) + + # 验证调用了两次 LLM(初始调用 + 1 次重试) + assert mock_ai_client.generate.call_count == 2 + + # 验证日志包含错误信息 + assert any("JSON 解析失败" in record.message for record in caplog.records) + assert any("降级到规则分段" in record.message for record in caplog.records) + + def test_json_decode_error_success_on_retry( + self, + mock_ai_client, + sample_segments, + sample_metadata, + invalid_json_response, + valid_llm_response, + caplog + ): + """测试 JSONDecodeError 重试后成功""" + # 第一次返回无效 JSON,第二次返回有效响应 + mock_ai_client.generate.side_effect = [ + invalid_json_response, + valid_llm_response + ] + + strategy = UnifiedSegmentationStrategy( + ai_client=mock_ai_client, + fallback_segmenter=None + ) + + with caplog.at_level(logging.INFO): + result = strategy.segment(sample_segments, sample_metadata) + + # 应该返回 LLM 分段结果 + assert isinstance(result, list) + assert len(result) == 2 + assert result[0].title == "机器学习概述" + assert result[1].title is None + + # 验证调用了两次 LLM(初始调用 + 1 次重试) + assert mock_ai_client.generate.call_count == 2 + + # 验证日志包含重试信息 + assert any("重试" in record.message or "retry" in record.message.lower() + for record in caplog.records) + + def test_validation_error_no_retry( + self, + mock_ai_client, + sample_segments, + sample_metadata, + pydantic_invalid_response, + caplog + ): + """测试 ValidationError(Pydantic 验证失败)直接降级,不重试""" + # 返回 Pydantic 验证失败的响应 + mock_ai_client.generate.return_value = pydantic_invalid_response + + strategy = UnifiedSegmentationStrategy( + ai_client=mock_ai_client, + fallback_segmenter=None + ) + from kb_workflow.srt.segmenter import SemanticSegmenter + strategy.fallback_segmenter = SemanticSegmenter() + + with caplog.at_level(logging.WARNING): + result = strategy.segment(sample_segments, sample_metadata) + + # 应该降级到规则分段 + assert isinstance(result, list) + assert len(result) > 0 + + # 验证只调用了 1 次 LLM(不重试) + assert mock_ai_client.generate.call_count == 1 + + # 验证日志包含验证失败信息 + assert any("Pydantic 验证失败" in record.message or "验证失败" in record.message + for record in caplog.records) + + def test_parse_error_logs_llm_output( + self, + mock_ai_client, + sample_segments, + sample_metadata, + caplog + ): + """测试解析失败时记录 LLM 输出片段(前 500 字符)""" + # 创建一个超过 500 字符的无效响应 + long_invalid_response = "这是一个很长的无效响应。" * 100 + + mock_ai_client.generate.side_effect = [ + long_invalid_response, + long_invalid_response + ] + + strategy = UnifiedSegmentationStrategy( + ai_client=mock_ai_client, + fallback_segmenter=None + ) + from kb_workflow.srt.segmenter import SemanticSegmenter + strategy.fallback_segmenter = SemanticSegmenter() + + with caplog.at_level(logging.ERROR): + result = strategy.segment(sample_segments, sample_metadata) + + # 验证日志包含 LLM 输出片段 + error_logs = [record for record in caplog.records if record.levelno == logging.ERROR] + assert len(error_logs) > 0 + + # 至少有一个错误日志包含 LLM 输出片段 + assert any( + "LLM 输出片段" in record.message or "原始 LLM 输出" in record.message + for record in error_logs + ) + + +# ============================================================================ +# Test: 降级日志 (Task 14-02-02) +# ============================================================================ + +class TestFallbackLogging: + """测试降级时的日志记录""" + + @pytest.fixture + def sample_segments(self): + """样本字幕分段""" + segments = [] + for i in range(10): + seg = Mock() + seg.text = f"这是第{i+1}段字幕内容。" + segments.append(seg) + return segments + + @pytest.fixture + def sample_metadata(self): + """样本视频元数据""" + return { + "title": "测试视频", + "duration": 600, + "category": "知识类" + } + + @pytest.fixture + def mock_ai_client(self): + """模拟 AI 客户端""" + client = Mock() + client.api_key = "test-key" + return client + + @pytest.fixture + def invalid_json_response(self): + """无效的 JSON 响应""" + return "这不是一个有效的 JSON 格式。" + + @pytest.fixture + def valid_llm_response(self): + """有效的 LLM 响应""" + return json.dumps({ + "video_type": "知识类", + "segments": [ + { + "title": "机器学习概述", + "content": "机器学习是人工智能的一个分支。", + "start_index": 0, + "end_index": 20 + } + ], + "actual_segment_count": 1 + }, ensure_ascii=False) + + def test_fallback_logs_error_type_and_message( + self, + mock_ai_client, + sample_segments, + sample_metadata, + invalid_json_response, + caplog + ): + """测试降级时记录错误类型和消息""" + mock_ai_client.generate.side_effect = [ + invalid_json_response, + invalid_json_response + ] + + strategy = UnifiedSegmentationStrategy( + ai_client=mock_ai_client, + fallback_segmenter=None + ) + from kb_workflow.srt.segmenter import SemanticSegmenter + strategy.fallback_segmenter = SemanticSegmenter() + + with caplog.at_level(logging.WARNING): + result = strategy.segment(sample_segments, sample_metadata) + + # 验证有 WARNING 级别的日志 + warning_logs = [record for record in caplog.records if record.levelno == logging.WARNING] + assert len(warning_logs) > 0 + + # 验证日志包含具体的错误信息 + assert any( + "降级到规则分段" in record.message + for record in warning_logs + ) + + def test_retry_logs_info_level( + self, + mock_ai_client, + sample_segments, + sample_metadata, + invalid_json_response, + valid_llm_response, + caplog + ): + """测试重试时使用 INFO 级别日志""" + mock_ai_client.generate.side_effect = [ + invalid_json_response, + valid_llm_response + ] + + strategy = UnifiedSegmentationStrategy( + ai_client=mock_ai_client + ) + + with caplog.at_level(logging.INFO): + result = strategy.segment(sample_segments, sample_metadata) + + # 验证有 INFO 级别的重试日志 + info_logs = [record for record in caplog.records if record.levelno == logging.INFO] + assert len(info_logs) > 0 + + +# ============================================================================ +# Test: extract_json_from_llm_response +# ============================================================================ + +class TestExtractJSON: + """测试 JSON 提取函数""" + + def test_extract_pure_json(self): + """测试提取纯 JSON""" + response = '{"key": "value"}' + result = extract_json_from_llm_response(response) + assert result == {"key": "value"} + + def test_extract_json_from_markdown_block(self): + """测试从 markdown 代码块中提取 JSON""" + response = '''```json +{ + "key": "value" +} +```''' + result = extract_json_from_llm_response(response) + assert result == {"key": "value"} + + def test_extract_json_from_text(self): + """测试从包含额外文本的响应中提取 JSON""" + response = '这是一些说明文本。{"key": "value"}还有更多文本。' + result = extract_json_from_llm_response(response) + assert result == {"key": "value"} + + def test_extract_json_raises_value_error_on_failure(self): + """测试无法提取 JSON 时抛出 ValueError""" + response = '这不包含任何 JSON 内容。' + with pytest.raises(ValueError, match="未找到 JSON 内容"): + extract_json_from_llm_response(response)