xiaoxia-saas/tests/unit/test_verification_code_service.py

"""验证码服务单元测试."""

from __future__ import annotations

from datetime import datetime, timedelta, timezone
from unittest.mock import MagicMock

import pytest

from packages.application.auth.verification_code_service import (
    CODE_TYPE_EMAIL_BIND,
    CODE_TYPE_PHONE_BIND,
    DAILY_LIMIT,
    DEFAULT_TTL_SECONDS,
    MAX_ATTEMPTS,
    RESEND_COOLDOWN_SECONDS,
    VerificationCodeService,
    normalize_phone,
    validate_email,
    validate_phone,
)
from packages.domain.verification_code import VerificationCode


@pytest.fixture
def mock_repo():
    return MagicMock()


@pytest.fixture
def code_service(mock_repo):
    return VerificationCodeService(mock_repo)


@pytest.fixture
def sample_code():
    code = VerificationCode.create(
        recipient="test@example.com",
        code_type=CODE_TYPE_EMAIL_BIND,
        ttl_seconds=300,
    )
    return code


class TestVerificationCodeServiceGenerate:
    """generate 方法测试"""

    def test_generate_success(self, code_service, mock_repo, sample_code):
        """生成验证码成功"""
        mock_repo.find_latest.return_value = None
        mock_repo.count_today.return_value = 0
        mock_repo.save.return_value = None

        code, error = code_service.generate("test@example.com", CODE_TYPE_EMAIL_BIND)

        assert error is None
        assert code is not None
        assert code.recipient == "test@example.com"
        assert code.code_type == CODE_TYPE_EMAIL_BIND
        mock_repo.save.assert_called_once()

    def test_generate_empty_recipient(self, code_service):
        """空接收方返回错误"""
        code, error = code_service.generate("", CODE_TYPE_EMAIL_BIND)
        assert code is None
        assert "接收方不能为空" in error

    def test_generate_invalid_type(self, code_service):
        """无效验证码类型返回错误"""
        code, error = code_service.generate("test@example.com", "invalid_type")
        assert code is None
        assert "无效的验证码类型" in error

    def test_generate_cooldown(self, code_service, mock_repo, sample_code):
        """冷却期内返回频控错误"""
        # 最新的验证码刚创建10秒前
        sample_code.created_at = datetime.now(timezone.utc) - timedelta(seconds=10)
        mock_repo.find_latest.return_value = sample_code
        mock_repo.count_today.return_value = 1

        code, error = code_service.generate("test@example.com", CODE_TYPE_EMAIL_BIND)

        assert code is None
        assert "发送太频繁" in error
        assert "秒后再试" in error

    def test_generate_daily_limit_exceeded(self, code_service, mock_repo):
        """超过每日上限返回错误"""
        mock_repo.find_latest.return_value = None  # 没有冷却期问题
        mock_repo.count_today.return_value = DAILY_LIMIT

        code, error = code_service.generate("test@example.com", CODE_TYPE_EMAIL_BIND)

        assert code is None
        assert "今日发送次数已达上限" in error

    def test_generate_recipient_stripped(self, code_service, mock_repo, sample_code):
        """recipient 会被 strip"""
        mock_repo.find_latest.return_value = None
        mock_repo.count_today.return_value = 0
        mock_repo.save.return_value = None

        code_service.generate("  test@example.com  ", CODE_TYPE_EMAIL_BIND)

        # 传给 repo 的应该是 strip 后的值
        save_call = mock_repo.save.call_args[0][0]
        assert save_call.recipient == "test@example.com"

    def test_generate_custom_code(self, code_service, mock_repo):
        """使用自定义验证码"""
        mock_repo.find_latest.return_value = None
        mock_repo.count_today.return_value = 0
        mock_repo.save.return_value = None

        code, _ = code_service.generate("test@example.com", CODE_TYPE_EMAIL_BIND, custom_code="123456")
        assert code.code == "123456"

    def test_generate_custom_ttl(self, code_service, mock_repo):
        """自定义 TTL"""
        mock_repo.find_latest.return_value = None
        mock_repo.count_today.return_value = 0
        mock_repo.save.return_value = None

        code, _ = code_service.generate("test@example.com", CODE_TYPE_EMAIL_BIND, ttl_seconds=600)
        assert code is not None


