xiaoxia-saas/tests/unit/test_verification_code_service.py

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

import re
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_EMAIL_LOGIN,
    CODE_TYPE_PHONE_BIND,
    DAILY_LIMIT,
    MAX_ATTEMPTS,
    RESEND_COOLDOWN_SECONDS,
    VerificationCodeService,
    normalize_phone,
    validate_email,
    validate_phone,
)
from packages.domain.verification_code import VerificationCode

# ── Test Fixtures ────────────────────────────────────────────────────────────


@pytest.fixture
def mock_repo():
    """mock 验证码仓储."""
    repo = MagicMock()
    repo.find_latest.return_value = None
    repo.count_today.return_value = 0
    return repo


@pytest.fixture
def service(mock_repo):
    """验证码服务实例."""
    return VerificationCodeService(repo=mock_repo)


def _make_code(
    recipient="test@example.com",
    code_type=CODE_TYPE_EMAIL_BIND,
    code="123456",
    ttl=300,
    used=False,
    attempts=0,
    created_at=None,
):
    """创建一个测试用验证码实体."""
    now = created_at or datetime.now(timezone.utc)
    vc = VerificationCode(
        id="test-code-id",
        recipient=recipient,
        code=code,
        code_type=code_type,
        expires_at=now + timedelta(seconds=ttl),
        used_at=now if used else None,
        attempts=attempts,
        created_at=now,
    )
    return vc


# ── generate 方法测试 ───────────────────────────────────────────────────────


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

    def test_generate_success(self, service, mock_repo):
        """成功生成验证码."""
        code, error = service.generate("user@example.com", CODE_TYPE_EMAIL_BIND)

        assert error is None
        assert code is not None
        assert code.recipient == "user@example.com"
        assert code.code_type == CODE_TYPE_EMAIL_BIND
        assert len(code.code) == 6
        assert code.code.isdigit()
        assert not code.is_used
        mock_repo.save.assert_called_once()

    def test_generate_with_custom_code(self, service, mock_repo):
        """使用自定义验证码."""
        code, error = service.generate("user@example.com", CODE_TYPE_EMAIL_LOGIN, custom_code="999999")

        assert error is None
        assert code.code == "999999"

    def test_generate_custom_ttl(self, service, mock_repo):
        """自定义 TTL."""
        code, _ = service.generate("u@e.com", CODE_TYPE_EMAIL_BIND, ttl_seconds=600)
        delta = code.expires_at - code.created_at
        assert delta.total_seconds() == 600

    def test_generate_default_ttl(self, service, mock_repo):
        """默认 TTL."""
        code, _ = service.generate("u@e.com", CODE_TYPE_EMAIL_BIND)
        delta = code.expires_at - code.created_at
        assert delta.total_seconds() == 300  # 默认5分钟

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

    def test_generate_whitespace_recipient(self, service):
        """全空白接收方."""
        code, error = service.generate("   ", CODE_TYPE_EMAIL_BIND)
        assert code is None
        assert "不能为空" in error

    def test_generate_invalid_type(self, service):
        """无效验证码类型."""
        code, error = service.generate("u@e.com", "invalid_type")
        assert code is None
        assert "无效的验证码类型" in error

    def test_generate_recipient_stripped(self, service, mock_repo):
        """接收方前后空格会被清理."""
        code, _ = service.generate("  user@e.com  ", CODE_TYPE_EMAIL_BIND)
        assert code.recipient == "user@e.com"

    def test_generate_phone_code(self, service, mock_repo):
        """手机验证码生成."""
        code, error = service.generate("13800138000", CODE_TYPE_PHONE_BIND)
        assert error is None
        assert code.code_type == CODE_TYPE_PHONE_BIND
        assert len(code.code) == 6


# ── generate 频控测试 ───────────────────────────────────────────────────────


