shaguabijia-app-server/tests/test_sms_chuanglan.py

"""创蓝云智(253)短信 provider(Mode B 自管码)单元测试。

HTTP 交互隔离在 chuanglan._call_chuanglan(薄封装:签名 + httpx POST + 解析),映射逻辑在
chuanglan._send_via_chuanglan。本文件 monkeypatch 这两个接缝(或更底层 httpx.post)→ 不发真网络。
测的是:HMAC 签名算法、请求体不上行 password、错误码→HTTP 码、自管码存/校验(与极光同语义)、mock 短路。
"""
from __future__ import annotations

import hashlib
import hmac
import json
import time

import httpx
import pytest

from app.core.config import settings
from app.integrations.sms import chuanglan
from app.integrations.sms.base import SmsError

PHONE = "13800138000"


class _FakeResp:
    """极简 httpx.Response 替身:只暴露 status_code / json() / text。"""

    def __init__(self, status_code: int = 200, payload: dict | None = None, text: str = "") -> None:
        self.status_code = status_code
        self._payload = payload if payload is not None else {}
        self.text = text or json.dumps(self._payload)

    def json(self) -> dict:
        return self._payload


def _configure(monkeypatch, *, mock: bool = False) -> None:
    """配齐创蓝凭证 + 设 SMS_MOCK;清本地存码/冷却隔离用例。"""
    monkeypatch.setattr(settings, "SMS_MOCK", mock)
    monkeypatch.setattr(settings, "CHUANGLAN_SMS_ACCOUNT", "YZM0000001")
    monkeypatch.setattr(settings, "CHUANGLAN_SMS_PASSWORD", "secret")
    monkeypatch.setattr(settings, "CHUANGLAN_SMS_TEMPLATE_ID", "1021143438")
    monkeypatch.setattr(settings, "CHUANGLAN_SMS_SIGNATURE", "【创蓝云智】")
    chuanglan._codes.clear()
    chuanglan._last_sent.clear()


def _ok_payload(**over) -> dict:
    p = {
        "code": "000000",
        "msgId": "25071018345400902898000000000001",
        "time": "20250710183454",
        "successNum": "1",
        "failNum": "0",
        "errorMsg": "",
    }
    p.update(over)
    return p


# ============================ 签名算法 ============================

def test_sign_implements_documented_hmac() -> None:
    """_sign = HmacSHA256(key=md5(password), msg=sorted([md5pwd,ts,nonce]) 拼接),小写 hex。"""
    password, ts, nonce = "secret", "1752143733", "0123456789abcdef0123456789abcdef"
    md5pwd = hashlib.md5(password.encode()).hexdigest()
    expected = hmac.new(
        md5pwd.encode(),
        "".join(sorted([md5pwd, ts, nonce])).encode(),
        hashlib.sha256,
    ).hexdigest()

    sig = chuanglan._sign(password, ts, nonce)
    assert sig == expected
    assert len(sig) == 64 and sig == sig.lower()


def test_sign_changes_with_nonce() -> None:
    assert chuanglan._sign("p", "1", "nonceA") != chuanglan._sign("p", "1", "nonceB")


# ============================ _call_chuanglan(HTTP 接缝)============================

def test_call_chuanglan_builds_signed_request_without_password(monkeypatch) -> None:
    _configure(monkeypatch)
    captured: dict = {}

    def fake_post(url, **kw):
        captured.update(url=url, body=kw.get("json"), headers=kw.get("headers"), timeout=kw.get("timeout"))
        return _FakeResp(200, _ok_payload())

    monkeypatch.setattr(httpx, "post", fake_post)

    result = chuanglan._call_chuanglan(PHONE, "123456")

    assert result["code"] == "000000"
    assert captured["url"] == settings.CHUANGLAN_SMS_ENDPOINT
    body = captured["body"]
    assert body["account"] == "YZM0000001"
    assert body["phoneNumbers"] == PHONE
    assert body["templateId"] == "1021143438"
    assert body["templateParamJson"] == json.dumps([{"param1": "123456"}])
    assert body["signature"] == "【创蓝云智】"
    assert "password" not in body            # HMAC 方式:密码只用于算签,不上行
    assert len(body["nonce"]) == 32
    assert captured["headers"]["X-QA-Hmac-Signature"] == chuanglan._sign(
        "secret", body["timestamp"], body["nonce"]
    )
    assert captured["timeout"] == settings.CHUANGLAN_SMS_TIMEOUT_SEC


