shaguabijia-app-server/tests/test_auth.py

"""auth 流程测试。

- mock 路径(SMS_MOCK=true,conftest 设置):验证码不真发,校验放行任意 6 位,测 JWT 闭环。
- real 路径:monkeypatch SMS_MOCK=false + 拦 httpx,测极光发送协议 + 真实校验(一次性/
  防爆破/余额回滚),不真发短信。
"""
from __future__ import annotations

import time

import pytest

from app.integrations import sms


def _reset(phone: str) -> None:
    """清该号的进程内存状态,隔离 real 模式用例。"""
    sms._codes.pop(phone, None)
    sms._last_sent.pop(phone, None)
    sms._daily_count.pop(phone, None)


class _OkResp:
    """模拟极光发送成功响应。"""
    status_code = 200

    def json(self):
        return {"msg_id": "test-msg-id"}


# ============================ mock 路径(JWT 闭环)============================

def test_sms_login_and_me_flow(client) -> None:
    """sms_send → sms_login(mock 任意6位)→ /me → refresh/logout 全闭环。"""
    phone = "13800138000"

    r = client.post("/api/v1/auth/sms/send", json={"phone": phone})
    assert r.status_code == 200, r.text
    body = r.json()
    assert body["sent"] is True
    assert body["mock"] is True

    r = client.post("/api/v1/auth/sms/login", json={"phone": phone, "code": "123456"})
    assert r.status_code == 200, r.text
    data = r.json()
    assert "access_token" in data and "refresh_token" in data
    assert data["user"]["phone"] == phone
    assert data["user"]["register_channel"] == "sms"
    access, refresh = data["access_token"], data["refresh_token"]

    r = client.get("/api/v1/auth/me", headers={"Authorization": f"Bearer {access}"})
    assert r.status_code == 200, r.text
    assert r.json()["phone"] == phone

    r = client.get("/api/v1/auth/me")
    assert r.status_code == 401

    r = client.post("/api/v1/auth/refresh", json={"refresh_token": refresh})
    assert r.status_code == 200, r.text
    assert "access_token" in r.json()

    # 用 access 当 refresh → 拒绝(防 token 类型混用)
    r = client.post("/api/v1/auth/refresh", json={"refresh_token": access})
    assert r.status_code == 401

    r = client.post("/api/v1/auth/logout", headers={"Authorization": f"Bearer {access}"})
    assert r.status_code == 200
    assert r.json()["ok"] is True


def test_sms_send_too_frequent(client) -> None:
    phone = "13900139000"
    assert client.post("/api/v1/auth/sms/send", json={"phone": phone}).status_code == 200
    assert client.post("/api/v1/auth/sms/send", json={"phone": phone}).status_code == 429


def test_sms_login_bad_code(client) -> None:
    """mock 下校验位数:5 位过 schema 但业务要 6 位 → 400。"""
    phone = "13700137000"
    client.post("/api/v1/auth/sms/send", json={"phone": phone})
    r = client.post("/api/v1/auth/sms/login", json={"phone": phone, "code": "12345"})
    assert r.status_code == 400
    assert "invalid sms code" in r.json()["detail"]


def test_phone_format_validation(client) -> None:
    r = client.post("/api/v1/auth/sms/send", json={"phone": "1234567"})
    assert r.status_code == 422


def test_sms_daily_limit(monkeypatch) -> None:
    """单号每日上限(send 的防刷逻辑 mock/real 都跑;函数级绕开 60s 冷却)。"""
    phone = "13511135000"
    _reset(phone)
    limit = 3
    monkeypatch.setattr(sms.settings, "SMS_DAILY_LIMIT_PER_PHONE", limit)

    for _ in range(limit):
        sms.send_code(phone)
        sms._last_sent.pop(phone, None)  # 绕开冷却,只测每日上限

    with pytest.raises(sms.SmsError, match="上限"):
        sms.send_code(phone)


# ============================ real 路径(不真发)============================

