guke
7a2b7cb8ed
现有 sms.py(极光自管码)升级为 app/integrations/sms/ 包:
- base : SmsError + provider 无关的 mock_verify
- jiguang: 原极光自管码逻辑逐字迁入,行为零改动(git 识别为 sms.py 的 rename)
- aliyun : 新增阿里云 dypns 号码认证(Mode A:阿里云生成+下发+校验,核验免费)
- __init__: 按 SMS_PROVIDER 每次调用路由的分派器(默认 jiguang,可秒切回退)
关键决策:
- Mode A:发码 SendSmsVerifyCode(##code## 占位)、校验 CheckSmsVerifyCode(PASS/UNKNOWN);
本服务不再存码 → 消除极光路径「内存存码、多 worker 不共享」技术债。
- 防爆破与极光一致:aliyun 保留 per-phone 失败计数(SMS_MAX_VERIFY_ATTEMPTS),达上限本地作废,
避免两 provider 行为不同致排查困惑(此为 aliyun 路径唯一本地态)。
- 校验降级:阿里云接口异常 → verify_code 抛 SmsError(503),auth 两处 try/except 透出 503(非误报 400)。
- 官方 SDK alibabacloud_dypnsapi20170525;SDK 交互隔离在 _call_send/_call_check(惰性 import + 惰性建
client),单测 monkeypatch 不触真网络。
测试:test_sms_aliyun(17)+ test_sms_dispatch(3)全绿;test_auth 内部访问 retarget 到 jiguang.*。
配置:SMS_PROVIDER + ALIYUN_SMS_*(见 .env.example);文档 docs/integrations/sms.md + aliyun 接口参考。
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
48 lines
2.1 KiB
Python
48 lines
2.1 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, 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(settings, "SMS_PROVIDER", "jiguang")
|
|
assert sms.send_code("13800138000") == 60
|
|
monkeypatch.setattr(settings, "SMS_PROVIDER", "aliyun")
|
|
assert sms.send_code("13800138000") == 60
|
|
|
|
assert calls == ["jiguang", "aliyun"] # 每次按当前 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(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
|
|
|
|
assert calls == ["jiguang", "aliyun"]
|
|
|
|
|
|
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"]
|