xiaoxia-saas/tests/unit/test_verification_code_service.py

"""
验证码服务单元测试（第十七波）

覆盖：
- VerificationCodeService.generate
- VerificationCodeService.verify
- 频控逻辑（冷却 + 每日上限）
"""

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,
    DEFAULT_TTL_SECONDS,
    MAX_ATTEMPTS,
    RESEND_COOLDOWN_SECONDS,
    VerificationCodeService,
)
from packages.domain.verification_code import VerificationCode


@pytest.fixture
def mock_repo():
    """mock 验证码仓储"""
    return MagicMock()


@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)
    return 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,
    )


# ============================================================
# generate - 参数校验
# ============================================================


class TestGenerateParamValidation:
    """generate 参数校验"""

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

    def test_whitespace_recipient_stripped(self, service, mock_repo):
        """前后空格会被 strip 掉，正常生成"""
        mock_repo.find_latest.return_value = None
        mock_repo.count_today.return_value = 0
        code, err = service.generate("  test@example.com  ", CODE_TYPE_EMAIL_BIND)
        assert err is None
        assert code is not None
        assert code.recipient == "test@example.com"

    def test_invalid_code_type(self, service):
        """无效验证码类型"""
        code, err = service.generate("test@example.com", "invalid_type")
        assert code is None
        assert "无效的验证码类型" in err


# ============================================================
# generate - 正常生成
# ============================================================


class TestGenerateNormal:
    """generate 正常生成场景"""

    def test_generate_success(self, service, mock_repo):
        """正常生成验证码"""
        mock_repo.find_latest.return_value = None
        mock_repo.count_today.return_value = 0

        code, err = service.generate("test@example.com", CODE_TYPE_EMAIL_BIND)

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

    def test_custom_code(self, service, mock_repo):
        """自定义验证码"""
        mock_repo.find_latest.return_value = None
        mock_repo.count_today.return_value = 0

        code, err = service.generate("test@example.com", CODE_TYPE_EMAIL_BIND, custom_code="888888")

        assert err is None
        assert code.code == "888888"

    def test_custom_ttl(self, service, mock_repo):
        """自定义有效期"""
        mock_repo.find_latest.return_value = None
        mock_repo.count_today.return_value = 0

        code, err = service.generate("test@example.com", CODE_TYPE_EMAIL_BIND, ttl_seconds=60)

        assert err is None
        # 过期时间 - 创建时间 ≈ 60 秒
        delta = (code.expires_at - code.created_at).total_seconds()
        assert delta == 60

    def test_default_ttl_used_when_not_specified(self, service, mock_repo):
        """未指定 ttl 时使用默认值"""
        mock_repo.find_latest.return_value = None
        mock_repo.count_today.return_value = 0

        code, err = service.generate("test@example.com", CODE_TYPE_EMAIL_BIND)

        assert err is None
        delta = (code.expires_at - code.created_at).total_seconds()
        assert delta == DEFAULT_TTL_SECONDS

    def test_phone_bind_type(self, service, mock_repo):
        """手机号绑定类型也支持"""
        mock_repo.find_latest.return_value = None
        mock_repo.count_today.return_value = 0

        code, err = service.generate("13800138000", CODE_TYPE_PHONE_BIND)

        assert err is None
        assert code.code_type == CODE_TYPE_PHONE_BIND


# ============================================================
# generate - 频控
# ============================================================


