"""语义分段鲁棒性测试 测试 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