|
|
#!/usr/bin/env python3
|
|
|
"""
|
|
|
测试前后端API集成的脚本
|
|
|
"""
|
|
|
import requests
|
|
|
import json
|
|
|
import time
|
|
|
|
|
|
# API基础URL
|
|
|
BASE_URL = "http://127.0.0.1:8000/api"
|
|
|
|
|
|
def test_health_check():
|
|
|
"""测试健康检查端点"""
|
|
|
try:
|
|
|
response = requests.get(f"{BASE_URL}/health/")
|
|
|
print(f"✓ 健康检查: {response.status_code} - {response.json()}")
|
|
|
return response.status_code == 200
|
|
|
except Exception as e:
|
|
|
print(f"✗ 健康检查失败: {e}")
|
|
|
return False
|
|
|
|
|
|
def test_preprocess():
|
|
|
"""测试预处理端点"""
|
|
|
try:
|
|
|
data = {"test": "data"}
|
|
|
response = requests.post(f"{BASE_URL}/preprocess/", json=data)
|
|
|
result = response.json()
|
|
|
print(f"✓ 预处理: {response.status_code} - {result.get('message', '无消息')}")
|
|
|
return response.status_code == 200 or response.status_code == 400 # 400可能是数据为空
|
|
|
except Exception as e:
|
|
|
print(f"✗ 预处理失败: {e}")
|
|
|
return False
|
|
|
|
|
|
def test_merge():
|
|
|
"""测试格式合并端点"""
|
|
|
try:
|
|
|
data = {"test": "data"}
|
|
|
response = requests.post(f"{BASE_URL}/merge/", json=data)
|
|
|
result = response.json()
|
|
|
print(f"✓ 格式合并: {response.status_code} - {result.get('message', '无消息')}")
|
|
|
return response.status_code == 200 or response.status_code == 400
|
|
|
except Exception as e:
|
|
|
print(f"✗ 格式合并失败: {e}")
|
|
|
return False
|
|
|
|
|
|
def test_correct():
|
|
|
"""测试单词纠错端点"""
|
|
|
try:
|
|
|
data = {"test": "data"}
|
|
|
response = requests.post(f"{BASE_URL}/correct/", json=data)
|
|
|
result = response.json()
|
|
|
print(f"✓ 单词纠错: {response.status_code} - {result.get('message', '无消息')}")
|
|
|
return response.status_code == 200 or response.status_code == 400
|
|
|
except Exception as e:
|
|
|
print(f"✗ 单词纠错失败: {e}")
|
|
|
return False
|
|
|
|
|
|
def test_analyze():
|
|
|
"""测试大模型分析端点"""
|
|
|
try:
|
|
|
data = {"test": "data"}
|
|
|
response = requests.post(f"{BASE_URL}/analyze/", json=data)
|
|
|
result = response.json()
|
|
|
print(f"✓ 大模型分析: {response.status_code} - {result.get('message', '无消息')}")
|
|
|
return response.status_code == 200 or response.status_code == 400
|
|
|
except Exception as e:
|
|
|
print(f"✗ 大模型分析失败: {e}")
|
|
|
return False
|
|
|
|
|
|
def test_data_endpoint():
|
|
|
"""测试数据获取端点"""
|
|
|
try:
|
|
|
response = requests.get(f"{BASE_URL}/data/")
|
|
|
result = response.json()
|
|
|
print(f"✓ 数据获取: {response.status_code} - {result.get('message', '无消息')}")
|
|
|
return True
|
|
|
except Exception as e:
|
|
|
print(f"✗ 数据获取失败: {e}")
|
|
|
return False
|
|
|
|
|
|
def main():
|
|
|
"""运行所有测试"""
|
|
|
print("开始测试前后端API集成...")
|
|
|
print("=" * 50)
|
|
|
|
|
|
tests = [
|
|
|
("健康检查", test_health_check),
|
|
|
("数据获取", test_data_endpoint),
|
|
|
("预处理", test_preprocess),
|
|
|
("格式合并", test_merge),
|
|
|
("单词纠错", test_correct),
|
|
|
("大模型分析", test_analyze),
|
|
|
]
|
|
|
|
|
|
passed = 0
|
|
|
total = len(tests)
|
|
|
|
|
|
for name, test_func in tests:
|
|
|
print(f"\n测试 {name}...")
|
|
|
if test_func():
|
|
|
passed += 1
|
|
|
time.sleep(0.5) # 避免请求过快
|
|
|
|
|
|
print("=" * 50)
|
|
|
print(f"测试完成: {passed}/{total} 通过")
|
|
|
|
|
|
if passed == total:
|
|
|
print("🎉 所有API端点测试通过！前后端集成成功。")
|
|
|
else:
|
|
|
print("⚠️ 部分测试失败，请检查后端服务状态。")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
|
main() |
