#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ API客户端测试脚本 用于在没有Godot引擎的情况下测试后端API接口 使用方法: python api_client_test.py """ import requests import json import time import sys from typing import Dict, Any, Optional from dataclasses import dataclass from datetime import datetime # API配置 API_BASE_URL = "https://whaletownend.xinghangee.icu" DEFAULT_TIMEOUT = 30 @dataclass class TestResult: """测试结果数据类""" success: bool message: str response_code: int response_data: Dict[str, Any] error_info: Optional[Dict[str, Any]] = None execution_time: float = 0.0 class APIClient: """API客户端类""" def __init__(self, base_url: str = API_BASE_URL): self.base_url = base_url self.session = requests.Session() self.session.headers.update({ 'Content-Type': 'application/json', 'User-Agent': 'WhaleTown-API-Test/1.0' }) def _make_request(self, method: str, endpoint: str, data: Optional[Dict] = None, timeout: int = DEFAULT_TIMEOUT) -> TestResult: """发送HTTP请求""" url = f"{self.base_url}{endpoint}" start_time = time.time() try: print(f"🚀 发送 {method} 请求到: {url}") if data: print(f"📦 请求数据: {json.dumps(data, ensure_ascii=False, indent=2)}") response = self.session.request( method=method, url=url, json=data if data else None, timeout=timeout ) execution_time = time.time() - start_time print(f"📊 响应状态码: {response.status_code}") print(f"⏱️ 执行时间: {execution_time:.3f}s") # 尝试解析JSON响应 try: response_data = response.json() print(f"📄 响应数据: {json.dumps(response_data, ensure_ascii=False, indent=2)}") except json.JSONDecodeError: response_data = {"raw_response": response.text} print(f"📄 原始响应: {response.text}") # 判断请求是否成功 is_success = False if 200 <= response.status_code < 300: # HTTP成功状态码 business_success = response_data.get("success", True) if business_success: is_success = True elif response.status_code == 206 and response_data.get("error_code") == "TEST_MODE_ONLY": # 测试模式也算成功 is_success = True return TestResult( success=is_success, message=response_data.get("message", ""), response_code=response.status_code, response_data=response_data, execution_time=execution_time ) except requests.exceptions.Timeout: execution_time = time.time() - start_time return TestResult( success=False, message="请求超时", response_code=0, response_data={}, error_info={"error_type": "TIMEOUT"}, execution_time=execution_time ) except requests.exceptions.ConnectionError: execution_time = time.time() - start_time return TestResult( success=False, message="网络连接失败", response_code=0, response_data={}, error_info={"error_type": "CONNECTION_ERROR"}, execution_time=execution_time ) except Exception as e: execution_time = time.time() - start_time return TestResult( success=False, message=f"请求异常: {str(e)}", response_code=0, response_data={}, error_info={"error_type": "UNKNOWN_ERROR", "details": str(e)}, execution_time=execution_time ) def get_app_status(self) -> TestResult: """获取应用状态""" return self._make_request("GET", "/") def login(self, identifier: str, password: str) -> TestResult: """用户登录""" data = { "identifier": identifier, "password": password } return self._make_request("POST", "/auth/login", data) def register(self, username: str, password: str, nickname: str, email: str = "", email_verification_code: str = "") -> TestResult: """用户注册""" data = { "username": username, "password": password, "nickname": nickname } if email: data["email"] = email if email_verification_code: data["email_verification_code"] = email_verification_code return self._make_request("POST", "/auth/register", data) def send_email_verification(self, email: str) -> TestResult: """发送邮箱验证码""" data = {"email": email} return self._make_request("POST", "/auth/send-email-verification", data) def verify_email(self, email: str, verification_code: str) -> TestResult: """验证邮箱验证码""" data = { "email": email, "verification_code": verification_code } return self._make_request("POST", "/auth/verify-email", data) def send_login_verification_code(self, identifier: str) -> TestResult: """发送登录验证码""" data = {"identifier": identifier} return self._make_request("POST", "/auth/send-login-verification-code", data) def verification_code_login(self, identifier: str, verification_code: str) -> TestResult: """验证码登录""" data = { "identifier": identifier, "verification_code": verification_code } return self._make_request("POST", "/auth/verification-code-login", data) def forgot_password(self, identifier: str) -> TestResult: """忘记密码 - 发送重置验证码""" data = {"identifier": identifier} return self._make_request("POST", "/auth/forgot-password", data) def reset_password(self, identifier: str, verification_code: str, new_password: str) -> TestResult: """重置密码""" data = { "identifier": identifier, "verification_code": verification_code, "new_password": new_password } return self._make_request("POST", "/auth/reset-password", data) class APITester: """API测试器""" def __init__(self): self.client = APIClient() self.test_results = [] self.test_email = "test@example.com" self.test_username = "testuser" self.test_password = "password123" self.verification_code = None def print_header(self, title: str): """打印测试标题""" print("\n" + "="*60) print(f"🧪 {title}") print("="*60) def print_result(self, test_name: str, result: TestResult): """打印测试结果""" status = "✅ 成功" if result.success else "❌ 失败" print(f"\n📋 测试: {test_name}") print(f"🎯 结果: {status}") print(f"💬 消息: {result.message}") print(f"📊 状态码: {result.response_code}") print(f"⏱️ 耗时: {result.execution_time:.3f}s") # 检查特殊情况 if result.response_code == 206: print("🧪 检测到测试模式响应") elif result.response_code == 409: print("⚠️ 检测到资源冲突") elif result.response_code == 429: print("⏰ 检测到频率限制") # 提取验证码（如果有） if result.response_data.get("data", {}).get("verification_code"): self.verification_code = result.response_data["data"]["verification_code"] print(f"🔑 获取到验证码: {self.verification_code}") self.test_results.append((test_name, result)) print("-" * 40) def test_app_status(self): """测试应用状态""" self.print_header("应用状态测试") result = self.client.get_app_status() self.print_result("获取应用状态", result) def test_email_verification_flow(self): """测试邮箱验证流程""" self.print_header("邮箱验证流程测试") # 1. 发送邮箱验证码 result = self.client.send_email_verification(self.test_email) self.print_result("发送邮箱验证码", result) if self.verification_code: # 2. 验证邮箱验证码 result = self.client.verify_email(self.test_email, self.verification_code) self.print_result("验证邮箱验证码", result) def test_registration_flow(self): """测试注册流程""" self.print_header("用户注册流程测试") # 使用之前获取的验证码进行注册 if self.verification_code: result = self.client.register( username=self.test_username, password=self.test_password, nickname="测试用户", email=self.test_email, email_verification_code=self.verification_code ) self.print_result("用户注册（带邮箱验证）", result) else: # 无邮箱注册 result = self.client.register( username=f"{self.test_username}_no_email", password=self.test_password, nickname="测试用户（无邮箱）" ) self.print_result("用户注册（无邮箱）", result) def test_login_flow(self): """测试登录流程""" self.print_header("用户登录流程测试") # 1. 密码登录 result = self.client.login(self.test_username, self.test_password) self.print_result("密码登录", result) # 2. 发送登录验证码 result = self.client.send_login_verification_code(self.test_email) self.print_result("发送登录验证码", result) # 3. 验证码登录 if self.verification_code: result = self.client.verification_code_login(self.test_email, self.verification_code) self.print_result("验证码登录", result) def test_password_reset_flow(self): """测试密码重置流程""" self.print_header("密码重置流程测试") # 1. 发送重置验证码 result = self.client.forgot_password(self.test_email) self.print_result("发送重置验证码", result) # 2. 重置密码 if self.verification_code: result = self.client.reset_password( identifier=self.test_email, verification_code=self.verification_code, new_password="newpassword123" ) self.print_result("重置密码", result) def test_error_scenarios(self): """测试错误场景""" self.print_header("错误场景测试") # 1. 无效邮箱格式 result = self.client.send_email_verification("invalid-email") self.print_result("无效邮箱格式", result) # 2. 错误的验证码 result = self.client.verify_email(self.test_email, "000000") self.print_result("错误验证码", result) # 3. 用户名冲突 result = self.client.register( username=self.test_username, # 重复用户名 password=self.test_password, nickname="重复用户" ) self.print_result("用户名冲突", result) # 4. 错误的登录凭据 result = self.client.login("nonexistent", "wrongpassword") self.print_result("错误登录凭据", result) def test_rate_limiting(self): """测试频率限制""" self.print_header("频率限制测试") print("🔄 快速发送多个验证码请求以触发频率限制...") for i in range(3): result = self.client.send_email_verification(f"test{i}@example.com") self.print_result(f"验证码请求 #{i+1}", result) if result.response_code == 429: print("✅ 成功触发频率限制") break time.sleep(0.5) # 短暂延迟 def run_all_tests(self): """运行所有测试""" print("🎯 开始API接口测试") print(f"🌐 测试服务器: {API_BASE_URL}") print(f"📧 测试邮箱: {self.test_email}") print(f"👤 测试用户名: {self.test_username}") try: # 基础测试 self.test_app_status() # 邮箱验证流程 self.test_email_verification_flow() # 注册流程 self.test_registration_flow() # 登录流程 self.test_login_flow() # 密码重置流程 self.test_password_reset_flow() # 错误场景测试 self.test_error_scenarios() # 频率限制测试 self.test_rate_limiting() except KeyboardInterrupt: print("\n⚠️ 测试被用户中断") except Exception as e: print(f"\n❌ 测试过程中发生异常: {e}") finally: self.print_summary() def print_summary(self): """打印测试总结""" self.print_header("测试总结") total_tests = len(self.test_results) successful_tests = sum(1 for _, result in self.test_results if result.success) failed_tests = total_tests - successful_tests print(f"📊 总测试数: {total_tests}") print(f"✅ 成功: {successful_tests}") print(f"❌ 失败: {failed_tests}") print(f"📈 成功率: {(successful_tests/total_tests*100):.1f}%" if total_tests > 0 else "0%") if failed_tests > 0: print("\n❌ 失败的测试:") for test_name, result in self.test_results: if not result.success: print(f" • {test_name}: {result.message} (状态码: {result.response_code})") print(f"\n⏱️ 总执行时间: {sum(result.execution_time for _, result in self.test_results):.3f}s") print("🎉 测试完成!") def main(): """主函数""" print("🐋 WhaleTown API 接口测试工具") print("=" * 60) # 检查网络连接 try: response = requests.get(API_BASE_URL, timeout=5) print(f"✅ 服务器连接正常 (状态码: {response.status_code})") except Exception as e: print(f"❌ 无法连接到服务器: {e}") print("请检查网络连接或服务器地址") sys.exit(1) # 运行测试 tester = APITester() tester.run_all_tests() if __name__ == "__main__": main()