Function test_servers

internbootcamp/utils/tool_server/cli.py:236–307

测试服务器功能

(server_url, tool_names, test_timeout=10, connectivity_only=False)

def test_servers(server_url, tool_names, test_timeout=10, connectivity_only=False):
    """测试服务器功能"""
    print(f"🧪 测试服务器: {server_url}")
    
    try:
        # 基础连通性测试
        print(f"🔍 测试Master连通性: {server_url}/health")
        response = requests.get(f"{server_url}/health", timeout=test_timeout)
        if response.status_code == 200:
            data = response.json()
            print("  ✅ Master健康检查通过")
            print(f"    - 支持工具: {data.get('tools', [])}")
            print(f"    - 注册Worker数: {data.get('registered_workers', 0)}")
            print(f"    - 活跃Worker数: {len([w for w in data.get('workers', {}).values() if w.get('status') == 'alive'])}")
            
            if not connectivity_only:
                # 详细端点测试
                print(f"\n🧪 测试工具端点...")
                
                for tool_name in tool_names:
                    print(f"\n--- 测试工具: {tool_name} ---")
                    success = True
                    
                    # 测试创建端点
                    try:
                        create_url = f"{server_url}/{tool_name}/create"
                        test_data = {"instance_id": f"test_instance_{tool_name}", "identity": {"test": True}}
                        response = requests.post(create_url, json=test_data, timeout=test_timeout)
                        if response.status_code == 200 and response.json().get("success"):
                            print(f"  ✅ 创建端点测试通过: {create_url}")
                        else:
                            print(f"  ❌ 创建端点测试失败: 状态码 {response.status_code}")
                            success = False
                    except Exception as e:
                        print(f"  ❌ 创建端点请求失败: {e}")
                        success = False

                    # 测试执行端点
                    try:
                        execute_url = f"{server_url}/{tool_name}/execute"
                        test_data = {"instance_id": f"test_instance_{tool_name}", "test_param": "value"}
                        response = requests.post(execute_url, json=test_data, timeout=test_timeout)
                        if response.status_code == 200:
                            print(f"  ✅ 执行端点测试通过: {execute_url}")
                        else:
                            print(f"  ❌ 执行端点测试失败: 状态码 {response.status_code}")
                            success = False
                    except Exception as e:
                        print(f"  ❌ 执行端点请求失败: {e}")
                        success = False

                    # 测试释放端点
                    try:
                        release_url = f"{server_url}/{tool_name}/release"
                        test_data = {"instance_id": f"test_instance_{tool_name}"}
                        response = requests.post(release_url, json=test_data, timeout=test_timeout)
                        if response.status_code == 200 and response.json().get("success"):
                            print(f"  ✅ 释放端点测试通过: {release_url}")