guke
143 lines
5.3 KiB
Python
143 lines
5.3 KiB
Python
"""极光→创蓝 容灾 fallback（分派层 send_code / verify_code）。
|
|
|
|
monkeypatch 各 provider 的 send_code/verify_code 为标记函数，不发真网络；
|
|
断言：仅主 503 才转备；429/400 不转；verify try-both 只遍历启用的链。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from app.core.config import settings
|
|
from app.integrations import sms
|
|
from app.integrations.sms import chuanglan, jiguang
|
|
from app.integrations.sms.base import SendResult, SmsError
|
|
|
|
PHONE = "13800138000"
|
|
|
|
|
|
def _raiser(exc: Exception):
|
|
"""返回一个调用即 raise exc 的 send_code 替身（lambda 不能 raise）。"""
|
|
def _f(phone):
|
|
raise exc
|
|
return _f
|
|
|
|
|
|
def test_fallback_provider_defaults_empty():
|
|
assert settings.SMS_FALLBACK_PROVIDER == ""
|
|
|
|
|
|
def test_send_code_primary_success_returns_sendresult(monkeypatch):
|
|
monkeypatch.setattr(settings, "SMS_PROVIDER", "jiguang")
|
|
monkeypatch.setattr(settings, "SMS_FALLBACK_PROVIDER", "")
|
|
monkeypatch.setattr(jiguang, "send_code", lambda phone: 60)
|
|
|
|
r = sms.send_code(PHONE)
|
|
assert isinstance(r, SendResult)
|
|
assert r.cooldown_sec == 60
|
|
assert r.provider == "jiguang"
|
|
assert r.fallback is False
|
|
|
|
|
|
def test_send_code_503_falls_back_to_configured_backup(monkeypatch):
|
|
calls = []
|
|
monkeypatch.setattr(settings, "SMS_PROVIDER", "jiguang")
|
|
monkeypatch.setattr(settings, "SMS_FALLBACK_PROVIDER", "chuanglan")
|
|
monkeypatch.setattr(jiguang, "send_code",
|
|
_raiser(SmsError("极光不可用", status_code=503)))
|
|
monkeypatch.setattr(chuanglan, "send_code",
|
|
lambda phone: (calls.append("chuanglan"), 60)[1])
|
|
|
|
r = sms.send_code(PHONE)
|
|
assert r.provider == "chuanglan"
|
|
assert r.fallback is True
|
|
assert r.cooldown_sec == 60
|
|
assert calls == ["chuanglan"]
|
|
|
|
|
|
@pytest.mark.parametrize("status", [429, 400])
|
|
def test_send_code_non_503_does_not_fall_back(monkeypatch, status):
|
|
calls = []
|
|
monkeypatch.setattr(settings, "SMS_PROVIDER", "jiguang")
|
|
monkeypatch.setattr(settings, "SMS_FALLBACK_PROVIDER", "chuanglan")
|
|
monkeypatch.setattr(jiguang, "send_code",
|
|
_raiser(SmsError("x", status_code=status)))
|
|
monkeypatch.setattr(chuanglan, "send_code",
|
|
lambda phone: (calls.append("chuanglan"), 60)[1])
|
|
|
|
with pytest.raises(SmsError) as ei:
|
|
sms.send_code(PHONE)
|
|
assert ei.value.status_code == status
|
|
assert calls == []
|
|
|
|
|
|
def test_send_code_503_without_backup_reraises(monkeypatch):
|
|
monkeypatch.setattr(settings, "SMS_PROVIDER", "jiguang")
|
|
monkeypatch.setattr(settings, "SMS_FALLBACK_PROVIDER", "")
|
|
monkeypatch.setattr(jiguang, "send_code",
|
|
_raiser(SmsError("极光不可用", status_code=503)))
|
|
|
|
with pytest.raises(SmsError) as ei:
|
|
sms.send_code(PHONE)
|
|
assert ei.value.status_code == 503
|
|
|
|
|
|
def test_send_code_backup_also_fails_raises_backup_error(monkeypatch):
|
|
monkeypatch.setattr(settings, "SMS_PROVIDER", "jiguang")
|
|
monkeypatch.setattr(settings, "SMS_FALLBACK_PROVIDER", "chuanglan")
|
|
monkeypatch.setattr(jiguang, "send_code",
|
|
_raiser(SmsError("极光不可用", status_code=503)))
|
|
monkeypatch.setattr(chuanglan, "send_code",
|
|
_raiser(SmsError("创蓝也不可用", status_code=503)))
|
|
|
|
with pytest.raises(SmsError) as ei:
|
|
sms.send_code(PHONE)
|
|
assert "创蓝" in str(ei.value)
|
|
|
|
|
|
def test_verify_hits_primary_without_touching_backup(monkeypatch):
|
|
calls = []
|
|
monkeypatch.setattr(settings, "SMS_PROVIDER", "jiguang")
|
|
monkeypatch.setattr(settings, "SMS_FALLBACK_PROVIDER", "chuanglan")
|
|
monkeypatch.setattr(jiguang, "verify_code",
|
|
lambda p, c: (calls.append("jiguang"), True)[1])
|
|
monkeypatch.setattr(chuanglan, "verify_code",
|
|
lambda p, c: (calls.append("chuanglan"), True)[1])
|
|
|
|
assert sms.verify_code(PHONE, "123456") is True
|
|
assert calls == ["jiguang"]
|
|
|
|
|
|
def test_verify_falls_through_to_backup(monkeypatch):
|
|
calls = []
|
|
monkeypatch.setattr(settings, "SMS_PROVIDER", "jiguang")
|
|
monkeypatch.setattr(settings, "SMS_FALLBACK_PROVIDER", "chuanglan")
|
|
monkeypatch.setattr(jiguang, "verify_code",
|
|
lambda p, c: (calls.append("jiguang"), False)[1])
|
|
monkeypatch.setattr(chuanglan, "verify_code",
|
|
lambda p, c: (calls.append("chuanglan"), True)[1])
|
|
|
|
assert sms.verify_code(PHONE, "123456") is True
|
|
assert calls == ["jiguang", "chuanglan"]
|
|
|
|
|
|
def test_verify_all_miss_returns_false(monkeypatch):
|
|
monkeypatch.setattr(settings, "SMS_PROVIDER", "jiguang")
|
|
monkeypatch.setattr(settings, "SMS_FALLBACK_PROVIDER", "chuanglan")
|
|
monkeypatch.setattr(jiguang, "verify_code", lambda p, c: False)
|
|
monkeypatch.setattr(chuanglan, "verify_code", lambda p, c: False)
|
|
|
|
assert sms.verify_code(PHONE, "123456") is False
|
|
|
|
|
|
def test_verify_backup_not_touched_when_fallback_off(monkeypatch):
|
|
calls = []
|
|
monkeypatch.setattr(settings, "SMS_PROVIDER", "jiguang")
|
|
monkeypatch.setattr(settings, "SMS_FALLBACK_PROVIDER", "")
|
|
monkeypatch.setattr(jiguang, "verify_code",
|
|
lambda p, c: (calls.append("jiguang"), False)[1])
|
|
monkeypatch.setattr(chuanglan, "verify_code",
|
|
lambda p, c: (calls.append("chuanglan"), True)[1])
|
|
|
|
assert sms.verify_code(PHONE, "123456") is False
|
|
assert calls == ["jiguang"]
|