#!/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")