starmen1265
ee5062ca9c
- 添加 TestSplitIntoChunks 测试类：测试文本分块功能 - 添加 TestLongTextChunking 测试类：测试长文本自动分块分段 - 添加 TestChunkIndexCorrection 测试类：测试分块索引修正 - 添加 TestChunkFallback 测试类：测试单个分块失败时降级到规则分段 测试覆盖： - 分块边界条件（空文本、短文本、长文本） - 句子边界分割（。！？\n） - 索引修正（start_index/end_index 相对于原文） - 分块失败降级 Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
993 lines
34 KiB
Python
993 lines
34 KiB
Python
"""测试语义分段的鲁棒性功能 (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)
|
||
|