class TestGenerateRateLimit:
    """generate 频控测试"""

    def test_cooldown_active_rejects(self, service, mock_repo):
        """冷却期内拒绝重发."""
        recent = _make_code(created_at=datetime.now(timezone.utc) - timedelta(seconds=10))
        mock_repo.find_latest.return_value = recent

        code, error = service.generate("u@e.com", CODE_TYPE_EMAIL_BIND)
        assert code is None
        assert "发送太频繁" in error
        # 等待时间应该接近 50 秒 (60-10)
        match = re.search(r"(\d+)\s*秒", error)
        assert match
        wait = int(match.group(1))
        assert 45 <= wait <= 55

    def test_cooldown_expired_allows(self, service, mock_repo):
        """冷却期过后允许重发."""
        old = _make_code(created_at=datetime.now(timezone.utc) - timedelta(seconds=120))
        mock_repo.find_latest.return_value = old
        mock_repo.count_today.return_value = 1

        code, error = service.generate("u@e.com", CODE_TYPE_EMAIL_BIND)
        assert error is None
        assert code is not None

    def test_daily_limit_reached(self, service, mock_repo):
        """达到每日上限."""
        mock_repo.find_latest.return_value = None
        mock_repo.count_today.return_value = DAILY_LIMIT

        code, error = service.generate("u@e.com", CODE_TYPE_EMAIL_BIND)
        assert code is None
        assert "今日发送次数已达上限" in error

    def test_daily_limit_one_below_allows(self, service, mock_repo):
        """未达到上限时允许."""
        mock_repo.find_latest.return_value = None
        mock_repo.count_today.return_value = DAILY_LIMIT - 1

        code, error = service.generate("u@e.com", CODE_TYPE_EMAIL_BIND)
        assert error is None
        assert code is not None

    def test_custom_daily_limit(self, mock_repo):
        """自定义每日上限."""
        svc = VerificationCodeService(repo=mock_repo, daily_limit=3)
        mock_repo.count_today.return_value = 3

        code, error = svc.generate("u@e.com", CODE_TYPE_EMAIL_BIND)
        assert code is None
        assert "已达上限" in error

    def test_custom_cooldown(self, mock_repo):
        """自定义冷却时间."""
        svc = VerificationCodeService(repo=mock_repo, resend_cooldown=30)
        recent = _make_code(created_at=datetime.now(timezone.utc) - timedelta(seconds=10))
        mock_repo.find_latest.return_value = recent

        code, error = svc.generate("u@e.com", CODE_TYPE_EMAIL_BIND)
        assert code is None
        match = re.search(r"(\d+)\s*秒", error)
        assert match
        wait = int(match.group(1))
        assert 15 <= wait <= 25

    def test_cooldown_different_types_independent(self, service, mock_repo):
        """不同类型的验证码冷却独立."""
        # email_bind 类型有一个近期验证码
        recent = _make_code(code_type=CODE_TYPE_EMAIL_BIND)
        mock_repo.find_latest.side_effect = lambda r, t: recent if t == CODE_TYPE_EMAIL_BIND else None
        mock_repo.count_today.return_value = 0

        # email_login 类型应该可以正常发送
        code, error = service.generate("u@e.com", CODE_TYPE_EMAIL_LOGIN)
        assert error is None
        assert code is not None