def test_call_chuanglan_omits_signature_when_blank(monkeypatch) -> None:
    _configure(monkeypatch)
    monkeypatch.setattr(settings, "CHUANGLAN_SMS_SIGNATURE", "")
    captured: dict = {}
    monkeypatch.setattr(httpx, "post", lambda url, **kw: captured.update(body=kw.get("json")) or _FakeResp(200, _ok_payload()))

    chuanglan._call_chuanglan(PHONE, "123456")
    assert "signature" not in captured["body"]   # 模板自带签名时不传


def test_call_chuanglan_http_non_200_raises_503(monkeypatch) -> None:
    _configure(monkeypatch)
    monkeypatch.setattr(httpx, "post", lambda url, **kw: _FakeResp(500, {}, "oops"))
    with pytest.raises(SmsError) as ei:
        chuanglan._call_chuanglan(PHONE, "123456")
    assert ei.value.status_code == 503


def test_call_chuanglan_network_error_raises_503(monkeypatch) -> None:
    _configure(monkeypatch)

    def boom(url, **kw):
        raise httpx.ConnectError("down")

    monkeypatch.setattr(httpx, "post", boom)
    with pytest.raises(SmsError) as ei:
        chuanglan._call_chuanglan(PHONE, "123456")
    assert ei.value.status_code == 503


# ============================ _send_via_chuanglan(错误码映射)============================

@pytest.mark.parametrize(
    "code,expected",
    [
        ("000000", None),   # 成功不抛
        ("103", 429),       # 超频
        ("107", 400),       # 手机号错误
        ("109", 503),       # 无发送量/余额
        ("117", 503),       # IP 未白名单
        ("102", 503),       # 密码错误
        ("116", 503),       # 签名不合法
        ("124", 503),       # 模板内容不匹配
        ("152", 503),       # 模板不存在
        ("999999", 503),    # 未知码兜底
    ],
)
def test_send_via_chuanglan_maps_codes(monkeypatch, code, expected) -> None:
    _configure(monkeypatch)
    monkeypatch.setattr(chuanglan, "_call_chuanglan", lambda phone, c: _ok_payload(code=code, errorMsg=code))
    if expected is None:
        assert chuanglan._send_via_chuanglan(PHONE, "123456") is None
    else:
        with pytest.raises(SmsError) as ei:
            chuanglan._send_via_chuanglan(PHONE, "123456")
        assert ei.value.status_code == expected


def test_send_via_chuanglan_not_configured_raises_503_without_calling(monkeypatch) -> None:
    monkeypatch.setattr(settings, "SMS_MOCK", False)
    monkeypatch.setattr(settings, "CHUANGLAN_SMS_ACCOUNT", "")   # 凭证缺

    def _boom(phone, c):
        raise AssertionError("未配置时不应发起请求")

    monkeypatch.setattr(chuanglan, "_call_chuanglan", _boom)
    with pytest.raises(SmsError) as ei:
        chuanglan._send_via_chuanglan(PHONE, "123456")
    assert ei.value.status_code == 503


# ============================ send_code(自管码,复制自极光)============================

def test_send_code_success_stores_and_returns_cooldown(monkeypatch) -> None:
    _configure(monkeypatch)
    monkeypatch.setattr(chuanglan, "_send_via_chuanglan", lambda p, c: None)

    assert chuanglan.send_code(PHONE) == settings.SMS_SEND_INTERVAL_SEC
    assert PHONE in chuanglan._codes
    assert len(chuanglan._codes[PHONE].code) == settings.SMS_CODE_LENGTH


