test(14-02): add failing test for parse error retry and fallback

- Test JSONDecodeError triggers retry 1 time before fallback - Test ValidationError directly fallbacks without retry - Test parse error logs LLM output fragment (first 500 chars) - Test fallback logs error type and message - Test retry uses INFO level logging Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-23 22:36:05 +08:00
parent baf7c0f521
commit 970c13a93b
1 changed files with 384 additions and 1 deletions
@@ -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)