# ── verify 方法测试 ─────────────────────────────────────────────────────────


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

    def test_verify_success(self, service, mock_repo):
        """验证码正确."""
        code = _make_code(code="654321")
        mock_repo.find_latest.return_value = code

        ok, error = service.verify("test@example.com", CODE_TYPE_EMAIL_BIND, "654321")
        assert ok is True
        assert error is None
        assert code.is_used  # 标记为已使用
        assert mock_repo.save.call_count >= 2  # increment + mark_used

    def test_verify_wrong_code(self, service, mock_repo):
        """验证码错误."""
        code = _make_code(code="123456")
        mock_repo.find_latest.return_value = code

        ok, error = service.verify("test@e.com", CODE_TYPE_EMAIL_BIND, "000000")
        assert ok is False
        assert "验证码错误" in error
        assert not code.is_used  # 不标记为已使用
        assert code.attempts == 1  # 尝试次数+1

    def test_verify_no_code_found(self, service, mock_repo):
        """找不到验证码."""
        mock_repo.find_latest.return_value = None

        ok, error = service.verify("u@e.com", CODE_TYPE_EMAIL_BIND, "123456")
        assert ok is False
        assert "不存在或已过期" in error

    def test_verify_empty_params(self, service):
        """参数为空."""
        ok, error = service.verify("", CODE_TYPE_EMAIL_BIND, "123456")
        assert ok is False
        assert "参数不完整" in error

        ok2, error2 = service.verify("u@e.com", CODE_TYPE_EMAIL_BIND, "")
        assert ok2 is False
        assert "参数不完整" in error2

    def test_verify_whitespace_params(self, service, mock_repo):
        """参数前后空格会被清理."""
        code = _make_code(recipient="u@e.com", code="111111")
        mock_repo.find_latest.return_value = code

        ok, error = service.verify("  u@e.com  ", CODE_TYPE_EMAIL_BIND, "  111111  ")
        assert ok is True
        assert error is None

    def test_verify_already_used(self, service, mock_repo):
        """验证码已使用."""
        code = _make_code(used=True)
        mock_repo.find_latest.return_value = code

        ok, error = service.verify("u@e.com", CODE_TYPE_EMAIL_BIND, code.code)
        assert ok is False
        assert "已使用" in error

    def test_verify_expired(self, service, mock_repo):
        """验证码已过期."""
        code = _make_code(ttl=-60)  # 已过期
        mock_repo.find_latest.return_value = code

        ok, error = service.verify("u@e.com", CODE_TYPE_EMAIL_BIND, code.code)
        assert ok is False
        assert "已过期" in error

    def test_verify_too_many_attempts(self, service, mock_repo):
        """尝试次数过多."""
        code = _make_code(attempts=MAX_ATTEMPTS + 1)
        mock_repo.find_latest.return_value = code

        ok, error = service.verify("u@e.com", CODE_TYPE_EMAIL_BIND, code.code)
        assert ok is False
        assert "验证次数过多" in error

    def test_verify_attempts_increment_each_time(self, service, mock_repo):
        """每次错误尝试都增加尝试次数."""
        code = _make_code(code="123456", attempts=0)
        mock_repo.find_latest.return_value = code

        for _ in range(3):
            service.verify("u@e.com", CODE_TYPE_EMAIL_BIND, "wrong")

        assert code.attempts == 3

    def test_verify_without_consume(self, service, mock_repo):
        """验证成功但不标记为已使用（consume=False）."""
        code = _make_code(code="999999")
        mock_repo.find_latest.return_value = code

        ok, error = service.verify("u@e.com", CODE_TYPE_EMAIL_BIND, "999999", consume=False)
        assert ok is True
        assert error is None
        assert not code.is_used  # 不标记为已使用

    def test_verify_consume_default_true(self, service, mock_repo):
        """默认 consume=True."""
        code = _make_code(code="123456")
        mock_repo.find_latest.return_value = code

        service.verify("u@e.com", CODE_TYPE_EMAIL_BIND, "123456")
        assert code.is_used

    def test_verify_used_checked_before_attempts(self, service, mock_repo):
        """已使用优先于其他检查."""
        code = _make_code(used=True, attempts=0)
        mock_repo.find_latest.return_value = code

        ok, error = service.verify("u@e.com", CODE_TYPE_EMAIL_BIND, code.code)
        assert ok is False
        assert "已使用" in error
        # attempts 会被 increment，但错误原因是已使用
        assert code.attempts == 1

    def test_custom_max_attempts(self, mock_repo):
        """自定义最大尝试次数."""
        svc = VerificationCodeService(repo=mock_repo, max_attempts=2)
        code = _make_code(attempts=2)
        mock_repo.find_latest.return_value = code

        ok, error = svc.verify("u@e.com", CODE_TYPE_EMAIL_BIND, code.code)
        assert ok is False
        assert "验证次数过多" in error


# ── validate_phone 测试 ─────────────────────────────────────────────────────


class TestValidatePhone:
    """手机号格式校验测试"""

    def test_valid_11_digit(self):
        """标准11位手机号."""
        ok, msg = validate_phone("13800138000")
        assert ok is True
        assert msg == ""

    def test_valid_with_plus_86(self):
        """带+86前缀."""
        ok, msg = validate_phone("+8613800138000")
        assert ok is True

    def test_invalid_too_short(self):
        """位数不足."""
        ok, msg = validate_phone("1380013800")
        assert ok is False
        assert "格式不正确" in msg

    def test_invalid_too_long(self):
        """位数过多."""
        ok, msg = validate_phone("138001380001")
        assert ok is False

    def test_invalid_starts_with_2(self):
        """开头不是1."""
        ok, msg = validate_phone("23800138000")
        assert ok is False

    def test_invalid_starts_with_12(self):
        """第二位不在3-9."""
        ok, msg = validate_phone("12800138000")
        assert ok is False

    def test_invalid_empty(self):
        """空字符串."""
        ok, msg = validate_phone("")
        assert ok is False
        assert "不能为空" in msg

    def test_invalid_whitespace_only(self):
        """仅空白."""
        ok, msg = validate_phone("   ")
        assert ok is False
        assert "不能为空" in msg

    def test_valid_all_prefixes_3_to_9(self):
        """第二位3-9都有效."""
        for n in range(3, 10):
            ok, _ = validate_phone(f"1{n}800138000")
            assert ok is True, f"1{n} prefix should be valid"

    def test_invalid_contains_letters(self):
        """包含字母."""
        ok, msg = validate_phone("13800abc000")
        assert ok is False

    def test_strips_whitespace(self):
        """前后空格会被清理."""
        ok, msg = validate_phone("  13800138000  ")
        assert ok is True


