marco
0921cb84d9
- unicast notification 带 channel_id(默认 push_oplus_category_content, OPPO 系统内容渠道) - category 做成 config 可配(默认空); 升级 service 渠道+ACCOUNT 分类只改配置不改代码 - 客户端无需改动(OPPO 系统预置渠道 App 不自建) Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
102 lines
3.8 KiB
Python
102 lines
3.8 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"] == "内容"
|
|
# channel_id 必带(默认 shagua_wallet); category 默认空 → 不带
|
|
assert msg["notification"]["channel_id"] == "shagua_wallet"
|
|
assert "category" not in msg["notification"]
|
|
|
|
|
|
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")
|