guke
ed76820e97
读完了两份 spec + 分派器,这分支的全貌清晰了。下面是可直接粘进 Gitea PR 的 MR 说明(标题 + 正文): 建议标题 feat(sms): 短信验证码可切换多 provider(极光/阿里云/创蓝),默认极光零改动 MR 正文(复制以下 markdown): ## 概述 把原单文件 `app/integrations/sms.py` 重构为**可切换 provider 包** `app/integrations/sms/`,在保留极光(默认、行为零改动)的基础上,新增两家验证码短信 provider: - **阿里云号码认证 dypns**(Mode A:阿里云生成/存储/校验验证码,核验免费) - **创蓝云智 253**(Mode B:本服务自管码,httpx 直连 + HMAC 签名) Provider 由 `SMS_PROVIDER` 按调用实时选择,默认 `jiguang`。短信=花钱 + 登录关键路径,故新 provider **opt-in、可灰度、秒级回退**,默认路径零变更。 ## 为什么 现有极光路径本地内存存码(多 worker 不共享,已是技术债),且单一供应商无法灰度/切换。引入 provider 抽象后:阿里云托管码可消除存码债,创蓝作为备选降低单点依赖,三家随配置切换与回退。 ## 改动内容 **架构(`app/integrations/sms/`)** | 文件 | 说明 | |---|---| | `__init__.py` | 对外仍暴露 `send_code/verify_code/SmsError`(auth 导入不变);按 `SMS_PROVIDER` **每次调用**分派;未知值回退 `jiguang` | | `base.py` | `SmsError`(status_code→HTTP) + provider 无关的 `mock_verify` | | `jiguang.py` | 原 `sms.py` 逻辑**原样迁入**,行为零改动(git 识别为 rename) | | `aliyun.py` | 新增,Mode A:`SendSmsVerifyCode` + `CheckSmsVerifyCode`,惰性加载 SDK | | `chuanglan.py` | 新增,Mode B:自管码 + `tpl/send` + HMAC 签名 | **两种验证码模式** - Mode A(阿里云):不本地存码,阿里云 `##code##` 托管生成+校验;本地仅留 per-phone 失败计数防爆破。 - Mode B(极光/创蓝):`secrets` 生成 N 位 → 进程内存 → 供应商只下发;本地一次性校验 + 失败 N 次作废。创蓝**复制**极光存码机器(不重构极光,零回归风险)。 **配置(`config.py` + `.env.example`)** - `SMS_PROVIDER = jiguang | aliyun | chuanglan`(默认 jiguang) - `ALIYUN_SMS_*`(AK/签名/模板/方案名/时长…) + `aliyun_sms_configured` 门控 - `CHUANGLAN_SMS_*`(账号/密码/模板/签名/endpoint…) + `chuanglan_sms_configured` 门控 - 复用现有 `SMS_MOCK / SMS_CODE_LENGTH / SMS_CODE_TTL_SEC / SMS_SEND_INTERVAL_SEC / SMS_MAX_VERIFY_ATTEMPTS` - 切到某 provider 却未配齐 → `send_code` 抛 `SmsError(503)`,不静默 **auth.py(最小改动)** - `verify_code` 现在可能抛 `SmsError`(阿里云降级 503)→ `sms_login`、`wechat_bind_phone_sms` 两处各包 `try/except SmsError → HTTPException`,与 `send_code` 现有写法一致。 **依赖** - `pyproject.toml` 增 `alibabacloud_dypnsapi20170525`(仅阿里云 provider 惰性 import;jiguang/chuanglan 不加载)。创蓝零新依赖(httpx + 标准库)。 **测试** - 新增 `test_sms_aliyun.py` / `test_sms_chuanglan.py`(均 monkeypatch 网络接缝,不发真短信) + `test_sms_dispatch.py`(分派/回退)。 - `test_auth.py` 相应更新。 - 现有测试走 `SMS_MOCK=true` 在分派层短路,不受影响。 **文档** - 设计 spec:`docs/superpowers/specs/2026-07-25-aliyun-sms-verify-design.md`、`2026-07-26-chuanglan-sms-verify-design.md` - 接口调研:`docs/integrations/aliyun/*`、`docs/integrations/chuanglan/tpl-send.md`、`docs/integrations/sms.md` ## 兼容性 & 回退 - **默认 `SMS_PROVIDER=jiguang`,线上行为与现状完全一致**;不改极光逻辑、不动 API 层频控与测试账号短路。 - 切阿里云/创蓝仅改环境变量,出问题秒切回极光;未知 `SMS_PROVIDER` 一律回退极光,防误配打挂登录。 --------- Co-authored-by: guke <guke@autohome.com.cn> Reviewed-on: #188
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") == 60
|
|
monkeypatch.setattr(settings, "SMS_PROVIDER", "aliyun")
|
|
assert sms.send_code("13800138000") == 60
|
|
monkeypatch.setattr(settings, "SMS_PROVIDER", "chuanglan")
|
|
assert sms.send_code("13800138000") == 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"]
|