zhuzihao
59fe715245
Reviewed-on: #83 Co-authored-by: zhuzihao <zhuzihao@wonderable.ai> Co-committed-by: zhuzihao <zhuzihao@wonderable.ai>
315 lines
12 KiB
Python
315 lines
12 KiB
Python
"""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_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_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
|
|
|
|
|
|
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
|
|
|
|
|
|
# ============================ 用户名 / 默认昵称 ============================
|
|
|
|
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()
|