Blame view

test_search_integration.py 2.59 KB
16c42787   tangwang   feat: implement r...
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)