def test_sms_real_send_calls_jiguang(monkeypatch) -> None:
    """real 模式:本服务生成 code 并按协议调极光 /v1/messages(拦 httpx,不真发)。"""
    phone = "13422134000"
    _reset(phone)
    captured: dict = {}

    def _fake_post(url, json, headers, timeout):  # noqa: A002
        captured.update(url=url, body=json, auth=headers.get("Authorization", ""))
        return _OkResp()

    monkeypatch.setattr(sms.settings, "SMS_MOCK", False)
    monkeypatch.setattr(sms.httpx, "post", _fake_post)

    sms.send_code(phone)

    assert captured["url"] == sms.settings.SMS_SEND_ENDPOINT
    assert captured["body"]["mobile"] == phone
    assert captured["body"]["sign_id"] == sms.settings.SMS_SIGN_ID
    assert captured["body"]["temp_id"] == sms.settings.SMS_TEMPLATE_ID
    assert captured["body"]["temp_para"]["code"] == sms._codes[phone].code
    assert captured["auth"].startswith("Basic ")


def test_sms_real_verify_one_time_and_wrong(monkeypatch) -> None:
    """real 校验:错误码拒(不消费)→ 正确码成功 → 验过即作废。"""
    phone = "13455134000"
    _reset(phone)
    monkeypatch.setattr(sms.settings, "SMS_MOCK", False)
    monkeypatch.setattr(sms.httpx, "post", lambda *a, **k: _OkResp())

    sms.send_code(phone)
    code = sms._codes[phone].code
    wrong = "000000" if code != "000000" else "111111"

    assert sms.verify_code(phone, wrong) is False
    assert sms.verify_code(phone, code) is True
    assert sms.verify_code(phone, code) is False  # 已作废


def test_sms_real_verify_attempts_exhausted(monkeypatch) -> None:
    """real 校验:错误次数到上限即作废,正确码也不再通过(防爆破)。"""
    phone = "13466134000"
    _reset(phone)
    monkeypatch.setattr(sms.settings, "SMS_MOCK", False)
    monkeypatch.setattr(sms.settings, "SMS_MAX_VERIFY_ATTEMPTS", 3)
    monkeypatch.setattr(sms.httpx, "post", lambda *a, **k: _OkResp())

    sms.send_code(phone)
    code = sms._codes[phone].code
    wrong = "000000" if code != "000000" else "111111"

    for _ in range(3):
        assert sms.verify_code(phone, wrong) is False
    assert sms.verify_code(phone, code) is False  # 超限作废


def test_sms_real_balance_error_keeps_cooldown(monkeypatch) -> None:
    """极光余额不足(50014)→ SmsError(503);保留冷却(失败也限速,防重试狂打),只清未发出的码。"""
    phone = "13433134000"
    _reset(phone)

    class _ErrResp:
        status_code = 403

        def json(self):
            return {"error": {"code": 50014, "message": "no money"}}

    monkeypatch.setattr(sms.settings, "SMS_MOCK", False)
    monkeypatch.setattr(sms.httpx, "post", lambda *a, **k: _ErrResp())

    with pytest.raises(sms.SmsError) as ei:
        sms.send_code(phone)
    assert ei.value.status_code == 503
    assert phone not in sms._codes      # 没发出去的码已清
    assert phone in sms._last_sent      # 冷却保留:失败也限速

    # 立即重试 → 被冷却挡下(429),不会再打极光
    with pytest.raises(sms.SmsError) as ei2:
        sms.send_code(phone)
    assert ei2.value.status_code == 429


def test_sms_gc_purges_stale_only(monkeypatch) -> None:
    """GC 清过期码 / 旧冷却 / 隔日计数,但不动今天有效的(阈值设 0 强制每次扫)。"""
    monkeypatch.setattr(sms, "_GC_THRESHOLD", 0)
    sms._codes.clear()
    sms._last_sent.clear()
    sms._daily_count.clear()
    now = time.time()
    sms._codes["stale"] = sms._CodeRecord(code="111111", expires_at=now - 1)
    sms._codes["fresh"] = sms._CodeRecord(code="222222", expires_at=now + 999)
    sms._last_sent["old"] = now - 99999
    sms._last_sent["recent"] = now
    sms._daily_count["yesterday"] = ("2000-01-01", 3)
    sms._daily_count["today"] = (sms._today(), 1)

    sms._gc(now)

    assert "stale" not in sms._codes and "fresh" in sms._codes
    assert "old" not in sms._last_sent and "recent" in sms._last_sent
    assert "yesterday" not in sms._daily_count and "today" in sms._daily_count