test_api.py 5.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152
  1. #!/usr/bin/env python3
  2. """
  3. Custom React Agent API 测试脚本
  4. 测试基本的API功能,包括:
  5. 1. 健康检查
  6. 2. 普通问答
  7. 3. SQL查询
  8. 4. 错误处理
  9. 运行前请确保API服务已启动:
  10. python api.py
  11. """
  12. import asyncio
  13. import aiohttp
  14. import json
  15. import sys
  16. from typing import Dict, Any
  17. API_BASE_URL = "http://localhost:8000"
  18. class APITester:
  19. """API测试类"""
  20. def __init__(self, base_url: str = API_BASE_URL):
  21. self.base_url = base_url
  22. self.session = None
  23. async def __aenter__(self):
  24. self.session = aiohttp.ClientSession()
  25. return self
  26. async def __aexit__(self, exc_type, exc_val, exc_tb):
  27. if self.session:
  28. await self.session.close()
  29. async def test_health_check(self) -> bool:
  30. """测试健康检查"""
  31. print("🔍 测试健康检查...")
  32. try:
  33. async with self.session.get(f"{self.base_url}/health") as response:
  34. if response.status == 200:
  35. data = await response.json()
  36. print(f" ✅ 健康检查通过: {data}")
  37. return True
  38. else:
  39. print(f" ❌ 健康检查失败: HTTP {response.status}")
  40. return False
  41. except Exception as e:
  42. print(f" ❌ 健康检查异常: {e}")
  43. return False
  44. async def test_chat_api(self, question: str, user_id: str = "test_user",
  45. thread_id: str = None) -> Dict[str, Any]:
  46. """测试聊天API"""
  47. print(f"\n💬 测试问题: {question}")
  48. payload = {
  49. "question": question,
  50. "user_id": user_id
  51. }
  52. if thread_id:
  53. payload["thread_id"] = thread_id
  54. try:
  55. async with self.session.post(
  56. f"{self.base_url}/api/chat",
  57. json=payload,
  58. headers={"Content-Type": "application/json"}
  59. ) as response:
  60. response_data = await response.json()
  61. print(f" 📊 HTTP状态: {response.status}")
  62. print(f" 📋 响应代码: {response_data.get('code')}")
  63. print(f" 🎯 成功状态: {response_data.get('success')}")
  64. if response_data.get('success'):
  65. data = response_data.get('data', {})
  66. print(f" 💡 回答: {data.get('response', '')[:100]}...")
  67. if 'sql' in data:
  68. print(f" 🗄️ SQL: {data['sql'][:100]}...")
  69. if 'records' in data:
  70. records = data['records']
  71. print(f" 📈 数据行数: {records.get('total_row_count', 0)}")
  72. meta = data.get('react_agent_meta', {})
  73. print(f" 🔧 使用工具: {meta.get('tools_used', [])}")
  74. print(f" 🆔 会话ID: {meta.get('thread_id', '')}")
  75. return response_data
  76. else:
  77. error = response_data.get('error', '未知错误')
  78. print(f" ❌ 请求失败: {error}")
  79. return response_data
  80. except Exception as e:
  81. print(f" ❌ 请求异常: {e}")
  82. return {"success": False, "error": str(e)}
  83. async def run_test_suite(self):
  84. """运行完整的测试套件"""
  85. print("🚀 开始API测试套件")
  86. print("=" * 50)
  87. # 1. 健康检查
  88. health_ok = await self.test_health_check()
  89. if not health_ok:
  90. print("❌ 健康检查失败,停止测试")
  91. return
  92. # 2. 普通问答测试
  93. await self.test_chat_api("你好,你是谁?")
  94. # 3. SQL查询测试(假设有相关数据)
  95. result1 = await self.test_chat_api("请查询服务区的收入情况")
  96. # 4. 上下文对话测试
  97. thread_id = None
  98. if result1.get('success'):
  99. thread_id = result1.get('data', {}).get('react_agent_meta', {}).get('thread_id')
  100. if thread_id:
  101. await self.test_chat_api("请详细解释一下", thread_id=thread_id)
  102. # 5. 错误处理测试
  103. await self.test_chat_api("") # 空问题
  104. await self.test_chat_api("a" * 3000) # 超长问题
  105. print("\n" + "=" * 50)
  106. print("✅ 测试套件完成")
  107. async def main():
  108. """主函数"""
  109. print("Custom React Agent API 测试工具")
  110. print("请确保API服务已在 http://localhost:8000 启动")
  111. print()
  112. # 检查是否要运行特定测试
  113. if len(sys.argv) > 1:
  114. question = " ".join(sys.argv[1:])
  115. async with APITester() as tester:
  116. await tester.test_chat_api(question)
  117. else:
  118. # 运行完整测试套件
  119. async with APITester() as tester:
  120. await tester.run_test_suite()
  121. if __name__ == "__main__":
  122. asyncio.run(main())