ShopTRAINING/test/test_api_call.py
2025-07-02 11:05:23 +08:00

104 lines
2.9 KiB
Python

#!/usr/bin/env python
"""
测试API调用
"""
import requests
import json
def test_training_api():
"""测试训练API调用"""
print("=== 测试训练API ===")
# 准备训练请求数据
training_data = {
"product_id": "P001",
"model_type": "tcn",
"epochs": 5,
"training_mode": "product",
"store_id": None,
"aggregation_method": "sum"
}
try:
# 发送训练请求
response = requests.post(
"http://localhost:5000/api/train",
json=training_data,
timeout=10
)
print(f"响应状态码: {response.status_code}")
if response.status_code == 200:
result = response.json()
print("训练请求成功发送")
print(f"任务ID: {result.get('task_id', 'N/A')}")
else:
print(f"训练请求失败: {response.text}")
except requests.exceptions.ConnectionError:
print("无法连接到API服务器 (http://localhost:5000)")
print("请确保后端服务器正在运行")
except Exception as e:
print(f"API调用失败: {e}")
def test_products_api():
"""测试产品列表API"""
print("\n=== 测试产品列表API ===")
try:
response = requests.get("http://localhost:5000/api/products", timeout=5)
print(f"响应状态码: {response.status_code}")
if response.status_code == 200:
products = response.json()
print(f"获取到 {len(products)} 个产品")
if products:
print(f"第一个产品: {products[0]}")
else:
print(f"获取产品列表失败: {response.text}")
except requests.exceptions.ConnectionError:
print("无法连接到API服务器")
except Exception as e:
print(f"API调用失败: {e}")
def test_stores_api():
"""测试店铺列表API"""
print("\n=== 测试店铺列表API ===")
try:
response = requests.get("http://localhost:5000/api/stores", timeout=5)
print(f"响应状态码: {response.status_code}")
if response.status_code == 200:
stores = response.json()
print(f"获取到 {len(stores)} 个店铺")
if stores:
print(f"第一个店铺: {stores[0]}")
else:
print(f"获取店铺列表失败: {response.text}")
except requests.exceptions.ConnectionError:
print("无法连接到API服务器")
except Exception as e:
print(f"API调用失败: {e}")
def main():
"""主测试函数"""
print("开始API测试")
# 测试产品列表API
test_products_api()
# 测试店铺列表API
test_stores_api()
# 测试训练API
test_training_api()
print("\nAPI测试完成")
if __name__ == "__main__":
main()