# ── normalize_phone 测试 ────────────────────────────────────────────────────


class TestNormalizePhone:
    """手机号标准化测试"""

    def test_strip_plus_86(self):
        """去掉+86前缀."""
        assert normalize_phone("+8613800138000") == "13800138000"

    def test_no_prefix_stays_same(self):
        """无前缀保持不变."""
        assert normalize_phone("13800138000") == "13800138000"

    def test_strips_whitespace(self):
        """清理前后空格."""
        assert normalize_phone("  13800138000  ") == "13800138000"

    def test_plus_86_with_spaces(self):
        """带空格的+86."""
        assert normalize_phone("  +8613800138000  ") == "13800138000"


# ── validate_email 测试 ─────────────────────────────────────────────────────


class TestValidateEmail:
    """邮箱格式校验测试"""

    def test_valid_simple(self):
        """标准邮箱."""
        ok, msg = validate_email("user@example.com")
        assert ok is True
        assert msg == ""

    def test_valid_with_dots(self):
        """带点号的用户名."""
        ok, _ = validate_email("user.name@example.com")
        assert ok is True

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

    def test_valid_with_underscore(self):
        """带下划线."""
        ok, _ = validate_email("user_name@example.com")
        assert ok is True

    def test_valid_subdomain(self):
        """多级域名."""
        ok, _ = validate_email("user@mail.example.com")
        assert ok is True

    def test_invalid_no_at(self):
        """没有@."""
        ok, msg = validate_email("userexample.com")
        assert ok is False
        assert "格式不正确" in msg

    def test_invalid_empty_local(self):
        """@前为空."""
        ok, _ = validate_email("@example.com")
        assert ok is False

    def test_invalid_empty_domain(self):
        """@后为空."""
        ok, _ = validate_email("user@")
        assert ok is False

    def test_invalid_no_tld(self):
        """没有顶级域名."""
        ok, _ = validate_email("user@example")
        assert ok is False

    def test_invalid_empty(self):
        """空字符串."""
        ok, msg = validate_email("")
        assert ok is False
        assert "不能为空" in msg

    def test_invalid_spaces_only(self):
        """仅空白."""
        ok, msg = validate_email("   ")
        assert ok is False
        assert "不能为空" in msg

    def test_strips_whitespace(self):
        """前后空格会被清理."""
        ok, msg = validate_email("  user@e.com  ")
        assert ok is True

    def test_invalid_special_chars(self):
        """特殊字符."""
        ok, _ = validate_email("user name@e.com")
        assert ok is False

    def test_valid_numbers(self):
        """数字邮箱."""
        ok, _ = validate_email("12345@example.com")
        assert ok is True


# ── VerificationCode 实体辅助验证 ──────────────────────────────────────────


class TestVerificationCodeEntity:
    """VerificationCode 实体属性测试"""

    def test_is_expired_false_when_fresh(self):
        code = _make_code(ttl=300)
        assert code.is_expired is False

    def test_is_expired_true_when_past(self):
        code = _make_code(ttl=-1)
        assert code.is_expired is True

    def test_is_used_false_initially(self):
        code = _make_code()
        assert code.is_used is False

    def test_is_used_after_mark_used(self):
        code = _make_code()
        code.mark_used()
        assert code.is_used is True
        assert code.used_at is not None

    def test_is_valid_fresh(self):
        code = _make_code()
        assert code.is_valid is True

    def test_is_valid_when_expired(self):
        code = _make_code(ttl=-100)
        assert code.is_valid is False

    def test_is_valid_when_used(self):
        code = _make_code(used=True)
        assert code.is_valid is False

    def test_increment_attempts(self):
        code = _make_code(attempts=0)
        code.increment_attempts()
        assert code.attempts == 1
        code.increment_attempts()
        assert code.attempts == 2

    def test_create_generates_6_digit_code(self):
        code = VerificationCode.create("u@e.com", CODE_TYPE_EMAIL_BIND)
        assert len(code.code) == 6
        assert code.code.isdigit()

    def test_create_custom_code(self):
        code = VerificationCode.create("u@e.com", CODE_TYPE_EMAIL_BIND, custom_code="555555")
        assert code.code == "555555"

    def test_create_strips_recipient(self):
        code = VerificationCode.create("  u@e.com  ", CODE_TYPE_EMAIL_BIND)
        assert code.recipient == "u@e.com"

    def test_create_sets_expiry(self):
        code = VerificationCode.create("u@e.com", CODE_TYPE_EMAIL_BIND, ttl_seconds=120)
        delta = code.expires_at - code.created_at
        assert delta.total_seconds() == 120