forked from xiaohei/taiji-AI-PAD
150 lines
4.6 KiB
Python
150 lines
4.6 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
测试用户接口
|
|
- GET /api/user/models - 获取可用模型列表
|
|
- GET /api/user/custom-agents/templates - 获取框架模板列表
|
|
"""
|
|
|
|
import asyncio
|
|
import httpx
|
|
import os
|
|
from typing import Dict
|
|
|
|
# 测试配置
|
|
BASE_URL = os.getenv("BASE_URL", "http://localhost:8002")
|
|
# 需要先登录获取 token，这里使用测试token
|
|
TEST_TOKEN = os.getenv("TEST_TOKEN", "")
|
|
|
|
|
|
async def test_get_models(client: httpx.AsyncClient, token: str):
|
|
"""测试获取模型列表接口"""
|
|
print("\n" + "="*60)
|
|
print("测试: GET /api/user/models")
|
|
print("="*60)
|
|
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
|
|
try:
|
|
response = await client.get(
|
|
f"{BASE_URL}/api/user/models",
|
|
headers=headers,
|
|
timeout=10.0
|
|
)
|
|
|
|
print(f"状态码: {response.status_code}")
|
|
print(f"响应体:\n{response.text}")
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
if data.get("success"):
|
|
models = data.get("data", {}).get("models", [])
|
|
print(f"\n✅ 成功获取 {len(models)} 个模型")
|
|
for model in models:
|
|
print(f" - {model.get('name')} ({model.get('id')})")
|
|
print(f" Provider: {model.get('provider')}, Context: {model.get('contextWindow')}")
|
|
else:
|
|
print(f"❌ 失败: {data.get('message')}")
|
|
else:
|
|
print(f"❌ 请求失败: {response.status_code}")
|
|
|
|
except Exception as e:
|
|
print(f"❌ 异常: {str(e)}")
|
|
|
|
|
|
async def test_get_templates(client: httpx.AsyncClient, token: str):
|
|
"""测试获取框架模板列表接口"""
|
|
print("\n" + "="*60)
|
|
print("测试: GET /api/user/custom-agents/templates")
|
|
print("="*60)
|
|
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
|
|
try:
|
|
response = await client.get(
|
|
f"{BASE_URL}/api/user/custom-agents/templates",
|
|
headers=headers,
|
|
timeout=10.0
|
|
)
|
|
|
|
print(f"状态码: {response.status_code}")
|
|
print(f"响应体:\n{response.text}")
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
if data.get("success"):
|
|
templates = data.get("data", {}).get("templates", [])
|
|
print(f"\n✅ 成功获取 {len(templates)} 个框架模板")
|
|
for template in templates:
|
|
print(f" - {template}")
|
|
else:
|
|
print(f"❌ 失败: {data.get('message')}")
|
|
else:
|
|
print(f"❌ 请求失败: {response.status_code}")
|
|
|
|
except Exception as e:
|
|
print(f"❌ 异常: {str(e)}")
|
|
|
|
|
|
async def login(client: httpx.AsyncClient, username: str, password: str) -> str:
|
|
"""登录获取token"""
|
|
print("\n" + "="*60)
|
|
print("登录获取Token")
|
|
print("="*60)
|
|
|
|
try:
|
|
response = await client.post(
|
|
f"{BASE_URL}/api/auth/login",
|
|
json={"username": username, "password": password},
|
|
timeout=10.0
|
|
)
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
if data.get("success"):
|
|
token = data.get("data", {}).get("access_token")
|
|
print(f"✅ 登录成功")
|
|
return token
|
|
else:
|
|
print(f"❌ 登录失败: {data.get('message')}")
|
|
else:
|
|
print(f"❌ 登录失败: {response.status_code}")
|
|
print(f"响应: {response.text}")
|
|
|
|
except Exception as e:
|
|
print(f"❌ 登录异常: {str(e)}")
|
|
|
|
return ""
|
|
|
|
|
|
async def main():
|
|
"""主函数"""
|
|
print("用户接口测试工具")
|
|
print(f"服务地址: {BASE_URL}")
|
|
|
|
async with httpx.AsyncClient() as client:
|
|
# 获取token
|
|
token = TEST_TOKEN
|
|
if not token:
|
|
# 如果没有提供token，尝试登录
|
|
username = os.getenv("TEST_USERNAME", "admin")
|
|
password = os.getenv("TEST_PASSWORD", "admin123")
|
|
token = await login(client, username, password)
|
|
|
|
if not token:
|
|
print("\n❌ 无法获取Token，请设置 TEST_TOKEN 环境变量或提供登录凭据")
|
|
return
|
|
|
|
print(f"\n使用Token: {token[:20]}...")
|
|
|
|
# 测试接口
|
|
await test_get_models(client, token)
|
|
await test_get_templates(client, token)
|
|
|
|
print("\n" + "="*60)
|
|
print("测试完成")
|
|
print("="*60)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|