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
from app.integrations.sms import jiguang


def _reset(phone: str) -> None:
    """清该号的进程内存状态,隔离 real 模式用例。"""
    jiguang._codes.pop(phone, None)
    jiguang._last_sent.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_send_device_ip_rate_limit(client, monkeypatch) -> None:
    """发码防刷:同一设备(device_id) + 同一 IP 每小时最多 N 次发码,超出 429。
    用不同手机号(绕开单号 60s 冷却)证明限流按设备封顶 —— 堵「换号绕开单号冷却」的洞。
    conftest 默认关限流;本用例临时打开 + 调小阈值便于测 + 清计数隔离。"""
    from app.api.v1 import auth
    from app.core import ratelimit

    monkeypatch.setattr(ratelimit.settings, "RATE_LIMIT_ENABLED", True)
    monkeypatch.setattr(auth, "SMS_SEND_MAX_PER_HOUR_PER_DEVICE", 3)
    ratelimit._buckets.clear()

    device = "dev-send-A"
    # 同设备、每次换不同手机号(绕开单号冷却)发码,前 3 次 200
    for i in range(3):
        r = client.post(
            "/api/v1/auth/sms/send",
            json={"phone": f"137001370{i:02d}", "device_id": device},
        )
        assert r.status_code == 200, f"第 {i + 1} 次应放行: {r.text}"
    # 第 4 次:同设备超限 → 429(即便又换了手机号)
    r = client.post(
        "/api/v1/auth/sms/send",
        json={"phone": "13700137003", "device_id": device},
    )
    assert r.status_code == 429, r.text

    # 换个设备 → 独立计数(限流键含 device_id),放行
    r = client.post(
        "/api/v1/auth/sms/send",
        json={"phone": "13700137004", "device_id": "dev-send-B"},
    )
    assert r.status_code == 200, r.text


def test_sms_send_cooldown_reject_not_counted(client, monkeypatch) -> None:
    """发码额度只算「成功发码」:被单号 60s 冷却挡下的重发(429)不占设备额度。
    做法:同号狂发只成功 1 次、其余被冷却挡下;把小时额度设 2,证明换号后仍能再成功发 1 次
    —— 若冷却重发也计数,额度早被那几次耗尽。"""
    from app.api.v1 import auth
    from app.core import ratelimit

    monkeypatch.setattr(ratelimit.settings, "RATE_LIMIT_ENABLED", True)
    monkeypatch.setattr(auth, "SMS_SEND_MAX_PER_HOUR_PER_DEVICE", 2)
    ratelimit._buckets.clear()

    device = "dev-cooldown"
    phone_a = "13710137000"
    # 首发成功(小时闸计 1/2)
    assert client.post(
        "/api/v1/auth/sms/send", json={"phone": phone_a, "device_id": device}
    ).status_code == 200
    # 同号连发 3 次:都被单号 60s 冷却挡下 → 429,且**不占**设备额度
    for _ in range(3):
        r = client.post(
            "/api/v1/auth/sms/send", json={"phone": phone_a, "device_id": device}
        )
        assert r.status_code == 429, r.text
    # 换号再发:设备额度只用了 1/2(冷却那几次没算)→ 仍放行(计到 2/2)
    assert client.post(
        "/api/v1/auth/sms/send", json={"phone": "13710137001", "device_id": device}
    ).status_code == 200
    # 又换号:此时小时闸已 2/2 → 429(反证成功发码确实各计了 1)
    r = client.post(
        "/api/v1/auth/sms/send", json={"phone": "13710137002", "device_id": device}
    )
    assert r.status_code == 429, r.text


def test_sms_send_daily_cap(client, monkeypatch) -> None:
    """每天发码上限(设备 + IP):成功发码累计到日上限即 429(用不同手机号绕开单号冷却)。
    抬高小时闸单独测日闸;超限文案含「今日」以便前端提示明天再来。"""
    from app.api.v1 import auth
    from app.core import ratelimit

    monkeypatch.setattr(ratelimit.settings, "RATE_LIMIT_ENABLED", True)
    monkeypatch.setattr(auth, "SMS_SEND_MAX_PER_HOUR_PER_DEVICE", 100)  # 抬高小时闸,不干扰
    monkeypatch.setattr(auth, "SMS_SEND_MAX_PER_DAY_PER_DEVICE", 3)
    ratelimit._buckets.clear()

    device = "dev-daily"
    for i in range(3):
        r = client.post(
            "/api/v1/auth/sms/send",
            json={"phone": f"13720137{i:03d}", "device_id": device},
        )
        assert r.status_code == 200, f"第 {i + 1} 次应放行: {r.text}"
    # 第 4 次:同设备同 IP 当日超限 → 429
    r = client.post(
        "/api/v1/auth/sms/send",
        json={"phone": "13720137999", "device_id": device},
    )
    assert r.status_code == 429, r.text
    assert "今日" in r.json()["detail"]


