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>
71 lines
2.1 KiB
Python
71 lines
2.1 KiB
Python
"""push_notify.notify_withdraw_approved 单测:多 OPPO 设备全推 / 无设备返 0 / 发送失败吞错。"""
|
|
from __future__ import annotations
|
|
|
|
from app.db.session import SessionLocal
|
|
from app.integrations.oppo_push import OppoPushError
|
|
from app.repositories import device as device_repo
|
|
from app.repositories import user as user_repo
|
|
from app.services import push_notify
|
|
|
|
|
|
def _seed_user(phone: str) -> int:
|
|
db = SessionLocal()
|
|
try:
|
|
return user_repo.upsert_user_for_login(db, phone=phone, register_channel="sms").id
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def _add_oppo_device(user_id: int, device_id: str, oppo_reg: str) -> None:
|
|
db = SessionLocal()
|
|
try:
|
|
device_repo.register_or_update(
|
|
db, user_id=user_id, device_id=device_id,
|
|
registration_id=None, oppo_register_id=oppo_reg,
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def test_notify_no_oppo_device_returns_zero() -> None:
|
|
uid = _seed_user("13911100001")
|
|
db = SessionLocal()
|
|
try:
|
|
assert push_notify.notify_withdraw_approved(db, user_id=uid) == 0
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def test_notify_pushes_all_oppo_devices(monkeypatch) -> None:
|
|
uid = _seed_user("13911100002")
|
|
_add_oppo_device(uid, "dev_a", "OPPO_REG_A")
|
|
_add_oppo_device(uid, "dev_b", "OPPO_REG_B")
|
|
sent: list[str] = []
|
|
monkeypatch.setattr(
|
|
push_notify.oppo_push, "send_notification",
|
|
lambda rid, title, content, **kw: sent.append(rid) or "msgid",
|
|
)
|
|
db = SessionLocal()
|
|
try:
|
|
n = push_notify.notify_withdraw_approved(db, user_id=uid)
|
|
finally:
|
|
db.close()
|
|
assert n == 2
|
|
assert set(sent) == {"OPPO_REG_A", "OPPO_REG_B"}
|
|
|
|
|
|
def test_notify_swallows_send_error(monkeypatch) -> None:
|
|
uid = _seed_user("13911100003")
|
|
_add_oppo_device(uid, "dev_c", "OPPO_REG_C")
|
|
|
|
def _boom(rid, title, content, **kw):
|
|
raise OppoPushError("simulated failure")
|
|
|
|
monkeypatch.setattr(push_notify.oppo_push, "send_notification", _boom)
|
|
db = SessionLocal()
|
|
try:
|
|
# 发送失败被吞, 不抛; 成功数为 0
|
|
assert push_notify.notify_withdraw_approved(db, user_id=uid) == 0
|
|
finally:
|
|
db.close()
|