life-echo/api/tests/test_sms_verification.py

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