def test_send_code_cooldown_raises_429(monkeypatch) -> None:
    _configure(monkeypatch)
    monkeypatch.setattr(chuanglan, "_send_via_chuanglan", lambda p, c: None)
    chuanglan.send_code(PHONE)
    with pytest.raises(SmsError) as ei:
        chuanglan.send_code(PHONE)
    assert ei.value.status_code == 429


def test_send_code_failure_keeps_cooldown_drops_code(monkeypatch) -> None:
    _configure(monkeypatch)

    def boom(p, c):
        raise SmsError("no balance", status_code=503)

    monkeypatch.setattr(chuanglan, "_send_via_chuanglan", boom)
    with pytest.raises(SmsError) as ei:
        chuanglan.send_code(PHONE)
    assert ei.value.status_code == 503
    assert PHONE not in chuanglan._codes      # 没发出去的码删掉
    assert PHONE in chuanglan._last_sent       # 冷却保留:失败也限速


def test_send_code_unexpected_error_wrapped_503(monkeypatch) -> None:
    _configure(monkeypatch)

    def boom(p, c):
        raise RuntimeError("boom")

    monkeypatch.setattr(chuanglan, "_send_via_chuanglan", boom)
    with pytest.raises(SmsError) as ei:
        chuanglan.send_code(PHONE)
    assert ei.value.status_code == 503


def test_send_code_mock_short_circuits_no_network(monkeypatch) -> None:
    _configure(monkeypatch, mock=True)

    def boom(p, c):
        raise AssertionError("mock 不应发网络")

    monkeypatch.setattr(chuanglan, "_send_via_chuanglan", boom)
    assert chuanglan.send_code(PHONE) == settings.SMS_SEND_INTERVAL_SEC


# ============================ verify_code(自管码,复制自极光)============================

def test_verify_code_success_is_one_time(monkeypatch) -> None:
    _configure(monkeypatch)
    chuanglan._codes[PHONE] = chuanglan._CodeRecord(code="123456", expires_at=time.time() + 300)
    assert chuanglan.verify_code(PHONE, "123456") is True
    assert chuanglan.verify_code(PHONE, "123456") is False   # 验过即作废


def test_verify_code_wrong_caps_then_invalidates(monkeypatch) -> None:
    _configure(monkeypatch)
    chuanglan._codes[PHONE] = chuanglan._CodeRecord(code="123456", expires_at=time.time() + 300)
    for _ in range(settings.SMS_MAX_VERIFY_ATTEMPTS):
        assert chuanglan.verify_code(PHONE, "000000") is False
    # 达失败上限即作废:即便随后给对的码也 False
    assert chuanglan.verify_code(PHONE, "123456") is False


def test_verify_code_expired_false_and_cleared(monkeypatch) -> None:
    _configure(monkeypatch)
    chuanglan._codes[PHONE] = chuanglan._CodeRecord(code="123456", expires_at=time.time() - 1)
    assert chuanglan.verify_code(PHONE, "123456") is False
    assert PHONE not in chuanglan._codes


def test_verify_code_no_record_false(monkeypatch) -> None:
    _configure(monkeypatch)
    assert chuanglan.verify_code(PHONE, "123456") is False


def test_verify_code_mock_passes_any_ndigit(monkeypatch) -> None:
    _configure(monkeypatch, mock=True)
    assert chuanglan.verify_code(PHONE, "123456") is True    # 6 位数字放行
    assert chuanglan.verify_code(PHONE, "12345") is False    # 位数不对


# ============================ 端点:SMS_PROVIDER=chuanglan 端到端路由 ============================

def test_sms_send_chuanglan_outage_returns_503(client, monkeypatch) -> None:
    """SMS_PROVIDER=chuanglan 且发送时创蓝异常 → /sms/send 返 503(auth.py 现有 try/except 覆盖)。"""
    _configure(monkeypatch)
    monkeypatch.setattr(settings, "SMS_PROVIDER", "chuanglan")

    def boom(p, c):
        raise SmsError("chuanglan down", status_code=503)

    monkeypatch.setattr(chuanglan, "_send_via_chuanglan", boom)
    r = client.post("/api/v1/auth/sms/send", json={"phone": "13812345678"})
    assert r.status_code == 503, r.text