guke
31bff63ed4
背景 短信验证码已是可切换 provider 架构(极光 / 创蓝)。极光(默认)一旦供应商侧故障(欠费 / 网络 / 服务异常),/sms/send 直接 503 → 用户收不到码、登录中断。本 PR 把极光设为主、创蓝设为备,在极光供应商不可用时自动转创蓝补发,并让后台可区分每次实际走的渠道。 方案(4 个关键决策) # | 决策 | 结论 -- | -- | -- A | fallback 触发范围 | 仅主返回「供应商不可用」(SmsError.status_code == 503:网络 / 余额 / 服务故障)才转备。本地冷却 & 超频(429)、手机号无效(400)不转——不绕过防刷、不为无效号白烧 B | 校验路由 | try-both:极光转创蓝后码在创蓝内存,校验遍历「启用的 fallback 链」(主→备),任一命中即通过;关闭 fallback 时链中只有极光、创蓝零参与 C | 后台可见性 | 成功侧 EVENT_SMS_SEND.details 记 provider / fallback + 分派层日志,风控后台可按号/设备查本次走哪家、是否 fallback D | 默认开关 | SMS_FALLBACK_PROVIDER 默认空=关(保持现状零风险),生产设 chuanglan 开启,置空即秒回退。仅 Mode B(jiguang/chuanglan)互为主备 --------- Co-authored-by: guke <guke@autohome.com.cn> Reviewed-on: #204
54 lines
2.6 KiB
Python
54 lines
2.6 KiB
Python
"""SMS 分派器:按 settings.SMS_PROVIDER 路由到正确 provider。
|
|
|
|
契约:send_code / verify_code **每次调用**读 settings.SMS_PROVIDER 选 provider(支持运行时切换 /
|
|
灰度回退);默认 jiguang。此处 monkeypatch 两 provider 的实现为标记函数,断言路由命中 + 可秒切。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from app.core.config import settings
|
|
from app.integrations import sms
|
|
from app.integrations.sms import aliyun, chuanglan, jiguang
|
|
|
|
|
|
def test_send_code_routes_by_provider_and_switches_per_call(monkeypatch) -> None:
|
|
calls: list[str] = []
|
|
monkeypatch.setattr(jiguang, "send_code", lambda phone: (calls.append("jiguang"), 60)[1])
|
|
monkeypatch.setattr(aliyun, "send_code", lambda phone: (calls.append("aliyun"), 60)[1])
|
|
monkeypatch.setattr(chuanglan, "send_code", lambda phone: (calls.append("chuanglan"), 60)[1])
|
|
|
|
monkeypatch.setattr(settings, "SMS_PROVIDER", "jiguang")
|
|
assert sms.send_code("13800138000").cooldown_sec == 60
|
|
monkeypatch.setattr(settings, "SMS_PROVIDER", "aliyun")
|
|
assert sms.send_code("13800138000").cooldown_sec == 60
|
|
monkeypatch.setattr(settings, "SMS_PROVIDER", "chuanglan")
|
|
assert sms.send_code("13800138000").cooldown_sec == 60
|
|
|
|
assert calls == ["jiguang", "aliyun", "chuanglan"] # 每次按当前 provider 路由,运行时可切
|
|
|
|
|
|
def test_verify_code_routes_by_provider(monkeypatch) -> None:
|
|
calls: list[str] = []
|
|
monkeypatch.setattr(jiguang, "verify_code", lambda phone, code: (calls.append("jiguang"), True)[1])
|
|
monkeypatch.setattr(aliyun, "verify_code", lambda phone, code: (calls.append("aliyun"), True)[1])
|
|
monkeypatch.setattr(chuanglan, "verify_code", lambda phone, code: (calls.append("chuanglan"), True)[1])
|
|
|
|
monkeypatch.setattr(settings, "SMS_PROVIDER", "jiguang")
|
|
assert sms.verify_code("13800138000", "123456") is True
|
|
monkeypatch.setattr(settings, "SMS_PROVIDER", "aliyun")
|
|
assert sms.verify_code("13800138000", "123456") is True
|
|
monkeypatch.setattr(settings, "SMS_PROVIDER", "chuanglan")
|
|
assert sms.verify_code("13800138000", "123456") is True
|
|
|
|
assert calls == ["jiguang", "aliyun", "chuanglan"]
|
|
|
|
|
|
def test_unknown_provider_falls_back_to_jiguang(monkeypatch) -> None:
|
|
"""SMS_PROVIDER 非 aliyun 一律走 jiguang(默认兜底,防误配把登录打挂)。"""
|
|
calls: list[str] = []
|
|
monkeypatch.setattr(jiguang, "send_code", lambda phone: (calls.append("jiguang"), 60)[1])
|
|
monkeypatch.setattr(aliyun, "send_code", lambda phone: (calls.append("aliyun"), 60)[1])
|
|
monkeypatch.setattr(settings, "SMS_PROVIDER", "jiguang")
|
|
|
|
sms.send_code("13800138000")
|
|
assert calls == ["jiguang"]
|