kb-skills/tests/test_semantic_robustness.py

"""语义分段鲁棒性测试

测试 SemanticCache 缓存管理器的功能：
- 内容哈希计算
- 缓存读写
- 版本控制
- 错误处理

测试解析失败的重试和降级逻辑：
- JSONDecodeError 重试
- ValidationError 直接降级
- 日志记录
"""

import json
import logging
from pathlib import Path
from unittest.mock import Mock, MagicMock, patch
import pytest
import tempfile
import shutil

from kb_workflow.srt.cache.semantic_cache import (
    SemanticCache,
    compute_content_hash
)
from kb_workflow.srt.semantic_engine import (
    SegmentationResult,
    VideoSegment,
    UnifiedSegmentationStrategy,
    extract_json_from_llm_response,
)


class TestComputeContentHash:
    """测试内容哈希计算"""

    def test_same_content_same_hash(self):
        """相同内容应该产生相同的哈希值"""
        content = "这是测试内容"
        hash1 = compute_content_hash(content)
        hash2 = compute_content_hash(content)

        assert hash1 == hash2
        assert len(hash1) == 64  # SHA256 产生 64 字符的十六进制字符串
        assert isinstance(hash1, str)

    def test_different_content_different_hash(self):
        """不同内容应该产生不同的哈希值"""
        content1 = "这是测试内容1"
        content2 = "这是测试内容2"

        hash1 = compute_content_hash(content1)
        hash2 = compute_content_hash(content2)

        assert hash1 != hash2

    def test_empty_content_hash(self):
        """空内容也应该产生哈希值"""
        content = ""
        hash_value = compute_content_hash(content)

        assert len(hash_value) == 64
        assert isinstance(hash_value, str)


