forked from zhanggangyong/agent_management
329 lines
11 KiB
Python
329 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Agent Manager v2.0 - 综合测试脚本
|
|
测试新架构的所有主要功能
|
|
"""
|
|
|
|
import requests
|
|
import json
|
|
import time
|
|
import sys
|
|
|
|
BASE_URL = "http://localhost:8000"
|
|
|
|
def print_section(title):
|
|
"""打印测试章节标题"""
|
|
print(f"\n{'='*60}")
|
|
print(f" {title}")
|
|
print(f"{'='*60}\n")
|
|
|
|
def test_health_checks():
|
|
"""测试健康检查端点"""
|
|
print_section("1. 健康检查测试")
|
|
|
|
# 根端点
|
|
response = requests.get(f"{BASE_URL}/")
|
|
print(f"GET / : {response.status_code}")
|
|
print(json.dumps(response.json(), indent=2))
|
|
|
|
# 健康检查
|
|
response = requests.get(f"{BASE_URL}/health")
|
|
print(f"\nGET /health : {response.status_code}")
|
|
print(json.dumps(response.json(), indent=2))
|
|
|
|
# 就绪检查
|
|
response = requests.get(f"{BASE_URL}/ready")
|
|
print(f"\nGET /ready : {response.status_code}")
|
|
print(json.dumps(response.json(), indent=2))
|
|
|
|
def test_template_management():
|
|
"""测试模板管理"""
|
|
print_section("2. 模板管理测试")
|
|
|
|
# 列出所有模板
|
|
response = requests.get(f"{BASE_URL}/templates")
|
|
print(f"GET /templates : {response.status_code}")
|
|
templates = response.json()
|
|
print(f"找到 {len(templates)} 个模板")
|
|
for template in templates:
|
|
print(f" - {template['name']} ({template['agent_type']}): {template['display_name']}")
|
|
|
|
# 获取特定模板
|
|
if templates:
|
|
template_name = templates[0]['name']
|
|
response = requests.get(f"{BASE_URL}/templates/{template_name}")
|
|
print(f"\nGET /templates/{template_name} : {response.status_code}")
|
|
print(json.dumps(response.json(), indent=2))
|
|
|
|
# 创建自定义模板
|
|
print("\n创建自定义模板...")
|
|
new_template = {
|
|
"name": "test_custom_agent",
|
|
"display_name": "Test Custom Agent",
|
|
"description": "Test agent for v2.0 testing",
|
|
"agent_type": "custom",
|
|
"image": "nginx:latest",
|
|
"port": 80,
|
|
"env_requirements": {
|
|
"required": {
|
|
"API_KEY": "API key for testing"
|
|
},
|
|
"optional": {
|
|
"DEBUG": "Debug mode flag"
|
|
}
|
|
},
|
|
"cpu_request": "100m",
|
|
"cpu_limit": "200m",
|
|
"memory_request": "128Mi",
|
|
"memory_limit": "256Mi",
|
|
"min_replicas": 1,
|
|
"max_replicas": 3,
|
|
"target_cpu_utilization": 75
|
|
}
|
|
|
|
response = requests.post(f"{BASE_URL}/templates", json=new_template)
|
|
print(f"POST /templates : {response.status_code}")
|
|
if response.status_code == 201:
|
|
print("✓ 模板创建成功")
|
|
print(json.dumps(response.json(), indent=2))
|
|
elif response.status_code == 409:
|
|
print("✓ 模板已存在（预期行为）")
|
|
else:
|
|
print(f"✗ 创建失败: {response.text}")
|
|
|
|
def test_platform_agents():
|
|
"""测试平台Agent"""
|
|
print_section("3. 平台Agent测试")
|
|
|
|
# 创建平台Agent
|
|
print("创建平台Agent...")
|
|
platform_agent = {
|
|
"name": "test-platform-echo",
|
|
"template_name": "echo_agent",
|
|
"owner_id": "test_user_001",
|
|
"channel_id": "test_channel",
|
|
"tenant_id": "test_tenant"
|
|
}
|
|
|
|
response = requests.post(f"{BASE_URL}/platform-agents", json=platform_agent)
|
|
print(f"POST /platform-agents : {response.status_code}")
|
|
if response.status_code == 201:
|
|
print("✓ 平台Agent创建成功")
|
|
print(json.dumps(response.json(), indent=2))
|
|
agent_created = True
|
|
elif response.status_code == 409:
|
|
print("✓ Agent已存在（预期行为）")
|
|
agent_created = False
|
|
else:
|
|
print(f"✗ 创建失败: {response.text}")
|
|
agent_created = False
|
|
|
|
# 等待一下让Kubernetes创建资源
|
|
if agent_created:
|
|
print("\n等待5秒让Kubernetes创建资源...")
|
|
time.sleep(5)
|
|
|
|
# 列出平台Agents
|
|
print("\n列出所有平台Agents...")
|
|
response = requests.get(f"{BASE_URL}/platform-agents")
|
|
print(f"GET /platform-agents : {response.status_code}")
|
|
agents = response.json()
|
|
print(f"找到 {len(agents)} 个平台Agents")
|
|
for agent in agents:
|
|
print(f" - {agent['name']}: {agent['status']} (副本: {agent['current_replicas']})")
|
|
|
|
# 获取日志（如果Agent存在）
|
|
if agents:
|
|
agent_name = agents[0]['name']
|
|
print(f"\n获取Agent日志: {agent_name}")
|
|
response = requests.get(f"{BASE_URL}/platform-agents/{agent_name}/logs?lines=20")
|
|
print(f"GET /platform-agents/{agent_name}/logs : {response.status_code}")
|
|
if response.status_code == 200:
|
|
logs_data = response.json()
|
|
print(f"日志行数: {len(logs_data['logs'].split(chr(10)))}")
|
|
|
|
def test_custom_agents():
|
|
"""测试自定义Agent"""
|
|
print_section("4. 自定义Agent测试")
|
|
|
|
# 创建自定义Agent
|
|
print("创建自定义Agent...")
|
|
custom_agent = {
|
|
"name": "test-custom-nginx",
|
|
"template_name": "test_custom_agent",
|
|
"owner_id": "test_user_001",
|
|
"channel_id": "test_channel",
|
|
"environment_vars": {
|
|
"API_KEY": "test_api_key_12345",
|
|
"DEBUG": "true"
|
|
},
|
|
"cpu_request": "100m",
|
|
"cpu_limit": "200m",
|
|
"memory_request": "128Mi",
|
|
"memory_limit": "256Mi",
|
|
"scaling_config": {
|
|
"min_replicas": 1,
|
|
"max_replicas": 3,
|
|
"target_cpu_utilization": 75
|
|
}
|
|
}
|
|
|
|
response = requests.post(f"{BASE_URL}/custom-agents", json=custom_agent)
|
|
print(f"POST /custom-agents : {response.status_code}")
|
|
if response.status_code == 201:
|
|
print("✓ 自定义Agent创建成功")
|
|
print(json.dumps(response.json(), indent=2))
|
|
agent_created = True
|
|
elif response.status_code == 409:
|
|
print("✓ Agent已存在（预期行为）")
|
|
agent_created = False
|
|
else:
|
|
print(f"✗ 创建失败: {response.text}")
|
|
agent_created = False
|
|
|
|
# 列出自定义Agents
|
|
print("\n列出所有自定义Agents...")
|
|
response = requests.get(f"{BASE_URL}/custom-agents")
|
|
print(f"GET /custom-agents : {response.status_code}")
|
|
agents = response.json()
|
|
print(f"找到 {len(agents)} 个自定义Agents")
|
|
for agent in agents:
|
|
print(f" - {agent['name']}: {agent['status']}")
|
|
|
|
# 更新环境变量（如果Agent存在）
|
|
if agents and agent_created:
|
|
agent_name = agents[0]['name']
|
|
print(f"\n更新Agent环境变量: {agent_name}")
|
|
update_env = {
|
|
"environment_vars": {
|
|
"API_KEY": "updated_api_key_67890",
|
|
"DEBUG": "false"
|
|
}
|
|
}
|
|
response = requests.put(f"{BASE_URL}/custom-agents/{agent_name}/env", json=update_env)
|
|
print(f"PUT /custom-agents/{agent_name}/env : {response.status_code}")
|
|
if response.status_code == 200:
|
|
print("✓ 环境变量更新成功")
|
|
|
|
def test_statistics():
|
|
"""测试统计API"""
|
|
print_section("5. 统计API测试")
|
|
|
|
# 统计概览
|
|
response = requests.get(f"{BASE_URL}/stats/overview")
|
|
print(f"GET /stats/overview : {response.status_code}")
|
|
print(json.dumps(response.json(), indent=2))
|
|
|
|
# 按模板统计
|
|
print("\n按模板统计:")
|
|
response = requests.get(f"{BASE_URL}/stats/by-template")
|
|
print(f"GET /stats/by-template : {response.status_code}")
|
|
stats = response.json()
|
|
for stat in stats:
|
|
print(f" - {stat['template_name']}: {stat['agent_count']} agents, {stat['total_replicas']} replicas")
|
|
|
|
# 按所有者统计
|
|
print("\n按所有者统计:")
|
|
response = requests.get(f"{BASE_URL}/stats/by-owner")
|
|
print(f"GET /stats/by-owner : {response.status_code}")
|
|
stats = response.json()
|
|
for stat in stats:
|
|
print(f" - {stat['owner_id']}: {stat['agent_count']} agents "
|
|
f"(Platform: {stat['platform_agents']}, Custom: {stat['custom_agents']})")
|
|
|
|
def test_quotas():
|
|
"""测试配额API"""
|
|
print_section("6. 配额管理测试")
|
|
|
|
# 获取配额（测试用户）
|
|
owner_id = "test_user_001"
|
|
response = requests.get(f"{BASE_URL}/quotas/{owner_id}")
|
|
print(f"GET /quotas/{owner_id} : {response.status_code}")
|
|
if response.status_code == 200:
|
|
print(json.dumps(response.json(), indent=2))
|
|
else:
|
|
print(f"配额不存在（预期行为）: {response.status_code}")
|
|
|
|
# 获取默认租户配额
|
|
response = requests.get(f"{BASE_URL}/quotas/default_tenant")
|
|
print(f"\nGET /quotas/default_tenant : {response.status_code}")
|
|
if response.status_code == 200:
|
|
quota = response.json()
|
|
print(f"平台Pod配额: {quota['platform_pod_used']}/{quota['platform_pod_quota']}")
|
|
print(f"自定义CPU配额: {quota['custom_cpu_used']:.2f}/{quota['custom_cpu_quota']:.2f} 核")
|
|
print(f"自定义内存配额: {quota['custom_memory_used']:.2f}/{quota['custom_memory_quota']:.2f} MB")
|
|
|
|
def cleanup_test_resources():
|
|
"""清理测试资源"""
|
|
print_section("7. 清理测试资源")
|
|
|
|
cleanup = input("\n是否删除测试创建的Agents? (y/N): ").strip().lower()
|
|
if cleanup != 'y':
|
|
print("跳过清理")
|
|
return
|
|
|
|
# 删除测试平台Agent
|
|
print("\n删除平台Agent...")
|
|
response = requests.delete(f"{BASE_URL}/platform-agents/test-platform-echo")
|
|
print(f"DELETE /platform-agents/test-platform-echo : {response.status_code}")
|
|
if response.status_code in [200, 404]:
|
|
print("✓ 平台Agent已删除或不存在")
|
|
|
|
# 删除测试自定义Agent
|
|
print("\n删除自定义Agent...")
|
|
response = requests.delete(f"{BASE_URL}/custom-agents/test-custom-nginx")
|
|
print(f"DELETE /custom-agents/test-custom-nginx : {response.status_code}")
|
|
if response.status_code in [200, 404]:
|
|
print("✓ 自定义Agent已删除或不存在")
|
|
|
|
# 删除测试模板
|
|
print("\n删除测试模板...")
|
|
response = requests.delete(f"{BASE_URL}/templates/test_custom_agent")
|
|
print(f"DELETE /templates/test_custom_agent : {response.status_code}")
|
|
if response.status_code in [200, 404]:
|
|
print("✓ 测试模板已删除或不存在")
|
|
|
|
def main():
|
|
"""主测试流程"""
|
|
print("\n" + "="*60)
|
|
print(" Agent Manager v2.0 - 综合功能测试")
|
|
print("="*60)
|
|
print(f"\n测试服务器: {BASE_URL}")
|
|
print("\n确保服务正在运行: python app.py")
|
|
|
|
try:
|
|
# 测试连接
|
|
response = requests.get(f"{BASE_URL}/", timeout=5)
|
|
response.raise_for_status()
|
|
except Exception as e:
|
|
print(f"\n❌ 无法连接到服务器: {e}")
|
|
print("\n请先启动服务: python app.py")
|
|
sys.exit(1)
|
|
|
|
try:
|
|
# 执行所有测试
|
|
test_health_checks()
|
|
test_template_management()
|
|
test_platform_agents()
|
|
test_custom_agents()
|
|
test_statistics()
|
|
test_quotas()
|
|
cleanup_test_resources()
|
|
|
|
print_section("测试完成")
|
|
print("✅ 所有测试已执行完毕")
|
|
print("\n下一步:")
|
|
print("1. 查看 Swagger UI: http://localhost:8000/docs")
|
|
print("2. 查看数据库: sqlite3 agent_manager.db")
|
|
print("3. 查看 Kubernetes 资源: kubectl get all -n ai-agents")
|
|
|
|
except KeyboardInterrupt:
|
|
print("\n\n测试被中断")
|
|
except Exception as e:
|
|
print(f"\n❌ 测试失败: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
|
|
if __name__ == "__main__":
|
|
main()
|