marco
5ae4c474e4
- integrations/oppo_push.py: OPPO PUSH 服务端直连(sha256 鉴权 + auth_token 24h 缓存 + unicast 单推) - device_liveness 加 oppo_register_id 列 + 迁移; register/heartbeat 透传 + list 查询 - services/push_notify: 审核通过 fire-and-forget 推送(best-effort, 不阻塞审核响应) - admin withdraw approve(单笔+批量)挂钩(status=failed 已退款不推) - 单测 10 passed; 真调 OPPO 生产 auth/unicast 验证签名与请求构造 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
99 lines
3.6 KiB
Python
99 lines
3.6 KiB
Python
"""oppo_push 集成层单测:未配降级 / auth+unicast 正常流程 / token 缓存 / OPPO 错误码抛错。
|
|
|
|
httpx 全 monkeypatch, 不发真实网络。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
|
|
import pytest
|
|
|
|
from app.integrations import oppo_push
|
|
from app.integrations.oppo_push import OppoPushError
|
|
|
|
|
|
class _FakeResp:
|
|
def __init__(self, status_code: int, payload: dict) -> None:
|
|
self.status_code = status_code
|
|
self._payload = payload
|
|
self.text = json.dumps(payload)
|
|
|
|
def json(self) -> dict:
|
|
return self._payload
|
|
|
|
|
|
def _configure(monkeypatch) -> None:
|
|
"""配齐凭证 + 清 token 缓存(避免跨用例污染)。"""
|
|
monkeypatch.setattr(oppo_push.settings, "OPPO_PUSH_ENABLED", True)
|
|
monkeypatch.setattr(oppo_push.settings, "OPPO_PUSH_APP_KEY", "ak")
|
|
monkeypatch.setattr(oppo_push.settings, "OPPO_PUSH_MASTER_SECRET", "ms")
|
|
oppo_push._token_cache = None
|
|
|
|
|
|
def test_not_configured_raises(monkeypatch) -> None:
|
|
monkeypatch.setattr(oppo_push.settings, "OPPO_PUSH_APP_KEY", "")
|
|
monkeypatch.setattr(oppo_push.settings, "OPPO_PUSH_MASTER_SECRET", "")
|
|
oppo_push._token_cache = None
|
|
with pytest.raises(OppoPushError):
|
|
oppo_push.send_notification("regid", "t", "c")
|
|
|
|
|
|
def test_disabled_raises(monkeypatch) -> None:
|
|
_configure(monkeypatch)
|
|
monkeypatch.setattr(oppo_push.settings, "OPPO_PUSH_ENABLED", False)
|
|
with pytest.raises(OppoPushError):
|
|
oppo_push.send_notification("regid", "t", "c")
|
|
|
|
|
|
def test_success_flow(monkeypatch) -> None:
|
|
_configure(monkeypatch)
|
|
calls: list[tuple] = []
|
|
|
|
def _fake_post(url, **kw):
|
|
calls.append((url, kw))
|
|
if url == oppo_push.settings.OPPO_PUSH_AUTH_ENDPOINT:
|
|
return _FakeResp(200, {"code": 0, "message": "success", "data": {"auth_token": "TOK"}})
|
|
return _FakeResp(200, {"code": 0, "message": "success", "data": {"message_id": "MID"}})
|
|
|
|
monkeypatch.setattr(oppo_push.httpx, "post", _fake_post)
|
|
mid = oppo_push.send_notification("regid123", "标题", "内容")
|
|
assert mid == "MID"
|
|
# 先 auth 再 unicast, unicast header 带 auth_token
|
|
assert calls[0][0] == oppo_push.settings.OPPO_PUSH_AUTH_ENDPOINT
|
|
assert calls[1][0] == oppo_push.settings.OPPO_PUSH_UNICAST_ENDPOINT
|
|
assert calls[1][1]["headers"]["auth_token"] == "TOK"
|
|
# unicast body 的 message 是 JSON 字符串, 含 target_value / notification
|
|
msg = json.loads(calls[1][1]["data"]["message"])
|
|
assert msg["target_value"] == "regid123"
|
|
assert msg["notification"]["title"] == "标题"
|
|
assert msg["notification"]["content"] == "内容"
|
|
|
|
|
|
def test_auth_token_cached(monkeypatch) -> None:
|
|
_configure(monkeypatch)
|
|
auth_hits: list[int] = []
|
|
|
|
def _fake_post(url, **kw):
|
|
if url == oppo_push.settings.OPPO_PUSH_AUTH_ENDPOINT:
|
|
auth_hits.append(1)
|
|
return _FakeResp(200, {"code": 0, "data": {"auth_token": "TOK"}})
|
|
return _FakeResp(200, {"code": 0, "data": {"message_id": "MID"}})
|
|
|
|
monkeypatch.setattr(oppo_push.httpx, "post", _fake_post)
|
|
oppo_push.send_notification("r", "t", "c")
|
|
oppo_push.send_notification("r", "t", "c")
|
|
assert len(auth_hits) == 1 # 第二次复用缓存 token, 不再换 auth
|
|
|
|
|
|
def test_oppo_error_raises(monkeypatch) -> None:
|
|
_configure(monkeypatch)
|
|
|
|
def _fake_post(url, **kw):
|
|
if url == oppo_push.settings.OPPO_PUSH_AUTH_ENDPOINT:
|
|
return _FakeResp(200, {"code": 0, "data": {"auth_token": "TOK"}})
|
|
return _FakeResp(200, {"code": -2, "message": "invalid target"})
|
|
|
|
monkeypatch.setattr(oppo_push.httpx, "post", _fake_post)
|
|
with pytest.raises(OppoPushError):
|
|
oppo_push.send_notification("r", "t", "c")
|