107 lines
3.0 KiB
Python
107 lines
3.0 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
测试Soil Water Balance Model API
|
||
"""
|
||
|
||
import requests
|
||
import json
|
||
import pandas as pd
|
||
import os
|
||
|
||
def test_health_check():
|
||
"""测试健康检查端点"""
|
||
try:
|
||
response = requests.get("http://localhost:8000/health")
|
||
if response.status_code == 200:
|
||
print("✓ 健康检查通过")
|
||
return True
|
||
else:
|
||
print("✗ 健康检查失败")
|
||
return False
|
||
except Exception as e:
|
||
print(f"✗ 健康检查错误: {e}")
|
||
return False
|
||
|
||
def load_test_data():
|
||
"""加载测试数据"""
|
||
base_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), os.pardir))
|
||
|
||
# 加载参数文件
|
||
with open(os.path.join(base_dir, 'data', 'management.json')) as f:
|
||
management = json.load(f)
|
||
|
||
with open(os.path.join(base_dir, 'data', 'SOILPROP.json')) as f:
|
||
soil_params = json.load(f)
|
||
|
||
with open(os.path.join(base_dir, 'data', 'N_organ.json')) as f:
|
||
n_params = json.load(f)
|
||
|
||
with open(os.path.join(base_dir, 'data', 'plant_para.json')) as f:
|
||
plant_params = json.load(f)
|
||
|
||
with open(os.path.join(base_dir, 'data', 'para.json')) as f:
|
||
cultivar_params = json.load(f)
|
||
|
||
# 加载气象数据
|
||
weather_data = pd.read_csv(
|
||
os.path.join(base_dir, 'data', '57278_2024.csv'),
|
||
index_col='date',
|
||
parse_dates=True
|
||
)
|
||
|
||
# 转换为字典列表格式
|
||
weather_list = []
|
||
for date, row in weather_data.iterrows():
|
||
weather_dict = {'date': date.strftime('%m/%d/%Y')}
|
||
weather_dict.update(row.to_dict())
|
||
weather_list.append(weather_dict)
|
||
|
||
return {
|
||
"start_date": "10/9/2023",
|
||
"end_date": "10/15/2023", # 缩短测试时间
|
||
"cultivar_params": cultivar_params,
|
||
"plant_params": plant_params,
|
||
"soil_params": soil_params,
|
||
"management": management,
|
||
"n_params": n_params,
|
||
"weather_data": weather_list
|
||
}
|
||
|
||
def test_simulation():
|
||
"""测试模拟计算"""
|
||
try:
|
||
test_data = load_test_data()
|
||
|
||
response = requests.post(
|
||
"http://localhost:8000/simulate",
|
||
json=test_data,
|
||
headers={"Content-Type": "application/json"}
|
||
)
|
||
|
||
if response.status_code == 200:
|
||
result = response.json()
|
||
print("✓ 模拟计算成功")
|
||
print(f"结果包含 {len(result.get('results', {}))} 天的计算结果")
|
||
return True
|
||
else:
|
||
print(f"✗ 模拟计算失败: {response.status_code}")
|
||
print(response.text)
|
||
return False
|
||
except Exception as e:
|
||
print(f"✗ 模拟计算错误: {e}")
|
||
return False
|
||
|
||
if __name__ == "__main__":
|
||
print("开始测试Soil Water Balance Model API...")
|
||
|
||
# 测试健康检查
|
||
if not test_health_check():
|
||
print("API服务未运行，请先启动API服务")
|
||
exit(1)
|
||
|
||
# 测试模拟计算
|
||
if test_simulation():
|
||
print("所有测试通过！")
|
||
else:
|
||
print("测试失败")
|