init
This commit is contained in:
Ubuntu
@@ -0,0 +1,90 @@
|
||||
"""
|
||||
测试脚本 - 创建AI Agent
|
||||
"""
|
||||
import requests
|
||||
import json
|
||||
import sys
|
||||
|
||||
# 配置
|
||||
BASE_URL = "http://localhost:8000"
|
||||
|
||||
|
||||
def test_create_agent():
|
||||
"""测试创建Agent"""
|
||||
print("=" * 50)
|
||||
print("测试: 创建AI Agent")
|
||||
print("=" * 50)
|
||||
|
||||
# 测试数据
|
||||
test_cases = [
|
||||
{
|
||||
"name": "test-echo-1",
|
||||
"template": "echo_agent",
|
||||
"config": {
|
||||
"replicas": 1,
|
||||
"cpu_request": "100m",
|
||||
"cpu_limit": "500m",
|
||||
"memory_request": "128Mi",
|
||||
"memory_limit": "512Mi"
|
||||
}
|
||||
},
|
||||
{
|
||||
"name": "test-chat-1",
|
||||
"template": "chat_agent",
|
||||
"config": {
|
||||
"replicas": 1,
|
||||
"cpu_request": "200m",
|
||||
"cpu_limit": "1000m",
|
||||
"memory_request": "256Mi",
|
||||
"memory_limit": "1Gi"
|
||||
}
|
||||
},
|
||||
{
|
||||
"name": "test-code-1",
|
||||
"template": "code_agent",
|
||||
"config": {
|
||||
"replicas": 1
|
||||
}
|
||||
}
|
||||
]
|
||||
|
||||
for i, test_data in enumerate(test_cases, 1):
|
||||
print(f"\n测试用例 {i}: 创建 {test_data['name']}")
|
||||
print(f"模板: {test_data['template']}")
|
||||
print(f"配置: {json.dumps(test_data['config'], indent=2)}")
|
||||
|
||||
try:
|
||||
response = requests.post(
|
||||
f"{BASE_URL}/agents",
|
||||
json=test_data,
|
||||
headers={"Content-Type": "application/json"}
|
||||
)
|
||||
|
||||
print(f"\n状态码: {response.status_code}")
|
||||
|
||||
if response.status_code == 200:
|
||||
result = response.json()
|
||||
print("✅ 创建成功!")
|
||||
print(f"响应: {json.dumps(result, indent=2, ensure_ascii=False)}")
|
||||
else:
|
||||
print(f"❌ 创建失败!")
|
||||
print(f"错误: {response.text}")
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ 请求失败: {str(e)}")
|
||||
|
||||
print("-" * 50)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
# 先检查服务是否可用
|
||||
print("检查服务状态...")
|
||||
response = requests.get(f"{BASE_URL}")
|
||||
print(f"服务状态: {response.json()}\n")
|
||||
|
||||
test_create_agent()
|
||||
|
||||
except requests.exceptions.ConnectionError:
|
||||
print("❌ 无法连接到服务。请确保服务正在运行: python app.py")
|
||||
sys.exit(1)
|
||||
@@ -0,0 +1,64 @@
|
||||
"""
|
||||
测试脚本 - 删除Agent
|
||||
"""
|
||||
import requests
|
||||
import json
|
||||
import sys
|
||||
|
||||
# 配置
|
||||
BASE_URL = "http://localhost:8000"
|
||||
|
||||
|
||||
def test_delete_agent():
|
||||
"""测试删除Agent"""
|
||||
print("=" * 50)
|
||||
print("测试: 删除AI Agent")
|
||||
print("=" * 50)
|
||||
|
||||
# 测试的Agent名称列表
|
||||
agent_names = ["test-echo-1", "test-chat-1", "test-code-1"]
|
||||
|
||||
for agent_name in agent_names:
|
||||
print(f"\n删除Agent: {agent_name}")
|
||||
|
||||
try:
|
||||
response = requests.delete(f"{BASE_URL}/agents/{agent_name}")
|
||||
|
||||
print(f"状态码: {response.status_code}")
|
||||
|
||||
if response.status_code == 200:
|
||||
result = response.json()
|
||||
print("✅ 删除成功!")
|
||||
print(f"响应: {json.dumps(result, indent=2, ensure_ascii=False)}")
|
||||
elif response.status_code == 404:
|
||||
print(f"⚠️ Agent不存在")
|
||||
print(f"错误: {response.text}")
|
||||
else:
|
||||
print(f"❌ 删除失败!")
|
||||
print(f"错误: {response.text}")
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ 请求失败: {str(e)}")
|
||||
|
||||
print("-" * 50)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
# 先检查服务是否可用
|
||||
print("检查服务状态...")
|
||||
response = requests.get(f"{BASE_URL}/")
|
||||
print(f"服务状态: {response.json()}\n")
|
||||
|
||||
# 警告
|
||||
print("⚠️ 警告: 此脚本将删除测试Agents!")
|
||||
confirm = input("确认继续? (yes/no): ")
|
||||
|
||||
if confirm.lower() == "yes":
|
||||
test_delete_agent()
|
||||
else:
|
||||
print("已取消")
|
||||
|
||||
except requests.exceptions.ConnectionError:
|
||||
print("❌ 无法连接到服务。请确保服务正在运行: python app.py")
|
||||
sys.exit(1)
|
||||
@@ -0,0 +1,145 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Test script to verify environment variable passing to agent pods
|
||||
"""
|
||||
import requests
|
||||
import json
|
||||
import sys
|
||||
|
||||
API_URL = "http://localhost:8000"
|
||||
|
||||
def test_create_mysql_agent_with_env():
|
||||
"""Test creating a MySQL agent with environment variables"""
|
||||
|
||||
payload = {
|
||||
"name": "test-mysql-agent",
|
||||
"template": "mysql_agent",
|
||||
"env": {
|
||||
"MYSQL_HOST": "test-mysql-server.mysql.database.azure.com",
|
||||
"MYSQL_PORT": "3306",
|
||||
"MYSQL_USER": "testuser",
|
||||
"MYSQL_PASSWORD": "testpass",
|
||||
"MYSQL_DATABASE": "testdb",
|
||||
"OPENAI_API_KEY": "sk-test-key"
|
||||
},
|
||||
"config": {
|
||||
"cpu_request": "100m",
|
||||
"memory_request": "128Mi"
|
||||
}
|
||||
}
|
||||
|
||||
print("Creating MySQL agent with environment variables...")
|
||||
print(f"Payload: {json.dumps(payload, indent=2)}")
|
||||
|
||||
try:
|
||||
response = requests.post(f"{API_URL}/agents", json=payload)
|
||||
print(f"\nStatus Code: {response.status_code}")
|
||||
print(f"Response: {json.dumps(response.json(), indent=2)}")
|
||||
|
||||
if response.status_code == 200:
|
||||
print("\n✅ Agent created successfully!")
|
||||
return True
|
||||
else:
|
||||
print("\n❌ Agent creation failed!")
|
||||
return False
|
||||
|
||||
except Exception as e:
|
||||
print(f"\n❌ Error: {str(e)}")
|
||||
return False
|
||||
|
||||
def test_get_agent_status():
|
||||
"""Test getting agent status"""
|
||||
|
||||
agent_name = "test-mysql-agent"
|
||||
|
||||
print(f"\nGetting status for agent: {agent_name}")
|
||||
|
||||
try:
|
||||
response = requests.get(f"{API_URL}/agents/{agent_name}/status")
|
||||
print(f"Status Code: {response.status_code}")
|
||||
print(f"Response: {json.dumps(response.json(), indent=2)}")
|
||||
|
||||
if response.status_code == 200:
|
||||
print("\n✅ Status retrieved successfully!")
|
||||
return True
|
||||
else:
|
||||
print("\n❌ Failed to get status!")
|
||||
return False
|
||||
|
||||
except Exception as e:
|
||||
print(f"\n❌ Error: {str(e)}")
|
||||
return False
|
||||
|
||||
def test_get_template_info():
|
||||
"""Test getting template information"""
|
||||
|
||||
template = "mysql_agent"
|
||||
|
||||
print(f"\nGetting template info for: {template}")
|
||||
|
||||
try:
|
||||
response = requests.get(f"{API_URL}/templates/{template}")
|
||||
print(f"Status Code: {response.status_code}")
|
||||
print(f"Response: {json.dumps(response.json(), indent=2)}")
|
||||
|
||||
if response.status_code == 200:
|
||||
print("\n✅ Template info retrieved successfully!")
|
||||
return True
|
||||
else:
|
||||
print("\n❌ Failed to get template info!")
|
||||
return False
|
||||
|
||||
except Exception as e:
|
||||
print(f"\n❌ Error: {str(e)}")
|
||||
return False
|
||||
|
||||
def cleanup_test_agent():
|
||||
"""Cleanup test agent"""
|
||||
|
||||
agent_name = "test-mysql-agent"
|
||||
|
||||
print(f"\nCleaning up test agent: {agent_name}")
|
||||
|
||||
try:
|
||||
response = requests.delete(f"{API_URL}/agents/{agent_name}")
|
||||
print(f"Status Code: {response.status_code}")
|
||||
|
||||
if response.status_code == 200:
|
||||
print("✅ Test agent cleaned up!")
|
||||
else:
|
||||
print("⚠️ Cleanup may have failed (agent might not exist)")
|
||||
|
||||
except Exception as e:
|
||||
print(f"⚠️ Cleanup error: {str(e)}")
|
||||
|
||||
if __name__ == "__main__":
|
||||
print("=" * 60)
|
||||
print("Testing Agent Manager API - Environment Variables")
|
||||
print("=" * 60)
|
||||
|
||||
# Test 1: Get template info
|
||||
test_get_template_info()
|
||||
|
||||
print("\n" + "=" * 60)
|
||||
|
||||
# Test 2: Create agent with env vars
|
||||
test_create_mysql_agent_with_env()
|
||||
|
||||
print("\n" + "=" * 60)
|
||||
|
||||
# Wait a bit for pod to be created
|
||||
import time
|
||||
print("\nWaiting 5 seconds for pod creation...")
|
||||
time.sleep(5)
|
||||
|
||||
# Test 3: Get agent status
|
||||
test_get_agent_status()
|
||||
|
||||
print("\n" + "=" * 60)
|
||||
|
||||
# Cleanup
|
||||
cleanup_test_agent()
|
||||
|
||||
print("\n" + "=" * 60)
|
||||
print("Tests completed!")
|
||||
print("=" * 60)
|
||||
@@ -0,0 +1,57 @@
|
||||
"""
|
||||
测试脚本 - 获取Agent资源使用情况
|
||||
"""
|
||||
import requests
|
||||
import json
|
||||
import sys
|
||||
|
||||
# 配置
|
||||
BASE_URL = "http://localhost:8000"
|
||||
|
||||
|
||||
def test_get_agent_metrics():
|
||||
"""测试获取Agent资源信息"""
|
||||
print("=" * 50)
|
||||
print("测试: 获取AI Agent资源使用情况")
|
||||
print("=" * 50)
|
||||
|
||||
# 测试的Agent名称列表
|
||||
agent_names = ["test-echo-1", "test-chat-1", "test-code-1"]
|
||||
|
||||
for agent_name in agent_names:
|
||||
print(f"\n查询Agent资源: {agent_name}")
|
||||
|
||||
try:
|
||||
response = requests.get(f"{BASE_URL}/agents/{agent_name}/metrics")
|
||||
|
||||
print(f"状态码: {response.status_code}")
|
||||
|
||||
if response.status_code == 200:
|
||||
result = response.json()
|
||||
print("✅ 查询成功!")
|
||||
print(f"响应: {json.dumps(result, indent=2, ensure_ascii=False)}")
|
||||
elif response.status_code == 404:
|
||||
print(f"⚠️ Agent不存在")
|
||||
print(f"错误: {response.text}")
|
||||
else:
|
||||
print(f"❌ 查询失败!")
|
||||
print(f"错误: {response.text}")
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ 请求失败: {str(e)}")
|
||||
|
||||
print("-" * 50)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
# 先检查服务是否可用
|
||||
print("检查服务状态...")
|
||||
response = requests.get(f"{BASE_URL}/")
|
||||
print(f"服务状态: {response.json()}\n")
|
||||
|
||||
test_get_agent_metrics()
|
||||
|
||||
except requests.exceptions.ConnectionError:
|
||||
print("❌ 无法连接到服务。请确保服务正在运行: python app.py")
|
||||
sys.exit(1)
|
||||
@@ -0,0 +1,57 @@
|
||||
"""
|
||||
测试脚本 - 获取Agent状态
|
||||
"""
|
||||
import requests
|
||||
import json
|
||||
import sys
|
||||
|
||||
# 配置
|
||||
BASE_URL = "http://localhost:8000"
|
||||
|
||||
|
||||
def test_get_agent_status():
|
||||
"""测试获取Agent状态"""
|
||||
print("=" * 50)
|
||||
print("测试: 获取AI Agent状态")
|
||||
print("=" * 50)
|
||||
|
||||
# 测试的Agent名称列表
|
||||
agent_names = ["test-echo-1", "test-chat-1", "test-code-1"]
|
||||
|
||||
for agent_name in agent_names:
|
||||
print(f"\n查询Agent: {agent_name}")
|
||||
|
||||
try:
|
||||
response = requests.get(f"{BASE_URL}/agents/{agent_name}/status")
|
||||
|
||||
print(f"状态码: {response.status_code}")
|
||||
|
||||
if response.status_code == 200:
|
||||
result = response.json()
|
||||
print("✅ 查询成功!")
|
||||
print(f"响应: {json.dumps(result, indent=2, ensure_ascii=False)}")
|
||||
elif response.status_code == 404:
|
||||
print(f"⚠️ Agent不存在")
|
||||
print(f"错误: {response.text}")
|
||||
else:
|
||||
print(f"❌ 查询失败!")
|
||||
print(f"错误: {response.text}")
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ 请求失败: {str(e)}")
|
||||
|
||||
print("-" * 50)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
# 先检查服务是否可用
|
||||
print("检查服务状态...")
|
||||
response = requests.get(f"{BASE_URL}/")
|
||||
print(f"服务状态: {response.json()}\n")
|
||||
|
||||
test_get_agent_status()
|
||||
|
||||
except requests.exceptions.ConnectionError:
|
||||
print("❌ 无法连接到服务。请确保服务正在运行: python app.py")
|
||||
sys.exit(1)
|
||||
@@ -0,0 +1,72 @@
|
||||
"""
|
||||
测试脚本 - 列出所有Agents
|
||||
"""
|
||||
import requests
|
||||
import json
|
||||
import sys
|
||||
|
||||
# 配置
|
||||
BASE_URL = "http://localhost:8000"
|
||||
|
||||
|
||||
def test_list_agents():
|
||||
"""测试列出所有Agent"""
|
||||
print("=" * 50)
|
||||
print("测试: 列出所有AI Agents")
|
||||
print("=" * 50)
|
||||
|
||||
# 测试1: 列出所有agents
|
||||
print("\n测试1: 列出所有Agents")
|
||||
try:
|
||||
response = requests.get(f"{BASE_URL}/agents")
|
||||
print(f"状态码: {response.status_code}")
|
||||
|
||||
if response.status_code == 200:
|
||||
result = response.json()
|
||||
print("✅ 查询成功!")
|
||||
print(f"找到 {result['count']} 个Agents")
|
||||
print(f"响应: {json.dumps(result, indent=2, ensure_ascii=False)}")
|
||||
else:
|
||||
print(f"❌ 查询失败!")
|
||||
print(f"错误: {response.text}")
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ 请求失败: {str(e)}")
|
||||
|
||||
print("-" * 50)
|
||||
|
||||
# 测试2: 按模板类型过滤
|
||||
templates = ["echo_agent", "chat_agent", "code_agent"]
|
||||
for template in templates:
|
||||
print(f"\n测试2: 列出模板为 {template} 的Agents")
|
||||
try:
|
||||
response = requests.get(f"{BASE_URL}/agents?template={template}")
|
||||
print(f"状态码: {response.status_code}")
|
||||
|
||||
if response.status_code == 200:
|
||||
result = response.json()
|
||||
print("✅ 查询成功!")
|
||||
print(f"找到 {result['count']} 个 {template} Agents")
|
||||
print(f"响应: {json.dumps(result, indent=2, ensure_ascii=False)}")
|
||||
else:
|
||||
print(f"❌ 查询失败!")
|
||||
print(f"错误: {response.text}")
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ 请求失败: {str(e)}")
|
||||
|
||||
print("-" * 50)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
# 先检查服务是否可用
|
||||
print("检查服务状态...")
|
||||
response = requests.get(f"{BASE_URL}/")
|
||||
print(f"服务状态: {response.json()}\n")
|
||||
|
||||
test_list_agents()
|
||||
|
||||
except requests.exceptions.ConnectionError:
|
||||
print("❌ 无法连接到服务。请确保服务正在运行: python app.py")
|
||||
sys.exit(1)
|
||||
@@ -0,0 +1,328 @@
|
||||
#!/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()
|
||||
Reference in New Issue
Block a user