diff --git a/tests/test_semantic_robustness.py b/tests/test_semantic_robustness.py new file mode 100644 index 0000000..3d78cd9 --- /dev/null +++ b/tests/test_semantic_robustness.py @@ -0,0 +1,307 @@ +"""语义分段鲁棒性测试 + +测试 SemanticCache 缓存管理器的功能： +- 内容哈希计算 +- 缓存读写 +- 版本控制 +- 错误处理 +""" + +from pathlib import Path +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 + + +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