test_search_integration.py
2.59 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
#!/usr/bin/env python3
"""
测试搜索集成的自测脚本
验证请求上下文和日志系统是否正常工作
"""
import sys
import os
# 添加项目路径
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
def test_search_integration():
"""测试搜索集成"""
print("🧪 开始搜索集成自测...")
try:
# 导入模块
from context.request_context import create_request_context
from utils.logger import get_logger, setup_logging
# 设置日志
setup_logging(log_level="INFO", log_dir="test_logs")
logger = get_logger("test")
print("✅ 模块导入成功")
# 创建请求上下文
context = create_request_context("test123", "testuser")
print(f"✅ 请求上下文创建成功: reqid={context.reqid}")
# 测试日志记录
context.logger.info("测试日志记录", extra={'reqid': context.reqid, 'uid': context.uid})
print("✅ 日志记录正常")
# 测试存储中间结果
context.store_intermediate_result("test_query", "芭比娃娃")
context.store_intermediate_result("test_language", "zh")
print("✅ 中间结果存储正常")
# 测试查询分析存储
context.store_query_analysis(
original_query="芭比娃娃",
normalized_query="芭比娃娃",
rewritten_query="芭比娃娃",
detected_language="zh",
domain="default"
)
print("✅ 查询分析存储正常")
# 测试性能摘要
context.log_performance_summary()
print("✅ 性能摘要记录正常")
# 测试完整的上下文摘要
summary = context.get_summary()
print(f"✅ 上下文摘要生成成功，包含 {len(str(summary))} 字符的数据")
print("\n📊 测试摘要:")
print(f" 请求ID: {summary['request_info']['reqid']}")
print(f" 用户ID: {summary['request_info']['uid']}")
print(f" 查询: '{summary['query_analysis']['original_query']}'")
print(f" 语言: {summary['query_analysis']['detected_language']}")
print("\n🎉 所有自测通过！搜索集成功能正常工作。")
return True
except Exception as e:
print(f"❌ 自测失败: {e}")
import traceback
traceback.print_exc()
return False
if __name__ == "__main__":
success = test_search_integration()
if success:
print("\n✨ 系统已就绪，可以正常处理搜索请求！")
else:
print("\n💥 请检查错误信息并修复问题")
sys.exit(1)