159 lines
4.4 KiB
Python
159 lines
4.4 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
测试ZIP URL功能的脚本
|
||
"""
|
||
|
||
import requests
|
||
import json
|
||
import os
|
||
|
||
|
||
def test_missing_zip_url():
|
||
"""测试缺少zip_url参数的错误处理"""
|
||
|
||
base_url = "http://localhost:8000"
|
||
|
||
print("测试缺少zip_url参数...")
|
||
|
||
# 缺少zip_url的请求
|
||
test_request = {
|
||
"messages": [
|
||
{
|
||
"role": "user",
|
||
"content": "测试请求"
|
||
}
|
||
],
|
||
"model": "qwen3-next",
|
||
# 缺少zip_url参数
|
||
"stream": False
|
||
}
|
||
|
||
try:
|
||
response = requests.post(
|
||
f"{base_url}/chat/completions",
|
||
json=test_request,
|
||
timeout=10
|
||
)
|
||
|
||
if response.status_code == 400:
|
||
print("✅ 正确返回400错误（缺少zip_url）")
|
||
print(f"错误信息: {response.json()}")
|
||
else:
|
||
print(f"❌ 预期400错误，实际得到: {response.status_code}")
|
||
|
||
except Exception as e:
|
||
print(f"❌ 测试失败: {e}")
|
||
|
||
|
||
def test_zip_project_feature():
|
||
"""测试ZIP项目功能"""
|
||
|
||
# API基础URL
|
||
base_url = "http://localhost:8000"
|
||
|
||
print("测试ZIP项目功能...")
|
||
|
||
# 测试数据 - 使用一个示例ZIP文件URL
|
||
test_request = {
|
||
"messages": [
|
||
{
|
||
"role": "user",
|
||
"content": "请列出项目目录中的文件"
|
||
}
|
||
],
|
||
"model": "qwen3-next",
|
||
"model_server": "https://openrouter.ai/api/v1", # 示例model_server
|
||
"zip_url": "https://example.com/test-project.zip", # 示例URL，需要替换
|
||
"stream": False
|
||
}
|
||
|
||
try:
|
||
print("发送测试请求...")
|
||
response = requests.post(
|
||
f"{base_url}/chat/completions",
|
||
json=test_request,
|
||
timeout=30
|
||
)
|
||
|
||
print(f"响应状态码: {response.status_code}")
|
||
|
||
if response.status_code == 200:
|
||
print("✅ 请求成功")
|
||
result = response.json()
|
||
print(f"响应内容: {json.dumps(result, indent=2, ensure_ascii=False)}")
|
||
else:
|
||
print("❌ 请求失败")
|
||
print(f"错误信息: {response.text}")
|
||
|
||
except requests.exceptions.ConnectionError:
|
||
print("❌ 连接失败，请确保API服务正在运行")
|
||
except Exception as e:
|
||
print(f"❌ 测试失败: {e}")
|
||
|
||
|
||
def test_cache_cleanup():
|
||
"""测试缓存清理功能"""
|
||
|
||
base_url = "http://localhost:8000"
|
||
|
||
try:
|
||
print("测试缓存清理...")
|
||
response = requests.post(f"{base_url}/system/cleanup-cache")
|
||
|
||
if response.status_code == 200:
|
||
print("✅ 缓存清理成功")
|
||
print(f"响应: {response.json()}")
|
||
else:
|
||
print("❌ 缓存清理失败")
|
||
print(f"错误信息: {response.text}")
|
||
|
||
except Exception as e:
|
||
print(f"❌ 缓存清理测试失败: {e}")
|
||
|
||
|
||
def test_system_status():
|
||
"""测试系统状态"""
|
||
|
||
base_url = "http://localhost:8000"
|
||
|
||
try:
|
||
print("获取系统状态...")
|
||
response = requests.get(f"{base_url}/system/status")
|
||
|
||
if response.status_code == 200:
|
||
print("✅ 系统状态获取成功")
|
||
status = response.json()
|
||
print(f"系统状态: {json.dumps(status, indent=2, ensure_ascii=False)}")
|
||
else:
|
||
print("❌ 系统状态获取失败")
|
||
|
||
except Exception as e:
|
||
print(f"❌ 系统状态测试失败: {e}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
print("=== ZIP项目功能测试 ===")
|
||
|
||
# 测试系统状态
|
||
test_system_status()
|
||
print()
|
||
|
||
# 测试缺少zip_url参数的错误处理
|
||
test_missing_zip_url()
|
||
print()
|
||
|
||
# 测试ZIP项目功能
|
||
test_zip_project_feature()
|
||
print()
|
||
|
||
# 测试缓存清理
|
||
test_cache_cleanup()
|
||
|
||
print("\n=== 测试完成 ===")
|
||
print("\n使用说明:")
|
||
print("1. 确保API服务正在运行: python fastapi_app.py")
|
||
print("2. 将test_request中的zip_url替换为实际的ZIP文件URL")
|
||
print("3. 将model_server替换为实际的模型服务器地址")
|
||
print("4. 运行此脚本进行测试")
|
||
print("5. 可以通过POST /system/cleanup-cache清理缓存")
|
||
print("\n注意：现在model_server和zip_url参数都放在最外层，不再需要extra字段") |