class TestGenerateRateLimit:
    """generate 频控逻辑"""

    def test_resend_cooldown_blocked(self, service, mock_repo):
        """冷却期内发送被拒绝"""
        # 10 秒前刚发过一条
        recent = make_code(created_at=datetime.now(timezone.utc) - timedelta(seconds=10))
        mock_repo.find_latest.return_value = recent
        mock_repo.count_today.return_value = 1

        code, err = service.generate("test@example.com", CODE_TYPE_EMAIL_BIND)

        assert code is None
        assert "发送太频繁" in err
        assert "秒后再试" in err
        # 等待时间应接近 50 秒（60-10）
        # 提取数字验证范围
        import re

        match = re.search(r"(\d+)\s*秒", err)
        assert match
        wait = int(match.group(1))
        assert 45 <= wait <= 55

    def test_resend_after_cooldown_ok(self, service, mock_repo):
        """超过冷却期可以重发"""
        # 2 分钟前发的，已过冷却
        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, err = service.generate("test@example.com", CODE_TYPE_EMAIL_BIND)

        assert err is None
        assert code is not None

    def test_daily_limit_reached(self, service, mock_repo):
        """达到每日上限"""
        # 没有最近的（过了冷却），但今日已达上限
        old = make_code(created_at=datetime.now(timezone.utc) - timedelta(hours=2))
        mock_repo.find_latest.return_value = old
        mock_repo.count_today.return_value = DAILY_LIMIT

        code, err = service.generate("test@example.com", CODE_TYPE_EMAIL_BIND)

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

    def test_daily_limit_not_reached(self, service, mock_repo):
        """未达每日上限可以发"""
        mock_repo.find_latest.return_value = None
        mock_repo.count_today.return_value = DAILY_LIMIT - 1

        code, err = service.generate("test@example.com", CODE_TYPE_EMAIL_BIND)

        assert err is None
        assert code is not None

    def test_no_history_first_time_ok(self, service, mock_repo):
        """首次发送，无历史记录"""
        mock_repo.find_latest.return_value = None
        mock_repo.count_today.return_value = 0

        code, err = service.generate("test@example.com", CODE_TYPE_EMAIL_BIND)

        assert err is None
        assert code is not None
        mock_repo.save.assert_called_once()


# ============================================================
# generate - 自定义频控参数
# ============================================================


class TestGenerateCustomRateLimitParams:
    """自定义频控参数"""

    def test_custom_cooldown(self, mock_repo):
        """自定义冷却时间"""
        svc = VerificationCodeService(repo=mock_repo, resend_cooldown=300, daily_limit=5)
        # 60 秒前发的，默认冷却 60 秒就够了，但这里设了 300 秒
        recent = make_code(created_at=datetime.now(timezone.utc) - timedelta(seconds=60))
        mock_repo.find_latest.return_value = recent
        mock_repo.count_today.return_value = 1

        code, err = svc.generate("test@example.com", CODE_TYPE_EMAIL_BIND)

        assert code is None
        assert "发送太频繁" in err

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

        code, err = svc.generate("test@example.com", CODE_TYPE_EMAIL_BIND)

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


# ============================================================
# verify - 参数校验
# ============================================================


class TestVerifyParamValidation:
    """verify 参数校验"""

    def test_empty_recipient(self, service):
        """空接收方"""
        ok, err = service.verify("", CODE_TYPE_EMAIL_BIND, "123456")
        assert not ok
        assert "参数不完整" in err

    def test_empty_code(self, service):
        """空验证码"""
        ok, err = service.verify("test@example.com", CODE_TYPE_EMAIL_BIND, "")
        assert not ok
        assert "参数不完整" in err

    def test_whitespace_stripped(self, service, mock_repo):
        """前后空格会被 strip"""
        code = make_code(code="123456")
        mock_repo.find_latest.return_value = code
        mock_repo.count_today.return_value = 0

        ok, err = service.verify("  test@example.com  ", CODE_TYPE_EMAIL_BIND, "  123456  ")

        assert ok
        assert err is None


# ============================================================
# verify - 正常验证
# ============================================================


