406 lines
11 KiB
Python
Executable File
406 lines
11 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""
|
||
短信验证码功能测试脚本
|
||
|
||
测试用例：
|
||
1. 发送验证码（成功/频率限制）
|
||
2. 验证码验证（成功/过期/错误）
|
||
3. 验证码注册流程
|
||
4. 验证码登录流程
|
||
5. 密码重置流程
|
||
6. 修改密码（已登录）
|
||
7. 修改手机号
|
||
8. 登出所有设备
|
||
"""
|
||
|
||
import requests
|
||
import time
|
||
import json
|
||
from typing import Optional, Dict, Any
|
||
|
||
# 配置
|
||
BASE_URL = "http://localhost:8000"
|
||
API_PREFIX = "/api"
|
||
|
||
# 测试用户数据
|
||
TEST_PHONE = "13800138000"
|
||
TEST_PASSWORD = "test123456"
|
||
TEST_NICKNAME = "测试用户"
|
||
TEST_EMAIL = "test@example.com"
|
||
NEW_PHONE = "13900139000"
|
||
NEW_PASSWORD = "newpass123456"
|
||
|
||
# 全局变量
|
||
access_token: Optional[str] = None
|
||
refresh_token: Optional[str] = None
|
||
|
||
|
||
class Colors:
|
||
"""终端颜色"""
|
||
|
||
HEADER = "\033[95m"
|
||
OKBLUE = "\033[94m"
|
||
OKCYAN = "\033[96m"
|
||
OKGREEN = "\033[92m"
|
||
WARNING = "\033[93m"
|
||
FAIL = "\033[91m"
|
||
ENDC = "\033[0m"
|
||
BOLD = "\033[1m"
|
||
UNDERLINE = "\033[4m"
|
||
|
||
|
||
def print_header(text: str):
|
||
"""打印测试标题"""
|
||
print(f"\n{Colors.HEADER}{Colors.BOLD}{'=' * 60}{Colors.ENDC}")
|
||
print(f"{Colors.HEADER}{Colors.BOLD}{text}{Colors.ENDC}")
|
||
print(f"{Colors.HEADER}{Colors.BOLD}{'=' * 60}{Colors.ENDC}\n")
|
||
|
||
|
||
def print_success(text: str):
|
||
"""打印成功信息"""
|
||
print(f"{Colors.OKGREEN}✓ {text}{Colors.ENDC}")
|
||
|
||
|
||
def print_error(text: str):
|
||
"""打印错误信息"""
|
||
print(f"{Colors.FAIL}✗ {text}{Colors.ENDC}")
|
||
|
||
|
||
def print_info(text: str):
|
||
"""打印信息"""
|
||
print(f"{Colors.OKCYAN}ℹ {text}{Colors.ENDC}")
|
||
|
||
|
||
def print_warning(text: str):
|
||
"""打印警告"""
|
||
print(f"{Colors.WARNING}⚠ {text}{Colors.ENDC}")
|
||
|
||
|
||
def make_request(
|
||
method: str,
|
||
endpoint: str,
|
||
data: Optional[Dict[str, Any]] = None,
|
||
headers: Optional[Dict[str, str]] = None,
|
||
expected_status: int = 200,
|
||
) -> Optional[Dict[str, Any]]:
|
||
"""发送HTTP请求"""
|
||
url = f"{BASE_URL}{API_PREFIX}{endpoint}"
|
||
|
||
try:
|
||
if method.upper() == "GET":
|
||
response = requests.get(url, headers=headers)
|
||
elif method.upper() == "POST":
|
||
response = requests.post(url, json=data, headers=headers)
|
||
elif method.upper() == "PUT":
|
||
response = requests.put(url, json=data, headers=headers)
|
||
elif method.upper() == "DELETE":
|
||
response = requests.delete(url, headers=headers)
|
||
else:
|
||
print_error(f"不支持的HTTP方法: {method}")
|
||
return None
|
||
|
||
print_info(f"{method.upper()} {endpoint} - Status: {response.status_code}")
|
||
|
||
if response.status_code == expected_status:
|
||
print_success(f"请求成功 (状态码: {response.status_code})")
|
||
try:
|
||
return response.json()
|
||
except:
|
||
return {"status": "success"}
|
||
else:
|
||
print_error(
|
||
f"请求失败 (期望: {expected_status}, 实际: {response.status_code})"
|
||
)
|
||
try:
|
||
error_data = response.json()
|
||
print_error(
|
||
f"错误信息: {json.dumps(error_data, ensure_ascii=False, indent=2)}"
|
||
)
|
||
except:
|
||
print_error(f"响应内容: {response.text}")
|
||
return None
|
||
|
||
except requests.exceptions.ConnectionError:
|
||
print_error(f"连接失败: 无法连接到 {BASE_URL}")
|
||
print_warning("请确保后端服务正在运行")
|
||
return None
|
||
except Exception as e:
|
||
print_error(f"请求异常: {str(e)}")
|
||
return None
|
||
|
||
|
||
def test_send_verification_code(phone: str, purpose: str) -> bool:
|
||
"""测试发送验证码"""
|
||
print_header(f"测试发送验证码 - {purpose}")
|
||
|
||
data = {"phone": phone, "purpose": purpose}
|
||
|
||
result = make_request("POST", "/auth/sms/send", data=data)
|
||
|
||
if result:
|
||
print_success(f"验证码已发送: {result.get('message', '')}")
|
||
print_info(f"有效期: {result.get('expires_in', 0)} 秒")
|
||
return True
|
||
|
||
return False
|
||
|
||
|
||
def test_rate_limit(phone: str) -> bool:
|
||
"""测试频率限制"""
|
||
print_header("测试频率限制")
|
||
|
||
# 第一次发送应该成功
|
||
if not test_send_verification_code(phone, "register"):
|
||
return False
|
||
|
||
print_info("等待1秒后再次发送...")
|
||
time.sleep(1)
|
||
|
||
# 第二次发送应该被限制
|
||
data = {"phone": phone, "purpose": "register"}
|
||
|
||
result = make_request("POST", "/auth/sms/send", data=data, expected_status=429)
|
||
|
||
if result is None:
|
||
print_success("频率限制生效")
|
||
return True
|
||
else:
|
||
print_error("频率限制未生效")
|
||
return False
|
||
|
||
|
||
def test_register_with_sms(phone: str, code: str) -> bool:
|
||
"""测试验证码注册"""
|
||
print_header("测试验证码注册")
|
||
|
||
data = {
|
||
"phone": phone,
|
||
"code": code,
|
||
"password": TEST_PASSWORD,
|
||
"nickname": TEST_NICKNAME,
|
||
"email": TEST_EMAIL,
|
||
}
|
||
|
||
result = make_request("POST", "/auth/register/sms", data=data, expected_status=201)
|
||
|
||
if result:
|
||
global access_token, refresh_token
|
||
access_token = result.get("access_token")
|
||
refresh_token = result.get("refresh_token")
|
||
|
||
print_success("注册成功")
|
||
print_info(f"Access Token: {access_token[:20]}...")
|
||
print_info(f"Refresh Token: {refresh_token[:20]}...")
|
||
return True
|
||
|
||
return False
|
||
|
||
|
||
def test_login_with_sms(phone: str, code: str) -> bool:
|
||
"""测试验证码登录"""
|
||
print_header("测试验证码登录")
|
||
|
||
data = {"phone": phone, "code": code}
|
||
|
||
result = make_request("POST", "/auth/login/sms", data=data)
|
||
|
||
if result:
|
||
global access_token, refresh_token
|
||
access_token = result.get("access_token")
|
||
refresh_token = result.get("refresh_token")
|
||
|
||
print_success("登录成功")
|
||
print_info(f"Access Token: {access_token[:20]}...")
|
||
return True
|
||
|
||
return False
|
||
|
||
|
||
def test_reset_password(phone: str, code: str, new_password: str) -> bool:
|
||
"""测试重置密码"""
|
||
print_header("测试重置密码")
|
||
|
||
data = {"phone": phone, "code": code, "new_password": new_password}
|
||
|
||
result = make_request("POST", "/auth/password/reset", data=data)
|
||
|
||
if result:
|
||
print_success(f"密码重置成功: {result.get('message', '')}")
|
||
return True
|
||
|
||
return False
|
||
|
||
|
||
def test_change_password(old_password: str, new_password: str) -> bool:
|
||
"""测试修改密码（已登录）"""
|
||
print_header("测试修改密码")
|
||
|
||
if not access_token:
|
||
print_error("未登录，无法测试")
|
||
return False
|
||
|
||
data = {"old_password": old_password, "new_password": new_password}
|
||
|
||
headers = {"Authorization": f"Bearer {access_token}"}
|
||
|
||
result = make_request("POST", "/auth/password/change", data=data, headers=headers)
|
||
|
||
if result:
|
||
print_success(f"密码修改成功: {result.get('message', '')}")
|
||
return True
|
||
|
||
return False
|
||
|
||
|
||
def test_change_phone(new_phone: str, code: str) -> bool:
|
||
"""测试修改手机号"""
|
||
print_header("测试修改手机号")
|
||
|
||
if not access_token:
|
||
print_error("未登录，无法测试")
|
||
return False
|
||
|
||
data = {"new_phone": new_phone, "code": code}
|
||
|
||
headers = {"Authorization": f"Bearer {access_token}"}
|
||
|
||
result = make_request("POST", "/auth/phone/change", data=data, headers=headers)
|
||
|
||
if result:
|
||
print_success(f"手机号修改成功")
|
||
print_info(f"新手机号: {result.get('phone', '')}")
|
||
return True
|
||
|
||
return False
|
||
|
||
|
||
def test_logout_all() -> bool:
|
||
"""测试登出所有设备"""
|
||
print_header("测试登出所有设备")
|
||
|
||
if not access_token:
|
||
print_error("未登录，无法测试")
|
||
return False
|
||
|
||
headers = {"Authorization": f"Bearer {access_token}"}
|
||
|
||
result = make_request("POST", "/auth/logout/all", headers=headers)
|
||
|
||
if result:
|
||
print_success(f"登出成功: {result.get('message', '')}")
|
||
return True
|
||
|
||
return False
|
||
|
||
|
||
def test_get_current_user() -> bool:
|
||
"""测试获取当前用户信息"""
|
||
print_header("测试获取当前用户信息")
|
||
|
||
if not access_token:
|
||
print_error("未登录，无法测试")
|
||
return False
|
||
|
||
headers = {"Authorization": f"Bearer {access_token}"}
|
||
|
||
result = make_request("GET", "/auth/me", headers=headers)
|
||
|
||
if result:
|
||
print_success("获取用户信息成功")
|
||
print_info(f"用户信息: {json.dumps(result, ensure_ascii=False, indent=2)}")
|
||
return True
|
||
|
||
return False
|
||
|
||
|
||
def interactive_test():
|
||
"""交互式测试"""
|
||
print_header("短信验证码功能交互式测试")
|
||
print_info("此模式需要您手动输入收到的验证码")
|
||
print_warning("请确保已配置腾讯云短信服务")
|
||
|
||
phone = input(f"\n请输入测试手机号 (默认: {TEST_PHONE}): ").strip() or TEST_PHONE
|
||
|
||
# 1. 测试发送验证码
|
||
if not test_send_verification_code(phone, "register"):
|
||
print_error("发送验证码失败，测试终止")
|
||
return
|
||
|
||
code = input("\n请输入收到的验证码: ").strip()
|
||
|
||
if not code or len(code) != 6:
|
||
print_error("验证码格式错误")
|
||
return
|
||
|
||
# 2. 测试注册
|
||
if test_register_with_sms(phone, code):
|
||
print_success("注册测试通过")
|
||
|
||
# 3. 测试获取用户信息
|
||
test_get_current_user()
|
||
|
||
# 4. 测试修改密码
|
||
if input("\n是否测试修改密码? (y/n): ").lower() == "y":
|
||
test_change_password(TEST_PASSWORD, NEW_PASSWORD)
|
||
|
||
# 5. 测试修改手机号
|
||
if input("\n是否测试修改手机号? (y/n): ").lower() == "y":
|
||
new_phone = (
|
||
input(f"请输入新手机号 (默认: {NEW_PHONE}): ").strip() or NEW_PHONE
|
||
)
|
||
|
||
if test_send_verification_code(new_phone, "change_phone"):
|
||
code = input("请输入收到的验证码: ").strip()
|
||
test_change_phone(new_phone, code)
|
||
|
||
# 6. 测试登出所有设备
|
||
if input("\n是否测试登出所有设备? (y/n): ").lower() == "y":
|
||
test_logout_all()
|
||
|
||
|
||
def automated_test():
|
||
"""自动化测试（需要mock验证码）"""
|
||
print_header("短信验证码功能自动化测试")
|
||
print_warning("此模式需要后端支持测试验证码（如：123456）")
|
||
|
||
# 测试发送验证码
|
||
test_send_verification_code(TEST_PHONE, "register")
|
||
|
||
# 等待一段时间
|
||
print_info("等待60秒以测试频率限制...")
|
||
time.sleep(60)
|
||
|
||
# 测试频率限制
|
||
test_rate_limit(TEST_PHONE)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
print(f"{Colors.BOLD}{Colors.OKBLUE}")
|
||
print("=" * 60)
|
||
print("短信验证码功能测试脚本")
|
||
print("=" * 60)
|
||
print(f"{Colors.ENDC}")
|
||
|
||
print("\n请选择测试模式:")
|
||
print("1. 交互式测试（需要真实短信验证码）")
|
||
print("2. 自动化测试（需要测试验证码支持）")
|
||
print("3. 仅测试API连接")
|
||
|
||
choice = input("\n请输入选项 (1/2/3): ").strip()
|
||
|
||
if choice == "1":
|
||
interactive_test()
|
||
elif choice == "2":
|
||
automated_test()
|
||
elif choice == "3":
|
||
print_header("测试API连接")
|
||
result = make_request("GET", "/health", expected_status=200)
|
||
if result:
|
||
print_success("API连接正常")
|
||
else:
|
||
print_error("API连接失败")
|
||
else:
|
||
print_error("无效的选项")
|
||
|
||
print(f"\n{Colors.BOLD}{Colors.OKBLUE}测试完成{Colors.ENDC}\n")
|