class TestVerificationCodeServiceVerify:
    """verify 方法测试"""

    def test_verify_success(self, code_service, mock_repo, sample_code):
        """验证成功"""
        mock_repo.find_latest.return_value = sample_code

        success, error = code_service.verify("test@example.com", CODE_TYPE_EMAIL_BIND, sample_code.code)

        assert success is True
        assert error is None
        assert sample_code.is_used is True

    def test_verify_wrong_code(self, code_service, mock_repo, sample_code):
        """验证码错误"""
        mock_repo.find_latest.return_value = sample_code

        success, error = code_service.verify("test@example.com", CODE_TYPE_EMAIL_BIND, "wrongcode")

        assert success is False
        assert "验证码错误" in error

    def test_verify_not_found(self, code_service, mock_repo):
        """验证码不存在"""
        mock_repo.find_latest.return_value = None

        success, error = code_service.verify("test@example.com", CODE_TYPE_EMAIL_BIND, "123456")

        assert success is False
        assert "不存在或已过期" in error

    def test_verify_expired(self, code_service, mock_repo):
        """验证码已过期"""
        expired_code = VerificationCode.create(
            recipient="test@example.com",
            code_type=CODE_TYPE_EMAIL_BIND,
            ttl_seconds=1,  # 1秒过期
        )
        # 手动设置过期时间
        expired_code.expires_at = datetime.now(timezone.utc) - timedelta(seconds=10)
        mock_repo.find_latest.return_value = expired_code

        success, error = code_service.verify("test@example.com", CODE_TYPE_EMAIL_BIND, expired_code.code)

        assert success is False
        assert "已过期" in error

    def test_verify_already_used(self, code_service, mock_repo, sample_code):
        """验证码已使用"""
        sample_code.mark_used()
        mock_repo.find_latest.return_value = sample_code

        success, error = code_service.verify("test@example.com", CODE_TYPE_EMAIL_BIND, sample_code.code)

        assert success is False
        assert "已使用" in error

    def test_verify_max_attempts_exceeded(self, code_service, mock_repo, sample_code):
        """尝试次数过多"""
        # 先把尝试次数加到超过上限
        for _ in range(MAX_ATTEMPTS + 1):
            sample_code.increment_attempts()
        mock_repo.find_latest.return_value = sample_code

        success, error = code_service.verify("test@example.com", CODE_TYPE_EMAIL_BIND, sample_code.code)

        assert success is False
        assert "验证次数过多" in error

    def test_verify_empty_params(self, code_service):
        """空参数返回错误"""
        success, error = code_service.verify("", CODE_TYPE_EMAIL_BIND, "123456")
        assert success is False
        assert "参数不完整" in error

        success, error = code_service.verify("test@example.com", CODE_TYPE_EMAIL_BIND, "")
        assert success is False
        assert "参数不完整" in error

    def test_verify_increments_attempts(self, code_service, mock_repo, sample_code):
        """验证会增加尝试次数"""
        initial_attempts = sample_code.attempts
        mock_repo.find_latest.return_value = sample_code

        code_service.verify("test@example.com", CODE_TYPE_EMAIL_BIND, "wrong")

        assert sample_code.attempts == initial_attempts + 1

    def test_verify_no_consume(self, code_service, mock_repo, sample_code):
        """consume=False 时不标记为已使用"""
        mock_repo.find_latest.return_value = sample_code

        success, _ = code_service.verify("test@example.com", CODE_TYPE_EMAIL_BIND, sample_code.code, consume=False)

        assert success is True
        assert sample_code.is_used is False


class TestVerifyPhone:
    """validate_phone 函数测试"""

    def test_valid_phone(self):
        """有效手机号"""
        ok, err = validate_phone("13800000001")
        assert ok is True
        assert err == ""

    def test_valid_phone_with_plus86(self):
        """带 +86 前缀的手机号"""
        ok, err = validate_phone("+8613800000001")
        assert ok is True

    def test_invalid_phone_short(self):
        """太短的手机号"""
        ok, err = validate_phone("123")
        assert ok is False
        assert "格式不正确" in err

    def test_invalid_phone_wrong_prefix(self):
        """号段不对的手机号"""
        ok, err = validate_phone("11000000000")
        assert ok is False

    def test_empty_phone(self):
        """空手机号"""
        ok, err = validate_phone("")
        assert ok is False
        assert "不能为空" in err

    def test_phone_with_spaces(self):
        """带空格的手机号会被 strip"""
        ok, _ = validate_phone("  13800000001  ")
        assert ok is True


class TestNormalizePhone:
    """normalize_phone 函数测试"""

    def test_removes_plus86(self):
        """去掉 +86 前缀"""
        assert normalize_phone("+8613800000001") == "13800000001"

    def test_no_prefix_stays_same(self):
        """没有前缀保持不变"""
        assert normalize_phone("13800000001") == "13800000001"

    def test_strips_whitespace(self):
        """去掉两端空白"""
        assert normalize_phone("  13800000001  ") == "13800000001"


class TestValidateEmail:
    """validate_email 函数测试"""

    def test_valid_email(self):
        """有效邮箱"""
        ok, err = validate_email("test@example.com")
        assert ok is True
        assert err == ""

    def test_valid_email_with_subdomain(self):
        """带子域名的邮箱"""
        ok, _ = validate_email("user@mail.example.com")
        assert ok is True

    def test_valid_email_with_plus(self):
        """带 + 号的邮箱"""
        ok, _ = validate_email("user+tag@example.com")
        assert ok is True

    def test_invalid_email_no_at(self):
        """没有 @ 的邮箱"""
        ok, err = validate_email("notanemail")
        assert ok is False
        assert "格式不正确" in err

    def test_invalid_email_no_domain(self):
        """没有域名的邮箱"""
        ok, err = validate_email("user@")
        assert ok is False

    def test_empty_email(self):
        """空邮箱"""
        ok, err = validate_email("")
        assert ok is False
        assert "不能为空" in err

    def test_email_with_spaces(self):
        """带空格的邮箱会被 strip"""
        ok, _ = validate_email("  test@example.com  ")
        assert ok is True