def test_sms_login_device_ip_rate_limit(client, monkeypatch) -> None:
    """防刷:同一设备(device_id) + 同一 IP 每小时最多 SMS_LOGIN_MAX_PER_HOUR 次登录尝试,超出 429。
    conftest 默认 RATE_LIMIT_ENABLED=false(内存计数跨用例累加),本用例临时打开并清空计数隔离。"""
    from app.api.v1 import auth
    from app.core import ratelimit

    monkeypatch.setattr(ratelimit.settings, "RATE_LIMIT_ENABLED", True)
    ratelimit._buckets.clear()  # 隔离:清掉跨用例累加的内存计数

    device = "dev-rl-A"
    # 前 N 次故意每次换手机号、固定设备:mock 校验放行 → 200。证明限流按设备而非手机号,同设备共用一个桶
    for i in range(auth.SMS_LOGIN_MAX_PER_HOUR):
        r = client.post(
            "/api/v1/auth/sms/login",
            json={"phone": f"135331533{i:02d}", "code": "123456", "device_id": device},
        )
        assert r.status_code == 200, f"第 {i + 1} 次应放行: {r.text}"
    # 第 N+1 次:同设备同 IP 超限 → 429(即便又换了个手机号)
    r = client.post(
        "/api/v1/auth/sms/login",
        json={"phone": "13533153399", "code": "123456", "device_id": device},
    )
    assert r.status_code == 429, r.text
    assert "频繁" in r.json()["detail"]

    # 同 IP 换个设备 → 独立计数(限流键含 device_id、非纯 IP/手机号),仍放行
    r = client.post(
        "/api/v1/auth/sms/login",
        json={"phone": "13533153300", "code": "123456", "device_id": "dev-rl-B"},
    )
    assert r.status_code == 200, r.text


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


# ============================ 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(jiguang.settings, "SMS_MOCK", False)
    monkeypatch.setattr(jiguang.httpx, "post", _fake_post)

    jiguang.send_code(phone)

    assert captured["url"] == jiguang.settings.SMS_SEND_ENDPOINT
    assert captured["body"]["mobile"] == phone
    assert captured["body"]["sign_id"] == jiguang.settings.SMS_SIGN_ID
    assert captured["body"]["temp_id"] == jiguang.settings.SMS_TEMPLATE_ID
    assert captured["body"]["temp_para"]["code"] == jiguang._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(jiguang.settings, "SMS_MOCK", False)
    monkeypatch.setattr(jiguang.httpx, "post", lambda *a, **k: _OkResp())

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

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


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

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

    for _ in range(3):
        assert jiguang.verify_code(phone, wrong) is False
    assert jiguang.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(jiguang.settings, "SMS_MOCK", False)
    monkeypatch.setattr(jiguang.httpx, "post", lambda *a, **k: _ErrResp())

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

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


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

    jiguang._gc(now)

    assert "stale" not in jiguang._codes and "fresh" in jiguang._codes
    assert "old" not in jiguang._last_sent and "recent" in jiguang._last_sent


# ============================ 用户名 / 默认昵称 ============================

def test_login_assigns_username_and_nickname(client) -> None:
    """新用户创建即分配:11 位纯数字 username(首位非 0/1,与手机号天然区分)+ "用户"+9 位字母数字默认昵称。"""
    phone = "13600136000"
    client.post("/api/v1/auth/sms/send", json={"phone": phone})
    r = client.post("/api/v1/auth/sms/login", json={"phone": phone, "code": "123456"})
    assert r.status_code == 200, r.text
    user = r.json()["user"]

    uname = user["username"]
    assert uname.isdigit() and len(uname) == 11          # 11 位纯数字
    assert uname[0] not in ("0", "1")                     # 无前导 0、与手机号(均以 1 开头)区分

    nick = user["nickname"]
    assert nick.startswith("用户")                          # "用户" 前缀
    suffix = nick.removeprefix("用户")
    assert len(suffix) == 9 and suffix.isalnum()          # 后接 9 位字母+数字


def test_username_stable_and_unique_on_relogin() -> None:
    """同号重登 username 不变(不可变标识);不同号 username 不同(唯一)。
    直连 repository 绕开 HTTP 短信冷却。"""
    from app.db.session import SessionLocal
    from app.repositories import user as user_repo

    db = SessionLocal()
    try:
        a1 = user_repo.upsert_user_for_login(db, phone="13522220001", register_channel="sms")
        uname_a, id_a = a1.username, a1.id
        a2 = user_repo.upsert_user_for_login(db, phone="13522220001", register_channel="sms")
        assert a2.id == id_a and a2.username == uname_a    # 重登:同一行、username 稳定不变

        b = user_repo.upsert_user_for_login(db, phone="13522220002", register_channel="sms")
        assert b.username != uname_a                        # 不同用户、username 唯一
    finally:
        db.close()