class TestVerifyNormal:
    """verify 正常验证场景"""

    def test_verify_success_consume(self, service, mock_repo):
        """验证成功并消耗"""
        code = make_code(code="123456")
        mock_repo.find_latest.return_value = code

        ok, err = service.verify("test@example.com", CODE_TYPE_EMAIL_BIND, "123456", consume=True)

        assert ok
        assert err is None
        assert code.is_used  # 被标记为已使用
        # save 被调用了两次：一次 increment_attempts 后，一次 mark_used 后
        assert mock_repo.save.call_count >= 2

    def test_verify_success_no_consume(self, service, mock_repo):
        """验证成功但不消耗"""
        code = make_code(code="123456")
        mock_repo.find_latest.return_value = code

        ok, err = service.verify("test@example.com", CODE_TYPE_EMAIL_BIND, "123456", consume=False)

        assert ok
        assert err is None
        assert not code.is_used  # 未被标记

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

        ok, err = service.verify("test@example.com", CODE_TYPE_EMAIL_BIND, "123456")

        assert not ok
        assert "不存在或已过期" in err

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

        ok, err = service.verify("test@example.com", CODE_TYPE_EMAIL_BIND, "999999")

        assert not ok
        assert "验证码错误" in err
        # 尝试次数增加了
        assert code.attempts == 1

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

        ok, err = service.verify("test@example.com", CODE_TYPE_EMAIL_BIND, "123456")

        assert not ok
        assert "已使用" in err

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

        ok, err = service.verify("test@example.com", CODE_TYPE_EMAIL_BIND, "123456")

        assert not ok
        assert "已过期" in err

    def test_verify_attempts_exceeded(self, service, mock_repo):
        """超过最大尝试次数"""
        code = make_code(code="123456", attempts=MAX_ATTEMPTS)
        mock_repo.find_latest.return_value = code

        ok, err = service.verify("test@example.com", CODE_TYPE_EMAIL_BIND, "123456")

        assert not ok
        assert "验证次数过多" in err
        # verify 里先 increment_attempts 再判断，所以这里 attempts 应该是 MAX_ATTEMPTS + 1
        assert code.attempts == MAX_ATTEMPTS + 1

    def test_attempts_increment_on_wrong_code(self, service, mock_repo):
        """错误验证码会增加尝试次数"""
        code = make_code(code="123456", attempts=0)
        mock_repo.find_latest.return_value = code

        service.verify("test@example.com", CODE_TYPE_EMAIL_BIND, "000000")
        assert code.attempts == 1

        service.verify("test@example.com", CODE_TYPE_EMAIL_BIND, "000001")
        assert code.attempts == 2


# ============================================================
# verify - 不同 code_type 互不干扰
# ============================================================


class TestVerifyCodeTypeIsolation:
    """不同验证码类型互不干扰"""

    def test_email_bind_vs_email_login(self, service, mock_repo):
        """用 email_login 类型的验证码去验证 email_bind 应该失败"""
        code = make_code(code_type=CODE_TYPE_EMAIL_LOGIN, code="123456")
        mock_repo.find_latest.return_value = None  # 按 email_bind 查不到

        # find_latest 按 code_type 查询，传 email_bind 返回 None
        def side_effect(recipient, ct):
            if ct == CODE_TYPE_EMAIL_LOGIN:
                return code
            return None

        mock_repo.find_latest.side_effect = side_effect

        ok, err = service.verify("test@example.com", CODE_TYPE_EMAIL_BIND, "123456")
        assert not ok
        assert "不存在或已过期" in err


# ============================================================
# 常量值检查
# ============================================================


class TestConstants:
    """常量默认值校验"""

    def test_default_cooldown_60(self):
        assert RESEND_COOLDOWN_SECONDS == 60

    def test_default_daily_limit_10(self):
        assert DAILY_LIMIT == 10

    def test_default_max_attempts_5(self):
        assert MAX_ATTEMPTS == 5

    def test_default_ttl_300(self):
        assert DEFAULT_TTL_SECONDS == 300

    def test_valid_code_types_count(self):
        """5 种验证码类型"""
        from packages.application.auth.verification_code_service import VALID_CODE_TYPES

        assert len(VALID_CODE_TYPES) == 5