test_context.py
3.91 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
"""
测试RequestContext功能的简单脚本
"""
import sys
import os
# 添加项目根目录到Python路径
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from context import RequestContext, RequestContextStage, create_request_context
def test_basic_context_functionality():
"""测试基本的context功能"""
print("=== 测试基本RequestContext功能 ===")
# 创建context
context = create_request_context("test123", "user456")
print(f"Request ID: {context.reqid}")
print(f"User ID: {context.uid}")
# 测试阶段计时
context.start_stage(RequestContextStage.QUERY_PARSING)
import time
time.sleep(0.1) # 模拟工作
duration = context.end_stage(RequestContextStage.QUERY_PARSING)
print(f"查询解析阶段耗时: {duration:.2f}ms")
# 测试存储查询分析结果
context.store_query_analysis(
original_query="iphone 13",
normalized_query="iphone 13",
rewritten_query="apple iphone 13",
detected_language="en"
)
# 测试存储中间结果
context.store_intermediate_result('test_key', {'test': 'value'})
# 获取摘要
summary = context.get_summary()
print("Context摘要:")
print(f" - 原始查询: {summary['query_analysis']['original_query']}")
print(f" - 检测语言: {summary['query_analysis']['detected_language']}")
print(f" - 阶段耗时: {summary['performance']['stage_timings_ms']}")
print("✅ 基本功能测试通过\n")
def test_context_as_context_manager():
"""测试context作为上下文管理器的功能"""
print("=== 测试上下文管理器功能 ===")
# 使用上下文管理器
with create_request_context("cm123", "user789") as context:
context.start_stage(RequestContextStage.QUERY_PARSING)
import time
time.sleep(0.05)
context.end_stage(RequestContextStage.QUERY_PARSING)
context.start_stage(RequestContextStage.QUERY_BUILDING)
time.sleep(0.03)
context.end_stage(RequestContextStage.QUERY_BUILDING)
print(f"Context ID: {context.reqid}")
# 退出时会自动记录性能摘要
print("✅ 上下文管理器测试通过\n")
def test_error_handling():
"""测试错误处理功能"""
print("=== 测试错误处理功能 ===")
context = create_request_context("error123")
# 设置错误
try:
raise ValueError("这是一个测试错误")
except Exception as e:
context.set_error(e)
print(f"有错误: {context.has_error()}")
print(f"错误信息: {context.metadata['error_info']}")
print("✅ 错误处理测试通过\n")
def test_performance_summary():
"""测试性能摘要功能"""
print("=== 测试性能摘要功能 ===")
context = create_request_context("perf123")
# 模拟多个阶段
stages = [
RequestContextStage.QUERY_PARSING,
RequestContextStage.BOOLEAN_PARSING,
RequestContextStage.QUERY_BUILDING,
RequestContextStage.ELASTICSEARCH_SEARCH,
RequestContextStage.RESULT_PROCESSING
]
import time
durations = [50, 20, 80, 150, 30] # 模拟各阶段耗时(ms)
for stage, expected_duration in zip(stages, durations):
context.start_stage(stage)
time.sleep(expected_duration / 1000.0) # 转换为秒
context.end_stage(stage)
# 设置总耗时
context.performance_metrics.total_duration = sum(durations)
# 计算百分比
percentages = context.calculate_stage_percentages()
print("各阶段耗时占比:")
for stage, percentage in percentages.items():
print(f" - {stage}: {percentage}%")
print("✅ 性能摘要测试通过\n")
if __name__ == "__main__":
print("开始测试RequestContext功能...\n")
test_basic_context_functionality()
test_context_as_context_manager()
test_error_handling()
test_performance_summary()
print("🎉 所有测试通过！RequestContext功能正常。")