69 lines
2.1 KiB
Python
69 lines
2.1 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
快速API测试
|
|
"""
|
|
import requests
|
|
import json
|
|
import time
|
|
|
|
def test_api():
|
|
print("=== 快速API测试 ===")
|
|
|
|
# 测试API是否运行
|
|
try:
|
|
response = requests.get('http://localhost:5000/api/version')
|
|
if response.status_code == 200:
|
|
print("OK API服务器运行正常")
|
|
print(f"版本信息: {response.json()}")
|
|
else:
|
|
print(f"ERROR API响应异常: {response.status_code}")
|
|
return False
|
|
except Exception as e:
|
|
print(f"ERROR 无法连接API: {e}")
|
|
return False
|
|
|
|
# 启动一个简短的训练任务
|
|
print("\n=== 启动训练任务 ===")
|
|
training_data = {
|
|
"product_id": "P001",
|
|
"model_type": "mlstm",
|
|
"epochs": 2, # 只训练2个epoch
|
|
"training_mode": "product"
|
|
}
|
|
|
|
try:
|
|
print("发送训练请求...")
|
|
response = requests.post(
|
|
'http://localhost:5000/api/training/start',
|
|
json=training_data
|
|
)
|
|
|
|
if response.status_code == 200:
|
|
result = response.json()
|
|
task_id = result.get('task_id')
|
|
print(f"OK 训练任务启动成功: {task_id}")
|
|
|
|
# 监控状态5次
|
|
for i in range(5):
|
|
time.sleep(3)
|
|
status_response = requests.get(f'http://localhost:5000/api/training/status/{task_id}')
|
|
if status_response.status_code == 200:
|
|
status = status_response.json().get('status', 'unknown')
|
|
print(f"检查{i+1}: 状态={status}")
|
|
if status in ['completed', 'failed']:
|
|
break
|
|
else:
|
|
print(f"状态检查失败: {status_response.status_code}")
|
|
|
|
return True
|
|
else:
|
|
print(f"ERROR 训练启动失败: {response.status_code}")
|
|
print(f"响应: {response.text}")
|
|
return False
|
|
|
|
except Exception as e:
|
|
print(f"ERROR 训练请求失败: {e}")
|
|
return False
|
|
|
|
if __name__ == "__main__":
|
|
test_api() |