"""测试语义分段的鲁棒性功能 (Phase 14) 测试 LLM 调用失败时的重试和降级策略。 Plan 14-01: 实现可恢复错误检测和重试逻辑 """ import json import pytest from unittest.mock import Mock from requests.exceptions import ConnectionError, Timeout, HTTPError from kb_workflow.srt.parser import Segment from kb_workflow.srt.semantic_engine import ( UnifiedSegmentationStrategy, VideoSegment ) class TestRecoverableErrorDetection: """测试可恢复错误检测功能""" def test_is_recoverable_error_connection_error(self): """Test: ConnectionError 被识别为可恢复错误""" strategy = UnifiedSegmentationStrategy() # 创建模拟客户端，抛出 ConnectionError mock_client = Mock() mock_client.api_key = "test_key" mock_client.generate.side_effect = ConnectionError("Connection refused") strategy = UnifiedSegmentationStrategy(ai_client=mock_client) # 准备测试数据 segments = [ Segment(1, 0.0, 1.0, "第一句"), Segment(2, 1.0, 2.0, "第二句") ] # 应该降级到规则分段（ConnectionError 可恢复，重试 1 次后降级） result = strategy.segment(segments, {"title": "测试"}) # 验证降级结果 assert len(result) > 0 # 验证调用了 2 次（初始 + 1 次重试） assert mock_client.generate.call_count == 2 def test_is_recoverable_error_timeout(self): """Test: Timeout 被识别为可恢复错误""" mock_client = Mock() mock_client.api_key = "test_key" mock_client.generate.side_effect = Timeout("Request timeout") strategy = UnifiedSegmentationStrategy(ai_client=mock_client) segments = [Segment(1, 0.0, 1.0, "测试")] result = strategy.segment(segments, {"title": "测试"}) # 应该重试 1 次 assert mock_client.generate.call_count == 2 assert len(result) > 0 def test_is_recoverable_error_rate_limit_429(self): """Test: HTTP 429 (限流) 被识别为可恢复错误""" mock_client = Mock() mock_client.api_key = "test_key" # 创建 429 响应 response = Mock() response.status_code = 429 response.text = "Rate limit exceeded" http_error = HTTPError(response=response) mock_client.generate.side_effect = http_error strategy = UnifiedSegmentationStrategy(ai_client=mock_client) segments = [Segment(1, 0.0, 1.0, "测试")] result = strategy.segment(segments, {"title": "测试"}) # 应该重试 1 次 assert mock_client.generate.call_count == 2 assert len(result) > 0 def test_is_recoverable_error_server_error_500(self): """Test: HTTP 500 (服务器错误) 被识别为可恢复错误""" mock_client = Mock() mock_client.api_key = "test_key" response = Mock() response.status_code = 500 response.text = "Internal server error" http_error = HTTPError(response=response) mock_client.generate.side_effect = http_error strategy = UnifiedSegmentationStrategy(ai_client=mock_client) segments = [Segment(1, 0.0, 1.0, "测试")] result = strategy.segment(segments, {"title": "测试"}) # 应该重试 1 次 assert mock_client.generate.call_count == 2 assert len(result) > 0 def test_is_recoverable_error_server_error_502(self): """Test: HTTP 502 (网关错误) 被识别为可恢复错误""" mock_client = Mock() mock_client.api_key = "test_key" response = Mock() response.status_code = 502 response.text = "Bad gateway" http_error = HTTPError(response=response) mock_client.generate.side_effect = http_error strategy = UnifiedSegmentationStrategy(ai_client=mock_client) segments = [Segment(1, 0.0, 1.0, "测试")] result = strategy.segment(segments, {"title": "测试"}) # 应该重试 1 次 assert mock_client.generate.call_count == 2 assert len(result) > 0 def test_is_recoverable_error_server_error_503(self): """Test: HTTP 503 (服务不可用) 被识别为可恢复错误""" mock_client = Mock() mock_client.api_key = "test_key" response = Mock() response.status_code = 503 response.text = "Service unavailable" http_error = HTTPError(response=response) mock_client.generate.side_effect = http_error strategy = UnifiedSegmentationStrategy(ai_client=mock_client) segments = [Segment(1, 0.0, 1.0, "测试")] result = strategy.segment(segments, {"title": "测试"}) # 应该重试 1 次 assert mock_client.generate.call_count == 2 assert len(result) > 0 def test_is_recoverable_error_server_error_504(self): """Test: HTTP 504 (网关超时) 被识别为可恢复错误""" mock_client = Mock() mock_client.api_key = "test_key" response = Mock() response.status_code = 504 response.text = "Gateway timeout" http_error = HTTPError(response=response) mock_client.generate.side_effect = http_error strategy = UnifiedSegmentationStrategy(ai_client=mock_client) segments = [Segment(1, 0.0, 1.0, "测试")] result = strategy.segment(segments, {"title": "测试"}) # 应该重试 1 次 assert mock_client.generate.call_count == 2 assert len(result) > 0 class TestNonRecoverableErrorDetection: """测试不可恢复错误检测功能""" def test_non_recoverable_error_400(self): """Test: HTTP 400 (错误请求) 被识别为不可恢复错误""" mock_client = Mock() mock_client.api_key = "test_key" response = Mock() response.status_code = 400 response.text = "Bad request" http_error = HTTPError(response=response) mock_client.generate.side_effect = http_error strategy = UnifiedSegmentationStrategy(ai_client=mock_client) segments = [Segment(1, 0.0, 1.0, "测试")] result = strategy.segment(segments, {"title": "测试"}) # 不应该重试，直接降级 assert mock_client.generate.call_count == 1 assert len(result) > 0 def test_non_recoverable_error_401(self): """Test: HTTP 401 (未授权) 被识别为不可恢复错误""" mock_client = Mock() mock_client.api_key = "test_key" response = Mock() response.status_code = 401 response.text = "Unauthorized" http_error = HTTPError(response=response) mock_client.generate.side_effect = http_error strategy = UnifiedSegmentationStrategy(ai_client=mock_client) segments = [Segment(1, 0.0, 1.0, "测试")] result = strategy.segment(segments, {"title": "测试"}) # 不应该重试 assert mock_client.generate.call_count == 1 assert len(result) > 0 def test_non_recoverable_error_403(self): """Test: HTTP 403 (禁止访问) 被识别为不可恢复错误""" mock_client = Mock() mock_client.api_key = "test_key" response = Mock() response.status_code = 403 response.text = "Forbidden" http_error = HTTPError(response=response) mock_client.generate.side_effect = http_error strategy = UnifiedSegmentationStrategy(ai_client=mock_client) segments = [Segment(1, 0.0, 1.0, "测试")] result = strategy.segment(segments, {"title": "测试"}) # 不应该重试 assert mock_client.generate.call_count == 1 assert len(result) > 0 def test_non_recoverable_error_404(self): """Test: HTTP 404 (未找到) 被识别为不可恢复错误""" mock_client = Mock() mock_client.api_key = "test_key" response = Mock() response.status_code = 404 response.text = "Not found" http_error = HTTPError(response=response) mock_client.generate.side_effect = http_error strategy = UnifiedSegmentationStrategy(ai_client=mock_client) segments = [Segment(1, 0.0, 1.0, "测试")] result = strategy.segment(segments, {"title": "测试"}) # 不应该重试 assert mock_client.generate.call_count == 1 assert len(result) > 0 class TestRetryMechanism: """测试重试机制""" def test_retry_once_on_recoverable_error(self): """Test: 可恢复错误重试 1 次""" mock_client = Mock() mock_client.api_key = "test_key" mock_client.generate.side_effect = ConnectionError("Network error") strategy = UnifiedSegmentationStrategy(ai_client=mock_client) segments = [Segment(1, 0.0, 1.0, "测试")] result = strategy.segment(segments, {"title": "测试"}) # 验证调用次数：初始 1 次 + 重试 1 次 = 2 次 assert mock_client.generate.call_count == 2 def test_no_retry_on_non_recoverable_error(self): """Test: 不可恢复错误不重试""" mock_client = Mock() mock_client.api_key = "test_key" response = Mock() response.status_code = 401 http_error = HTTPError(response=response) mock_client.generate.side_effect = http_error strategy = UnifiedSegmentationStrategy(ai_client=mock_client) segments = [Segment(1, 0.0, 1.0, "测试")] result = strategy.segment(segments, {"title": "测试"}) # 验证只调用 1 次（不重试） assert mock_client.generate.call_count == 1 def test_immediate_retry_no_delay(self): """Test: 立即重试，无延迟""" import time mock_client = Mock() mock_client.api_key = "test_key" mock_client.generate.side_effect = ConnectionError("Network error") strategy = UnifiedSegmentationStrategy(ai_client=mock_client) segments = [Segment(1, 0.0, 1.0, "测试")] start_time = time.time() result = strategy.segment(segments, {"title": "测试"}) elapsed_time = time.time() - start_time # 验证重试是立即的（无延迟） # 正常重试应该有延迟，立即重试应该很快（< 0.1 秒） assert elapsed_time < 0.1 class TestFallbackToRuleBased: """测试降级到规则分段""" def test_fallback_on_llm_failure(self): """Test: LLM 调用失败时降级到规则分段""" mock_client = Mock() mock_client.api_key = "test_key" mock_client.generate.side_effect = Exception("LLM error") strategy = UnifiedSegmentationStrategy(ai_client=mock_client) segments = [ Segment(1, 0.0, 1.0, "第一句"), Segment(2, 1.0, 2.0, "第二句") ] result = strategy.segment(segments, {"title": "测试"}) # 验证返回了规则分段结果 assert len(result) > 0 # 验证返回的是 VideoSegment 对象（规则分段转换后的格式） assert all(isinstance(seg, VideoSegment) for seg in result) # 验证规则分段没有标题（降级分段标题为 null） assert all(seg.title is None for seg in result) def test_fallback_when_ai_client_unavailable(self): """Test: AI 客户端不可用时直接使用规则分段""" # 创建没有 AI 客户端的策略 strategy = UnifiedSegmentationStrategy(ai_client=None) segments = [ Segment(1, 0.0, 1.0, "第一句"), Segment(2, 1.0, 2.0, "第二句") ] result = strategy.segment(segments, {"title": "测试"}) # 验证返回了规则分段结果 assert len(result) > 0 assert all(isinstance(seg, VideoSegment) for seg in result) assert all(seg.title is None for seg in result) class TestLoggingBehavior: """测试日志记录行为""" def test_retry_logging(self, caplog): """Test: 重试时记录 info 级别日志""" import logging mock_client = Mock() mock_client.api_key = "test_key" mock_client.generate.side_effect = ConnectionError("Network error") strategy = UnifiedSegmentationStrategy(ai_client=mock_client) segments = [Segment(1, 0.0, 1.0, "测试")] with caplog.at_level(logging.INFO): result = strategy.segment(segments, {"title": "测试"}) # 验证日志包含重试信息 assert any("重试" in record.message or "retry" in record.message.lower() for record in caplog.records if record.levelno == logging.INFO) def test_fallback_warning_logging(self, caplog): """Test: 降级时记录 warning 级别日志""" import logging mock_client = Mock() mock_client.api_key = "test_key" response = Mock() response.status_code = 401 http_error = HTTPError(response=response) mock_client.generate.side_effect = http_error strategy = UnifiedSegmentationStrategy(ai_client=mock_client) segments = [Segment(1, 0.0, 1.0, "测试")] with caplog.at_level(logging.WARNING): result = strategy.segment(segments, {"title": "测试"}) # 验证日志包含降级信息和错误类型 assert any("降级" in record.message or "fallback" in record.message.lower() for record in caplog.records if record.levelno == logging.WARNING) # 验证日志包含错误类型（401 或 Unauthorized） assert any("401" in record.message or "Unauthorized" in record.message for record in caplog.records if record.levelno == logging.WARNING) # ============================================================================ # Test: 解析失败重试 (Plan 14-02 - Task 14-02-01) # ============================================================================ class TestParseFailureRetry: """测试解析失败时的重试逻辑（Plan 14-02）""" def test_json_decode_error_triggers_retry(self, caplog): """测试 JSONDecodeError 触发重试 1 次，仍失败则降级""" import logging # 创建模拟 AI 客户端，始终返回无效 JSON mock_client = Mock() mock_client.api_key = "test-key" invalid_response = "这不是有效的 JSON" mock_client.generate.side_effect = [invalid_response, invalid_response] strategy = UnifiedSegmentationStrategy(ai_client=mock_client) segments = [Segment(1, 0.0, 1.0, "测试")] with caplog.at_level(logging.WARNING): result = strategy.segment(segments, {"title": "测试"}) # 应该降级到规则分段 assert len(result) > 0 assert all(isinstance(seg, VideoSegment) for seg in result) # 验证调用了两次 LLM（初始 + 1 次重试） assert mock_client.generate.call_count == 2 # 验证日志包含解析失败和降级信息 # extract_json_from_llm_response 抛出 ValueError，不是 JSONDecodeError assert any( ("ValueError" in record.message or "JSON" in record.message) and "解析失败" in record.message or "未找到 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, caplog): """测试 JSONDecodeError 重试后成功""" import logging import json mock_client = Mock() mock_client.api_key = "test-key" # 第一次失败，第二次成功 invalid_response = "无效 JSON" valid_response = json.dumps({ "video_type": "知识类", "segments": [ { "title": "测试标题", "content": "测试内容", "start_index": 0, "end_index": 10 } ], "actual_segment_count": 1 }, ensure_ascii=False) mock_client.generate.side_effect = [invalid_response, valid_response] strategy = UnifiedSegmentationStrategy(ai_client=mock_client) segments = [Segment(1, 0.0, 1.0, "测试")] with caplog.at_level(logging.INFO): result = strategy.segment(segments, {"title": "测试"}) # 应该返回 LLM 分段结果 assert len(result) == 1 assert result[0].title == "测试标题" # 验证调用了两次 LLM（初始 + 1 次重试） assert mock_client.generate.call_count == 2 # 验证日志包含重试信息（可能是"重试"或"retrying"） assert any( "重试" in record.message or "retrying" in record.message.lower() for record in caplog.records ) def test_validation_error_no_retry(self, caplog): """测试 ValidationError（Pydantic 验证失败）直接降级，不重试""" import logging import json mock_client = Mock() mock_client.api_key = "test-key" # 返回 Pydantic 验证失败的响应（缺少必需字段） invalid_response = json.dumps({ "video_type": "知识类", "segments": [ { "title": "测试", # 缺少 content, start_index, end_index } ], "actual_segment_count": 1 }, ensure_ascii=False) mock_client.generate.return_value = invalid_response strategy = UnifiedSegmentationStrategy(ai_client=mock_client) segments = [Segment(1, 0.0, 1.0, "测试")] with caplog.at_level(logging.WARNING): result = strategy.segment(segments, {"title": "测试"}) # 应该降级到规则分段 assert len(result) > 0 # 验证只调用了 1 次 LLM（不重试） assert mock_client.generate.call_count == 1 # 验证日志包含验证失败信息 assert any( "ValidationError" in record.message or "验证失败" in record.message for record in caplog.records ) def test_parse_error_logs_llm_output(self, caplog): """测试解析失败时记录 LLM 输出片段（前 500 字符）""" import logging mock_client = Mock() mock_client.api_key = "test-key" # 创建一个超过 500 字符的无效响应 long_invalid_response = "这是一个很长的无效响应。" * 100 mock_client.generate.side_effect = [long_invalid_response, long_invalid_response] strategy = UnifiedSegmentationStrategy(ai_client=mock_client) segments = [Segment(1, 0.0, 1.0, "测试")] with caplog.at_level(logging.WARNING): result = strategy.segment(segments, {"title": "测试"}) # 验证日志包含 LLM 输出片段（使用 WARNING 级别） warning_logs = [record for record in caplog.records if record.levelno == logging.WARNING] assert len(warning_logs) > 0 # 验证日志包含 LLM 输出片段信息 assert any( "LLM 输出片段" in record.message for record in warning_logs ) class TestFallbackLogging: """测试降级时的日志记录（Plan 14-02 - Task 14-02-02）""" def test_fallback_logs_error_type_and_message(self, caplog): """测试降级时记录错误类型和消息""" import logging mock_client = Mock() mock_client.api_key = "test-key" invalid_response = "无效 JSON" mock_client.generate.side_effect = [invalid_response, invalid_response] strategy = UnifiedSegmentationStrategy(ai_client=mock_client) segments = [Segment(1, 0.0, 1.0, "测试")] with caplog.at_level(logging.WARNING): result = strategy.segment(segments, {"title": "测试"}) # 验证有 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, caplog): """测试重试时使用 INFO 级别日志""" import logging import json mock_client = Mock() mock_client.api_key = "test-key" invalid_response = "无效 JSON" valid_response = json.dumps({ "video_type": "知识类", "segments": [ { "title": "测试", "content": "内容", "start_index": 0, "end_index": 10 } ], "actual_segment_count": 1 }, ensure_ascii=False) mock_client.generate.side_effect = [invalid_response, valid_response] strategy = UnifiedSegmentationStrategy(ai_client=mock_client) segments = [Segment(1, 0.0, 1.0, "测试")] with caplog.at_level(logging.INFO): result = strategy.segment(segments, {"title": "测试"}) # 验证有 INFO 级别的重试日志 info_logs = [record for record in caplog.records if record.levelno == logging.INFO] assert len(info_logs) > 0 # 验证日志包含重试信息（可能是"重试"或"retrying"） assert any( "重试" in record.message or "retrying" in record.message.lower() for record in info_logs ) class TestExtractJSON: """测试 JSON 提取函数（Plan 14-02）""" def test_extract_pure_json(self): """测试提取纯 JSON""" from kb_workflow.srt.semantic_engine import extract_json_from_llm_response response = '{"key": "value"}' result = extract_json_from_llm_response(response) assert result == {"key": "value"} def test_extract_json_from_markdown_block(self): """测试从 markdown 代码块中提取 JSON""" from kb_workflow.srt.semantic_engine import extract_json_from_llm_response response = '''```json { "key": "value" } ```''' result = extract_json_from_llm_response(response) assert result == {"key": "value"} def test_extract_json_from_text(self): """测试从包含额外文本的响应中提取 JSON""" from kb_workflow.srt.semantic_engine import extract_json_from_llm_response 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""" from kb_workflow.srt.semantic_engine import extract_json_from_llm_response import pytest response = '这不包含任何 JSON 内容。' with pytest.raises(ValueError, match="未找到 JSON 内容"): extract_json_from_llm_response(response) # ============================================================================ # Test: 长文本分块处理 (Plan 14-04) # ============================================================================ class TestSplitIntoChunks: """测试文本分块功能（Plan 14-04 - Task 14-04-01）""" def test_split_into_chunks_basic(self): """测试基本的文本分块功能""" from kb_workflow.srt.semantic_engine import UnifiedSegmentationStrategy strategy = UnifiedSegmentationStrategy() # 创建一个包含多个句子的文本（约 6000 字符） text = "这是第一句。" * 1000 + "这是第二句。" * 1000 + "这是第三句。" * 1000 chunks = strategy._split_into_chunks(text, max_chunk_size=5000) # 验证分块数量 assert len(chunks) > 1 # 验证每个分块不超过最大字符数 for chunk_text, offset in chunks: assert len(chunk_text) <= 5000 # 验证偏移量是递增的 offsets = [offset for _, offset in chunks] assert offsets == sorted(offsets) # 验证所有分块拼接后等于原文 reconstructed = ''.join(chunk_text for chunk_text, _ in chunks) assert reconstructed == text def test_split_into_chunks_short_text(self): """测试短文本不分块""" from kb_workflow.srt.semantic_engine import UnifiedSegmentationStrategy strategy = UnifiedSegmentationStrategy() # 短文本（< 5000 字符） text = "这是短文本。" * 100 chunks = strategy._split_into_chunks(text, max_chunk_size=5000) # 验证只有一个分块 assert len(chunks) == 1 # 验证内容不变 chunk_text, offset = chunks[0] assert chunk_text == text assert offset == 0 def test_split_into_chunks_sentence_boundaries(self): """测试分块在句子边界分割""" from kb_workflow.srt.semantic_engine import UnifiedSegmentationStrategy strategy = UnifiedSegmentationStrategy() # 创建包含不同标点符号的文本 text = "第一句。第二句！第三句？第四句\n" * 1000 chunks = strategy._split_into_chunks(text, max_chunk_size=5000) # 验证每个分块在句子边界结束（以标点符号或换行符结尾） for chunk_text, offset in chunks: # 分块应该以句子结束字符结尾 assert chunk_text[-1] in '。！？\n' def test_split_into_chunks_empty_text(self): """测试空文本分块""" from kb_workflow.srt.semantic_engine import UnifiedSegmentationStrategy strategy = UnifiedSegmentationStrategy() chunks = strategy._split_into_chunks("", max_chunk_size=5000) # 空文本应该返回空列表（没有句子可以分块） assert len(chunks) == 0 class TestLongTextChunking: """测试长文本自动分块分段（Plan 14-04）""" def test_long_text_triggers_chunking(self, caplog): """测试长文本（>5000字符）自动触发分块处理""" import logging import json # 创建模拟 AI 客户端 mock_client = Mock() mock_client.api_key = "test-key" # 创建长文本（>5000 字符） long_text = "这是一个长句。" * 1000 # 计算会分成几个分块 expected_chunks = (len(long_text) + 4999) // 5000 # 为每个分块设置响应 responses = [] for i in range(expected_chunks): chunk_len = min(5000, len(long_text) - i * 5000) response = json.dumps({ "video_type": "知识类", "segments": [ { "title": "测试标题", "content": "分块内容" * (chunk_len // 4), "start_index": 0, "end_index": chunk_len } ], "actual_segment_count": 1 }, ensure_ascii=False) responses.append(response) mock_client.generate.side_effect = responses from kb_workflow.srt.semantic_engine import UnifiedSegmentationStrategy from kb_workflow.srt.parser import Segment strategy = UnifiedSegmentationStrategy(ai_client=mock_client) # 创建长文本字幕 segments = [Segment(1, 0.0, 1.0, long_text)] with caplog.at_level(logging.INFO): result = strategy.segment(segments, {"title": "测试"}) # 验证调用了 LLM 多次（每个分块一次） assert mock_client.generate.call_count == expected_chunks # 验证日志包含分块信息 assert any("splitting into chunks" in record.message.lower() for record in caplog.records) # 验证返回了分段结果 assert len(result) > 0 def test_short_text_no_chunking(self, caplog): """测试短文本（<=5000字符）不分块直接处理""" import logging import json mock_client = Mock() mock_client.api_key = "test-key" # 短文本（< 5000 字符） short_text = "这是短文本。" * 100 response = json.dumps({ "video_type": "知识类", "segments": [ { "title": "测试标题", "content": short_text, "start_index": 0, "end_index": len(short_text) } ], "actual_segment_count": 1 }, ensure_ascii=False) mock_client.generate.return_value = response from kb_workflow.srt.semantic_engine import UnifiedSegmentationStrategy from kb_workflow.srt.parser import Segment strategy = UnifiedSegmentationStrategy(ai_client=mock_client) segments = [Segment(1, 0.0, 1.0, short_text)] with caplog.at_level(logging.INFO): result = strategy.segment(segments, {"title": "测试"}) # 验证只调用了一次 LLM（不分块） assert mock_client.generate.call_count == 1 # 验证日志不包含分块信息 assert not any("splitting into chunks" in record.message.lower() for record in caplog.records) # 验证返回了分段结果 assert len(result) == 1 class TestChunkIndexCorrection: """测试分块索引修正（Plan 14-04）""" def test_chunk_index_correction(self): """测试分块索引相对于原文位置""" from kb_workflow.srt.semantic_engine import UnifiedSegmentationStrategy strategy = UnifiedSegmentationStrategy() # 模拟分块结果 chunk_text = "这是第一块。" offset = 100 # 创建模拟的分段结果（索引相对于分块） from kb_workflow.srt.semantic_engine import VideoSegment chunk_segments = [ VideoSegment( title="测试", content=chunk_text, start_index=0, end_index=len(chunk_text) ) ] # 合并分块结果 chunk_results = [(chunk_segments, offset)] merged = strategy._merge_segment_results(chunk_results) # 验证索引被修正为相对于原文 assert len(merged) == 1 assert merged[0].start_index == offset assert merged[0].end_index == offset + len(chunk_text) def test_multiple_chunks_index_correction(self): """测试多个分块的索引修正""" from kb_workflow.srt.semantic_engine import UnifiedSegmentationStrategy, VideoSegment strategy = UnifiedSegmentationStrategy() # 创建多个分块 chunk1_text = "第一块内容。" * 100 chunk2_text = "第二块内容。" * 100 chunk3_text = "第三块内容。" * 100 # 模拟分块结果 chunk1_segments = [VideoSegment(title="块1", content=chunk1_text, start_index=0, end_index=len(chunk1_text))] chunk2_segments = [VideoSegment(title="块2", content=chunk2_text, start_index=0, end_index=len(chunk2_text))] chunk3_segments = [VideoSegment(title="块3", content=chunk3_text, start_index=0, end_index=len(chunk3_text))] chunk_results = [ (chunk1_segments, 0), # 第一个分块偏移量为 0 (chunk2_segments, len(chunk1_text)), # 第二个分块偏移量为第一块的长度 (chunk3_segments, len(chunk1_text) + len(chunk2_text)) # 第三个分块偏移量为前两块的长度之和 ] # 合并分块结果 merged = strategy._merge_segment_results(chunk_results) # 验证所有分块的索引都被正确修正 assert len(merged) == 3 assert merged[0].start_index == 0 assert merged[0].end_index == len(chunk1_text) assert merged[1].start_index == len(chunk1_text) assert merged[1].end_index == len(chunk1_text) + len(chunk2_text) assert merged[2].start_index == len(chunk1_text) + len(chunk2_text) assert merged[2].end_index == len(chunk1_text) + len(chunk2_text) + len(chunk3_text) class TestChunkFallback: """测试分块失败时的降级处理（Plan 14-04）""" def test_chunk_fallback_to_rule_based(self, caplog): """测试单个分块失败时降级到规则分段""" import logging import json mock_client = Mock() mock_client.api_key = "test-key" # 创建长文本（会分成多个分块） long_text = "这是第一句。" * 1000 + "这是第二句。" * 1000 # 第一个分块成功，第二个分块失败 success_response = json.dumps({ "video_type": "知识类", "segments": [ { "title": "成功分块", "content": "内容", "start_index": 0, "end_index": 10 } ], "actual_segment_count": 1 }, ensure_ascii=False) # 第一个分块返回成功，第二个分块抛出异常 mock_client.generate.side_effect = [success_response, Exception("LLM error")] from kb_workflow.srt.semantic_engine import UnifiedSegmentationStrategy from kb_workflow.srt.parser import Segment strategy = UnifiedSegmentationStrategy(ai_client=mock_client) segments = [Segment(1, 0.0, 1.0, long_text)] with caplog.at_level(logging.WARNING): result = strategy.segment(segments, {"title": "测试"}) # 验证返回了分段结果（第一个分块的 LLM 结果 + 第二个分块的规则分段） assert len(result) > 0 # 验证日志包含分块失败和降级信息 assert any("failed" in record.message.lower() and "chunk" in record.message.lower() for record in caplog.records) def test_fallback_chunk_returns_valid_segments(self): """测试 _fallback_chunk 返回有效的 VideoSegment""" from kb_workflow.srt.semantic_engine import UnifiedSegmentationStrategy strategy = UnifiedSegmentationStrategy() chunk_text = "这是降级分块的内容。" * 100 offset = 500 # 调用降级方法 segments = strategy._fallback_chunk(chunk_text, offset) # 验证返回了分段列表 assert len(segments) > 0 # 验证所有分段都是 VideoSegment 类型 from kb_workflow.srt.semantic_engine import VideoSegment assert all(isinstance(seg, VideoSegment) for seg in segments) # 验证第一个分段的索引正确 assert segments[0].start_index == offset # 验证降级分段没有标题 assert all(seg.title is None for seg in segments)