class TestSemanticCache:
    """测试 SemanticCache 缓存管理器"""

    def setup_method(self):
        """每个测试前创建临时缓存目录"""
        self.temp_dir = tempfile.mkdtemp()
        self.cache_dir = Path(self.temp_dir) / "cache"
        self.cache = SemanticCache(cache_dir=self.cache_dir)

    def teardown_method(self):
        """每个测试后清理临时目录"""
        if Path(self.temp_dir).exists():
            shutil.rmtree(self.temp_dir)

    def test_cache_miss_returns_none(self):
        """缓存未命中时返回 None"""
        content = "不存在的缓存内容"
        result = self.cache.get(content)

        assert result is None

    def test_cache_hit_after_set(self):
        """设置缓存后应该能读取到"""
        content = "测试内容"

        # 创建测试数据
        segments = [
            VideoSegment(
                title="测试标题",
                content="分段内容",
                start_index=0,
                end_index=10
            )
        ]
        result_data = SegmentationResult(
            video_type="知识类",
            segments=segments,
            actual_segment_count=1
        )

        # 设置缓存
        self.cache.set(content, result_data)

        # 读取缓存
        cached_result = self.cache.get(content)

        assert cached_result is not None
        assert cached_result.video_type == "知识类"
        assert len(cached_result.segments) == 1
        assert cached_result.segments[0].title == "测试标题"
        assert cached_result.segments[0].content == "分段内容"
        assert cached_result.actual_segment_count == 1

    def test_cache_file_creation(self):
        """缓存文件应该正确创建"""
        content = "测试内容"

        segments = [
            VideoSegment(
                title="标题",
                content="内容",
                start_index=0,
                end_index=5
            )
        ]
        result_data = SegmentationResult(
            video_type="知识类",
            segments=segments,
            actual_segment_count=1
        )

        self.cache.set(content, result_data)

        # 验证缓存文件存在
        cache_key = compute_content_hash(content)
        cache_file = self.cache_dir / f"{cache_key}.json"

        assert cache_file.exists()
        assert cache_file.is_file()

        # 验证文件内容
        import json
        data = json.loads(cache_file.read_text(encoding='utf-8'))
        assert data['version'] == self.cache.CACHE_VERSION
        assert data['video_type'] == "知识类"
        assert 'segments' in data

    def test_cache_auto_creates_directory(self):
        """缓存目录不存在时应该自动创建"""
        # 使用不存在的子目录
        non_existent_dir = self.cache_dir / "deep" / "nested"
        cache = SemanticCache(cache_dir=non_existent_dir)

        content = "测试内容"
        segments = [
            VideoSegment(
                title="标题",
                content="内容",
                start_index=0,
                end_index=5
            )
        ]
        result_data = SegmentationResult(
            video_type="知识类",
            segments=segments,
            actual_segment_count=1
        )

        # 设置缓存应该自动创建目录
        cache.set(content, result_data)

        assert non_existent_dir.exists()

    def test_cache_version_mismatch_returns_none(self):
        """缓存版本不匹配时应该返回 None"""
        content = "测试内容"

        segments = [
            VideoSegment(
                title="标题",
                content="内容",
                start_index=0,
                end_index=5
            )
        ]
        result_data = SegmentationResult(
            video_type="知识类",
            segments=segments,
            actual_segment_count=1
        )

        # 设置缓存
        self.cache.set(content, result_data)

        # 手动修改缓存文件的版本
        cache_key = compute_content_hash(content)
        cache_file = self.cache_dir / f"{cache_key}.json"

        import json
        data = json.loads(cache_file.read_text(encoding='utf-8'))
        data['version'] = "0.0"  # 错误的版本
        cache_file.write_text(json.dumps(data, ensure_ascii=False), encoding='utf-8')

        # 读取缓存应该返回 None
        cached_result = self.cache.get(content)
        assert cached_result is None

    def test_cache_corrupted_file_returns_none(self):
        """缓存文件损坏时应该返回 None"""
        content = "测试内容"

        # 创建损坏的缓存文件
        cache_key = compute_content_hash(content)
        cache_file = self.cache_dir / f"{cache_key}.json"

        self.cache_dir.mkdir(parents=True, exist_ok=True)
        cache_file.write_text("invalid json content", encoding='utf-8')

        # 读取缓存应该返回 None
        cached_result = self.cache.get(content)
        assert cached_result is None

    def test_multiple_cached_items(self):
        """多个缓存项应该独立存储"""
        content1 = "内容1"
        content2 = "内容2"

        segments1 = [
            VideoSegment(
                title="标题1",
                content="内容1",
                start_index=0,
                end_index=5
            )
        ]
        result1 = SegmentationResult(
            video_type="知识类",
            segments=segments1,
            actual_segment_count=1
        )

        segments2 = [
            VideoSegment(
                title="标题2",
                content="内容2",
                start_index=0,
                end_index=5
            )
        ]
        result2 = SegmentationResult(
            video_type="故事类",
            segments=segments2,
            actual_segment_count=1
        )

        # 设置两个缓存
        self.cache.set(content1, result1)
        self.cache.set(content2, result2)

        # 验证两个缓存都能正确读取
        cached1 = self.cache.get(content1)
        cached2 = self.cache.get(content2)

        assert cached1 is not None
        assert cached1.video_type == "知识类"
        assert cached1.segments[0].title == "标题1"

        assert cached2 is not None
        assert cached2.video_type == "故事类"
        assert cached2.segments[0].title == "标题2"

    def test_cache_with_complex_segments(self):
        """测试包含多个分段的复杂缓存"""
        content = "复杂的视频内容"

        segments = [
            VideoSegment(
                title="第一部分",
                content="第一段内容",
                start_index=0,
                end_index=10
            ),
            VideoSegment(
                title="第二部分",
                content="第二段内容",
                start_index=11,
                end_index=20
            ),
            VideoSegment(
                title=None,  # 无标题分段
                content="第三段内容",
                start_index=21,
                end_index=30
            )
        ]
        result_data = SegmentationResult(
            video_type="知识类",
            segments=segments,
            actual_segment_count=3
        )

        # 设置缓存
        self.cache.set(content, result_data)

        # 读取并验证
        cached_result = self.cache.get(content)

        assert cached_result is not None
        assert cached_result.video_type == "知识类"
        assert len(cached_result.segments) == 3
        assert cached_result.segments[0].title == "第一部分"
        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)