#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ API测试脚本 - 测试端口8080上的所有端点 """ import requests import json import time from typing import Dict, List class CloudDriveAPITester: def __init__(self, base_url: str = "http://localhost:8080"): self.base_url = base_url self.session = requests.Session() self.session.headers.update({ 'Content-Type': 'application/json', 'Accept': 'application/json' }) def test_endpoint(self, method: str, endpoint: str, data: Dict = None, description: str = "") -> Dict: """测试单个端点""" url = f"{self.base_url}{endpoint}" try: print(f"📍 {description or f'{method.upper()} {endpoint}'}") if method.upper() == 'GET': response = self.session.get(url, timeout=10) elif method.upper() == 'POST': response = self.session.post(url, json=data, timeout=10) elif method.upper() == 'PUT': response = self.session.put(url, json=data, timeout=10) elif method.upper() == 'DELETE': response = self.session.delete(url, timeout=10) else: return {"status": "error", "message": f"不支持的HTTP方法: {method}"} if response.status_code == 200: try: data = response.json() print(f" ✅ 成功 (200): {json.dumps(data, ensure_ascii=False, indent=2)[:200]}...") return { "status": "success", "code": response.status_code, "data": data } except json.JSONDecodeError: print(f" ✅ 成功 (200): {response.text[:200]}...") return { "status": "success", "code": response.status_code, "text": response.text } else: print(f" ❌ 失败 ({response.status_code}): {response.text[:200]}...") return { "status": "failed", "code": response.status_code, "error": response.text } except requests.exceptions.ConnectionError: print(f" ❌ 连接失败 - 无法连接到 {url}") return {"status": "connection_failed", "error": "Connection failed"} except requests.exceptions.Timeout: print(f" ❌ 请求超时") return {"status": "timeout", "error": "Request timeout"} except Exception as e: print(f" ❌ 未知错误: {e}") return {"status": "error", "error": str(e)} def test_basic_endpoints(self) -> Dict: """测试基础端点""" print("🔍 测试基础端点") print("=" * 50) results = {} # 测试根路径 results["root"] = self.test_endpoint("GET", "/", description="根路径") # 测试健康检查端点 results["health"] = self.test_endpoint("GET", "/health", description="健康检查") results["api_health"] = self.test_endpoint("GET", "/api/v1/health", description="API健康检查") # 测试文档端点 results["docs"] = self.test_endpoint("GET", "/docs", description="API文档") results["redoc"] = self.test_endpoint("GET", "/redoc", description="ReDoc文档") results["openapi"] = self.test_endpoint("GET", "/openapi.json", description="OpenAPI规范") return results def test_auth_endpoints(self) -> Dict: """测试认证端点""" print("\n🔐 测试认证端点") print("=" * 50) results = {} # 测试注册端点 test_user = { "username": "testuser8080", "email": "test8080@example.com", "password": "Test123!@#", "confirm_password": "Test123!@#" } results["register"] = self.test_endpoint("POST", "/api/v1/auth/register", data=test_user, description="用户注册") # 测试登录端点 login_data = { "username": "testuser8080", "password": "Test123!@#" } results["login"] = self.test_endpoint("POST", "/api/v1/auth/token", data=login_data, description="用户登录") return results def test_file_endpoints(self) -> Dict: """测试文件端点""" print("\n📁 测试文件端点") print("=" * 50) results = {} # 测试文件列表 results["list_files"] = self.test_endpoint("GET", "/api/v1/files", description="获取文件列表") # 测试文件上传（模拟） results["upload_info"] = self.test_endpoint("GET", "/api/v1/files", description="文件上传信息") return results def run_all_tests(self) -> Dict: """运行所有测试""" print("🧪 开始API测试 - 端口8080") print("=" * 60) print(f"测试目标: {self.base_url}") print("=" * 60) all_results = {} # 测试基础端点 basic_results = self.test_basic_endpoints() all_results.update(basic_results) # 测试认证端点 auth_results = self.test_auth_endpoints() all_results.update(auth_results) # 测试文件端点 file_results = self.test_file_endpoints() all_results.update(file_results) # 生成测试报告 self.generate_report(all_results) return all_results def generate_report(self, results: Dict): """生成测试报告""" print("\n📊 测试报告") print("=" * 60) total_tests = len(results) success_tests = sum(1 for r in results.values() if r.get("status") == "success") failed_tests = total_tests - success_tests print(f"总测试数: {total_tests}") print(f"成功: {success_tests}") print(f"失败: {failed_tests}") print(f"成功率: {(success_tests/total_tests*100):.1f}%") if failed_tests == 0: print("\n🎉 所有测试通过！API服务运行正常") else: print(f"\n⚠️ 有 {failed_tests} 个测试失败，请检查服务状态") print("\n📋 详细结果:") for endpoint, result in results.items(): status = result.get("status", "unknown") if status == "success": print(f" ✅ {endpoint}") elif status == "connection_failed": print(f" 🔌 {endpoint} - 连接失败") elif status == "timeout": print(f" ⏰ {endpoint} - 超时") else: print(f" ❌ {endpoint} - {status}") def main(): """主函数""" import argparse parser = argparse.ArgumentParser(description="云盘API测试工具") parser.add_argument("--url", default="http://localhost:8080", help="API基础URL (默认: http://localhost:8080)") parser.add_argument("--wait", type=int, default=0, help="启动前等待时间(秒)") parser.add_argument("--basic", action="store_true", help="只测试基础端点") parser.add_argument("--auth", action="store_true", help="只测试认证端点") parser.add_argument("--files", action="store_true", help="只测试文件端点") args = parser.parse_args() if args.wait > 0: print(f"⏳ 等待 {args.wait} 秒后开始测试...") time.sleep(args.wait) tester = CloudDriveAPITester(args.url) try: if args.basic: tester.test_basic_endpoints() elif args.auth: tester.test_auth_endpoints() elif args.files: tester.test_file_endpoints() else: tester.run_all_tests() except KeyboardInterrupt: print("\n🛑 测试已中断") except Exception as e: print(f"\n❌ 测试过程中出现错误: {e}") if __name__ == "__main__": main()