forked from moyin/whale-town-front
test：完善API测试框架
- 添加Godot内置API测试脚本 - 实现Python API客户端测试套件 - 添加快速测试和完整测试脚本 - 支持跨平台测试运行(Windows/Linux) - 更新测试文档和使用指南
This commit is contained in:
427
tests/api/api_client_test.py
Normal file
427
tests/api/api_client_test.py
Normal file
@@ -0,0 +1,427 @@
|
||||
#!/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()
|
||||
Reference in New Issue
Block a user