diff --git a/.env.example b/.env.example index 110e939..f5c5c6f 100644 --- a/.env.example +++ b/.env.example @@ -32,6 +32,13 @@ HEARTBEAT_MONITOR_ENABLED=true HEARTBEAT_TIMEOUT_MINUTES=60 HEARTBEAT_SCAN_INTERVAL_SEC=60 +# ===== OPPO 推送(厂商通道直连,提现审核通过等交易类通知)===== +# OPPO 开放平台注册应用后分配, 服务端鉴权用 AppKey + MasterSecret(≠极光 JG_*)。 +# 缺任一 → oppo_push_configured=False, 推送静默跳过(不影响审核/打款)。 +OPPO_PUSH_APP_KEY= +OPPO_PUSH_MASTER_SECRET= +# 可选(一般不改): OPPO_PUSH_ENABLED=true / OPPO_PUSH_REQUEST_TIMEOUT_SEC=10 + # ===== 短信 (mock 模式) ===== # mock = true 时,任意 6 位数字均通过,且 /sms/send 不真发短信(只 log)。 # 后续接阿里云/腾讯云短信时,改成 false 并填供应商相关 key。 diff --git a/alembic/versions/device_oppo_register_id.py b/alembic/versions/device_oppo_register_id.py new file mode 100644 index 0000000..f1fd038 --- /dev/null +++ b/alembic/versions/device_oppo_register_id.py @@ -0,0 +1,34 @@ +"""device_liveness 加 oppo_register_id 列(OPPO 直连推送目标) + +提现审核通过等交易类通知走 OPPO 官方推送直连(≠极光)。客户端集成 OPPO SDK 后拿到 OPPO +registerID, 经 device/register + heartbeat 上报, 存本列; 与极光 registration_id 并列, +一台 OPPO 设备两套 token 并存。 + +Revision ID: device_oppo_register_id +Revises: comparison_llm_cost +Create Date: 2026-07-15 00:00:00.000000 + +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision: str = 'device_oppo_register_id' +down_revision: Union[str, Sequence[str], None] = 'comparison_llm_cost' +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + with op.batch_alter_table('device_liveness', schema=None) as batch_op: + batch_op.add_column( + sa.Column('oppo_register_id', sa.String(length=128), nullable=True) + ) + + +def downgrade() -> None: + with op.batch_alter_table('device_liveness', schema=None) as batch_op: + batch_op.drop_column('oppo_register_id') diff --git a/app/admin/routers/withdraw.py b/app/admin/routers/withdraw.py index dbc006d..7d9fe9d 100644 --- a/app/admin/routers/withdraw.py +++ b/app/admin/routers/withdraw.py @@ -37,6 +37,7 @@ from app.integrations import wxpay from app.models.admin import AdminUser from app.repositories import app_config from app.repositories import wallet as wallet_repo +from app.services import push_notify router = APIRouter( prefix="/admin/api/withdraws", @@ -315,6 +316,9 @@ def bulk_approve_withdraws( results.append( WithdrawBulkItemResult(out_bill_no=out_bill_no, ok=True, status=order.status) ) + # 审核通过 → OPPO 推送(fire-and-forget, 不阻塞批量审核; status=failed 已退款, 不推) + if order.status != "failed": + push_notify.notify_withdraw_approved_async(order.user_id) except wallet_repo.WithdrawOrderNotFound: db.rollback() results.append( @@ -438,6 +442,9 @@ def approve_withdraw_order( }, ip=get_client_ip(request), commit=True, ) + # 审核通过 → OPPO 推送(fire-and-forget, 不阻塞审核响应; status=failed 是打款失败已退款, 不推) + if order.status != "failed": + push_notify.notify_withdraw_approved_async(order.user_id) return WithdrawOrderOut.model_validate(order) diff --git a/app/api/v1/device.py b/app/api/v1/device.py index 56a5a85..6a2d46f 100644 --- a/app/api/v1/device.py +++ b/app/api/v1/device.py @@ -40,6 +40,7 @@ def register_device( user_id=user.id, device_id=req.device_id, registration_id=req.registration_id, + oppo_register_id=req.oppo_register_id, platform=req.platform, app_version=req.app_version, ) @@ -64,6 +65,7 @@ def report_heartbeat( device_id=req.device_id, accessibility_enabled=req.accessibility_enabled, registration_id=req.registration_id, + oppo_register_id=req.oppo_register_id, ) return OkResponse() diff --git a/app/core/config.py b/app/core/config.py index bf27932..d9c77d0 100644 --- a/app/core/config.py +++ b/app/core/config.py @@ -71,6 +71,28 @@ class Settings(BaseSettings): HEARTBEAT_TIMEOUT_MINUTES: int = 60 # 多久没心跳算掉线(1 小时,避免短暂离线误判被杀) HEARTBEAT_SCAN_INTERVAL_SEC: int = 60 # 扫描周期 + # ===== OPPO 推送(厂商通道直连) ===== + # 直连 OPPO PUSH 服务端 API(api.push.oppomobile.com), 不经极光。凭证来自 OPPO 开放平台 + # 注册应用后分配(≠极光 JG_*)。两步鉴权: 先 /server/v1/auth 用 sha256(app_key+timestamp+ + # master_secret) 换 auth_token(缓存 24h), 再带 auth_token 调 /message/notification/unicast + # 单推。提现审核通过等交易类通知走这条。未配凭证 → oppo_push_configured=False, 发送方静默 + # 跳过(不连累审核/打款)。 + OPPO_PUSH_ENABLED: bool = True # 总开关(配了凭证也能置 False 全局停发) + OPPO_PUSH_APP_KEY: str = "" + OPPO_PUSH_MASTER_SECRET: str = "" + OPPO_PUSH_AUTH_ENDPOINT: str = "https://api.push.oppomobile.com/server/v1/auth" + OPPO_PUSH_UNICAST_ENDPOINT: str = ( + "https://api.push.oppomobile.com/server/v1/message/notification/unicast" + ) + OPPO_PUSH_REQUEST_TIMEOUT_SEC: int = 10 + # 通知点击行为(OPPO click_action_type; 0=启动应用首页)。交易类通知点开进首页即可。 + OPPO_PUSH_CLICK_ACTION_TYPE: int = 0 + + @property + def oppo_push_configured(self) -> bool: + """OPPO 推送凭证是否齐备(app_key + master_secret)。缺任一 → 发送方静默跳过。""" + return bool(self.OPPO_PUSH_APP_KEY and self.OPPO_PUSH_MASTER_SECRET) + # ===== 短信 ===== SMS_MOCK: bool = True SMS_CODE_TTL_SEC: int = 300 diff --git a/app/integrations/oppo_push.py b/app/integrations/oppo_push.py new file mode 100644 index 0000000..6d519c6 --- /dev/null +++ b/app/integrations/oppo_push.py @@ -0,0 +1,176 @@ +"""OPPO 推送服务端直连(厂商通道)。 + +不经极光, 直接调 OPPO PUSH 服务端 API(api.push.oppomobile.com)。提现审核通过等交易类 +通知走这条。凭证来自 OPPO 开放平台(app_key/master_secret, 见 settings.OPPO_PUSH_*)。 + +两步鉴权(OPPO 协议): + 1. POST /server/v1/auth 用 sign=sha256(app_key+timestamp+master_secret) 换 auth_token + (有效期 24h)。token 进程内缓存复用, 到期前自动重取。 + 2. POST /server/v1/message/notification/unicast header 带 auth_token, 按单个 + registration_id(OPPO 的 registerID) 定向推送一条通知栏消息。 + +发送 best-effort: 未配凭证 / 网络失败 / OPPO 返错都抛 OppoPushError, 由调用方(push_notify) +吞掉只记日志, 绝不连累审核/打款主流程。风格对齐 integrations/sms.py(httpx 同步 + 未配降级)。 +""" +from __future__ import annotations + +import hashlib +import json +import logging +import time +from dataclasses import dataclass +from threading import Lock + +import httpx + +from app.core.config import settings + +logger = logging.getLogger("shagua.oppo_push") + +# OPPO target_type 枚举: 2 = 按 registration_id 单推 +_TARGET_TYPE_REGISTRATION_ID = 2 +# auth_token 官方有效期 24h; 提前 1h 视为过期重取, 留刷新余量(避免边界请求踩到刚过期) +_TOKEN_TTL_SEC = 24 * 3600 +_TOKEN_REFRESH_MARGIN_SEC = 3600 + + +class OppoPushError(Exception): + """OPPO 推送失败(未配置 / 鉴权失败 / 网络错误 / OPPO 返错)。调用方 best-effort 吞。""" + + +@dataclass +class _CachedToken: + token: str + expires_at: float # epoch 秒(= 取到 token 的时刻 + 24h) + + +# 进程内存 token 缓存(单 worker 有效; 多 worker 各自缓存, 各自换 token, OPPO 侧无副作用) +_token_cache: _CachedToken | None = None +_token_lock = Lock() + + +def _sign(app_key: str, timestamp_ms: int, master_secret: str) -> str: + """OPPO 鉴权签名: sha256(app_key + timestamp + master_secret) 十六进制小写。""" + raw = f"{app_key}{timestamp_ms}{master_secret}" + return hashlib.sha256(raw.encode("utf-8")).hexdigest() + + +def _parse_oppo_response(resp: httpx.Response, phase: str) -> dict | None: + """解析 OPPO 统一响应 {code, message, data}: code==0 返 data, 否则抛 OppoPushError。""" + try: + body = resp.json() + except Exception as e: + raise OppoPushError( + f"OPPO {phase} 响应非 JSON(http={resp.status_code}): {resp.text[:200]}" + ) from e + if body.get("code") != 0: + raise OppoPushError( + f"OPPO {phase} 失败 code={body.get('code')} message={body.get('message')!r}" + ) + return body.get("data") + + +def _fetch_auth_token() -> str: + """调 /server/v1/auth 换 auth_token。失败抛 OppoPushError。""" + timestamp_ms = int(time.time() * 1000) # 毫秒时间戳(OPPO 允许 ±10 分钟误差) + sign = _sign( + settings.OPPO_PUSH_APP_KEY, timestamp_ms, settings.OPPO_PUSH_MASTER_SECRET + ) + try: + resp = httpx.post( + settings.OPPO_PUSH_AUTH_ENDPOINT, + data={ + "app_key": settings.OPPO_PUSH_APP_KEY, + "sign": sign, + "timestamp": timestamp_ms, + }, + timeout=settings.OPPO_PUSH_REQUEST_TIMEOUT_SEC, + ) + except httpx.HTTPError as e: + raise OppoPushError(f"OPPO auth 网络错误: {e}") from e + + data = _parse_oppo_response(resp, "auth") + token = (data or {}).get("auth_token") + if not token: + raise OppoPushError(f"OPPO auth 未返回 auth_token: {resp.text[:200]}") + return token + + +def _get_auth_token(force_refresh: bool = False) -> str: + """取 auth_token(缓存优先; 到期或 force_refresh 才重取)。线程安全。""" + global _token_cache + now = time.time() + with _token_lock: + cache = _token_cache + if ( + not force_refresh + and cache is not None + and now < cache.expires_at - _TOKEN_REFRESH_MARGIN_SEC + ): + return cache.token + token = _fetch_auth_token() + _token_cache = _CachedToken(token=token, expires_at=now + _TOKEN_TTL_SEC) + return token + + +def send_notification( + registration_id: str, + title: str, + content: str, + *, + click_action_type: int | None = None, +) -> str: + """向单个 OPPO registerID 推一条通知栏消息, 返回 OPPO message_id。 + + 失败(未配置 / 鉴权 / 网络 / OPPO 返错)一律抛 OppoPushError, 由调用方吞。 + unicast 首次失败时强刷一次 token 重试(覆盖 token 被 OPPO 提前失效的情况); 网络错误不重试。 + """ + if not settings.OPPO_PUSH_ENABLED: + raise OppoPushError("OPPO 推送总开关关闭(OPPO_PUSH_ENABLED=false)") + if not settings.oppo_push_configured: + raise OppoPushError("OPPO 推送未配置(缺 OPPO_PUSH_APP_KEY/OPPO_PUSH_MASTER_SECRET)") + if not registration_id: + raise OppoPushError("registration_id 为空") + + if click_action_type is None: + click_action_type = settings.OPPO_PUSH_CLICK_ACTION_TYPE + message = { + "target_type": _TARGET_TYPE_REGISTRATION_ID, + "target_value": registration_id, + "notification": { + "title": title, + "content": content, + "click_action_type": click_action_type, + }, + } + # OPPO unicast body 是 form 编码, message 字段为 JSON 字符串 + payload = {"message": json.dumps(message, ensure_ascii=False)} + + last_err: OppoPushError | None = None + for attempt in range(2): + token = _get_auth_token(force_refresh=(attempt == 1)) + try: + resp = httpx.post( + settings.OPPO_PUSH_UNICAST_ENDPOINT, + data=payload, + headers={"auth_token": token}, + timeout=settings.OPPO_PUSH_REQUEST_TIMEOUT_SEC, + ) + except httpx.HTTPError as e: + # 网络错误与 token 无关, 不强刷重试 + raise OppoPushError(f"OPPO unicast 网络错误: {e}") from e + try: + data = _parse_oppo_response(resp, "unicast") + except OppoPushError as e: + last_err = e + if attempt == 0: + logger.warning("[OPPO] unicast 失败(%s), 强刷 token 重试一次", e) + continue + raise + message_id = (data or {}).get("message_id", "") + logger.info( + "[OPPO] 推送成功 regId=%s… message_id=%s", registration_id[:12], message_id + ) + return message_id + # 理论到不了(第二次失败已 raise); 防御性收尾 + raise last_err or OppoPushError("OPPO unicast 重试后仍失败") diff --git a/app/models/device.py b/app/models/device.py index e1d1261..9c27b80 100644 --- a/app/models/device.py +++ b/app/models/device.py @@ -44,6 +44,9 @@ class DeviceLiveness(Base): device_id: Mapped[str] = mapped_column(String(128), index=True, nullable=False) # 极光推送 registration id;拿到才填(JCollectionAuth 同意后才下发) registration_id: Mapped[str | None] = mapped_column(String(64), nullable=True) + # OPPO 官方推送 SDK(HeytapPushManager)的 registerID; 直连 OPPO 厂商通道用(≠极光 registration_id)。 + # 一台 OPPO 设备两套 token 并存: 极光走极光自有/其它厂商通道, 这个走 OPPO 直连(提现审核通过等交易通知)。 + oppo_register_id: Mapped[str | None] = mapped_column(String(128), nullable=True) platform: Mapped[str] = mapped_column(String(16), nullable=False, default="android") app_version: Mapped[str | None] = mapped_column(String(32), nullable=True) diff --git a/app/repositories/device.py b/app/repositories/device.py index c3e9047..c1bba83 100644 --- a/app/repositories/device.py +++ b/app/repositories/device.py @@ -22,16 +22,18 @@ def register_or_update( user_id: int, device_id: str, registration_id: str | None, + oppo_register_id: str | None = None, platform: str = "android", app_version: str | None = None, ) -> DeviceLiveness: - """注册设备或更新其 registration_id / 元信息。upsert by (user_id, device_id)。""" + """注册设备或更新其 registration_id / oppo_register_id / 元信息。upsert by (user_id, device_id)。""" device = _get(db, user_id=user_id, device_id=device_id) if device is None: device = DeviceLiveness( user_id=user_id, device_id=device_id, registration_id=registration_id, + oppo_register_id=oppo_register_id, platform=platform or "android", app_version=app_version, ) @@ -39,6 +41,8 @@ def register_or_update( else: if registration_id: device.registration_id = registration_id + if oppo_register_id: + device.oppo_register_id = oppo_register_id if platform: device.platform = platform if app_version: @@ -55,11 +59,13 @@ def touch_heartbeat( device_id: str, accessibility_enabled: bool, registration_id: str | None, + oppo_register_id: str | None = None, ) -> DeviceLiveness: """处理一次心跳(心跳也能自注册)。 service 心跳或 accessibility_enabled=true 时,刷新存活并把状态机重置回 alive、 - 清掉 notified_at(掉线恢复 → 下次再断才会再推一条)。 + 清掉 notified_at(掉线恢复 → 下次再断才会再推一条)。registration_id / oppo_register_id + 非空时顺带更新(客户端拿到 push token 后每次心跳带上, 后端最终一致)。 """ now = datetime.now(timezone.utc) device = _get(db, user_id=user_id, device_id=device_id) @@ -69,6 +75,8 @@ def touch_heartbeat( if registration_id: device.registration_id = registration_id + if oppo_register_id: + device.oppo_register_id = oppo_register_id device.last_report_protection_on = accessibility_enabled if accessibility_enabled: @@ -118,6 +126,25 @@ def get_device(db: Session, *, user_id: int, device_id: str) -> DeviceLiveness | return _get(db, user_id=user_id, device_id=device_id) +def list_oppo_register_ids_by_user(db: Session, *, user_id: int) -> list[str]: + """取某用户所有设备已登记的 OPPO registerID(去重保序, 非空)。 + + 提现审核通过推送用: 一个用户可能多台设备, 每台都推; 没 OPPO token 的设备(非 OPPO 机 / + 未集成 OPPO SDK)自然不在列。空列表 = 该用户没有可推的 OPPO 设备。 + """ + stmt = select(DeviceLiveness.oppo_register_id).where( + DeviceLiveness.user_id == user_id, + DeviceLiveness.oppo_register_id.is_not(None), + ) + seen: set[str] = set() + out: list[str] = [] + for rid in db.execute(stmt).scalars().all(): + if rid and rid not in seen: + seen.add(rid) + out.append(rid) + return out + + def ack_kill_alert(db: Session, *, user_id: int, device_id: str) -> None: """客户端已弹过「开启自启动」引导 → 清「待提醒」标记(幂等; 下次真掉线 worker 再置)。""" device = _get(db, user_id=user_id, device_id=device_id) diff --git a/app/schemas/device.py b/app/schemas/device.py index dd7a7c4..c1b5394 100644 --- a/app/schemas/device.py +++ b/app/schemas/device.py @@ -9,6 +9,7 @@ from pydantic import BaseModel, ConfigDict class DeviceRegisterRequest(BaseModel): device_id: str registration_id: str | None = None + oppo_register_id: str | None = None platform: str = "android" app_version: str | None = None @@ -18,6 +19,7 @@ class HeartbeatRequest(BaseModel): source: str = "service" # service | app accessibility_enabled: bool = True registration_id: str | None = None + oppo_register_id: str | None = None class DeviceOut(BaseModel): @@ -26,6 +28,7 @@ class DeviceOut(BaseModel): id: int device_id: str registration_id: str | None + oppo_register_id: str | None ever_protected: bool liveness_state: str last_heartbeat_at: datetime | None diff --git a/app/services/push_notify.py b/app/services/push_notify.py new file mode 100644 index 0000000..0b61e36 --- /dev/null +++ b/app/services/push_notify.py @@ -0,0 +1,74 @@ +"""推送通知编排(查设备 push token → 调厂商推送)。 + +薄编排层: repositories(取 token) + integrations(发送) 的粘合, 供 api / admin router 调用。 +所有发送 best-effort —— 失败只 log, 绝不抛给调用方(不连累审核/打款等主流程)。 + +当前只有 OPPO 直连一条通道(提现审核通过通知)。后续加华为/vivo 等厂商时, 在这里按设备 +token 做通道分发, 上层业务函数(notify_withdraw_approved 等)不感知通道差异。 +""" +from __future__ import annotations + +import logging +import threading + +from sqlalchemy.orm import Session + +from app.db.session import SessionLocal +from app.integrations import oppo_push +from app.integrations.oppo_push import OppoPushError +from app.repositories import device as device_repo + +logger = logging.getLogger("shagua.push_notify") + +_WITHDRAW_APPROVED_TITLE = "提现审核通过" +_WITHDRAW_APPROVED_CONTENT = "您的提现申请已审核通过，款项将很快到账，请留意微信零钱。" + + +def notify_withdraw_approved(db: Session, *, user_id: int) -> int: + """提现审核通过 → 给该用户所有 OPPO 设备各推一条通知。返回成功推送的设备数。 + + best-effort: 未配 OPPO / 无 OPPO 设备 / 单设备发送失败都不抛, 只 log。审核主流程绝不受影响。 + 调用方仍应包一层 try/except 兜底本函数自身的意外异常(如 DB 查询失败)。 + """ + reg_ids = device_repo.list_oppo_register_ids_by_user(db, user_id=user_id) + if not reg_ids: + logger.info("[push] withdraw_approved user_id=%s 无 OPPO 设备, 跳过", user_id) + return 0 + + sent = 0 + for rid in reg_ids: + try: + oppo_push.send_notification( + rid, _WITHDRAW_APPROVED_TITLE, _WITHDRAW_APPROVED_CONTENT + ) + sent += 1 + except OppoPushError as e: + logger.warning( + "[push] withdraw_approved user_id=%s regId=%s… 发送失败: %s", + user_id, rid[:12], e, + ) + logger.info( + "[push] withdraw_approved user_id=%s 推送 %d/%d 台 OPPO 设备", + user_id, sent, len(reg_ids), + ) + return sent + + +def notify_withdraw_approved_async(user_id: int) -> None: + """fire-and-forget 版: 后台守护线程新开 session 发推送, 不阻塞审核响应(对齐 best-effort)。 + + 审核 approve 已同步调微信转账, 推送不该再把外部 API 耗时叠进响应(多设备/OPPO 慢时最坏 + N×timeout)。线程内必须新开 SessionLocal —— 请求级 session 在响应返回后即关, 不能跨线程用。 + 起线程失败也只 log, 绝不连累审核。 + """ + def _run() -> None: + try: + with SessionLocal() as db: + notify_withdraw_approved(db, user_id=user_id) + except Exception: # noqa: BLE001 - 后台线程绝不抛 + logger.warning("[push] withdraw_approved async 异常 user_id=%s", user_id, exc_info=True) + + try: + threading.Thread(target=_run, name=f"oppo-push-{user_id}", daemon=True).start() + except Exception: # noqa: BLE001 - 起线程失败也绝不连累审核 + logger.warning("[push] withdraw_approved async 起线程失败 user_id=%s", user_id, exc_info=True) diff --git a/tests/test_admin_write.py b/tests/test_admin_write.py index ec6d39d..ab439c9 100644 --- a/tests/test_admin_write.py +++ b/tests/test_admin_write.py @@ -476,3 +476,81 @@ def test_withdraw_reconcile(admin_client: TestClient, finance_token: str, monkey ) assert r.status_code == 200, r.text assert "checked" in r.json() and "resolved" in r.json() + + +# ===== 提现审核通过 → OPPO 推送挂钩 ===== + +def test_approve_triggers_oppo_push( + admin_client: TestClient, finance_token: str, monkeypatch +) -> None: + """审核通过(非 failed)后触发一次 OPPO 推送, 带被审核单的 user_id。""" + uid = _seed_user("13900000021") + db = SessionLocal() + try: + db.add(WithdrawOrder( + user_id=uid, out_bill_no="pushapprove0001", amount_cents=100, status="reviewing" + )) + db.commit() + finally: + db.close() + + from app.admin.routers import withdraw as withdraw_router + + # 绕过真实微信转账: approve 直接把单置 pending 返回(不发 wxpay) + def _fake_approve(db, obn): + o = db.execute( + select(WithdrawOrder).where(WithdrawOrder.out_bill_no == obn) + ).scalar_one() + o.status = "pending" + db.commit() + return o + + monkeypatch.setattr(withdraw_router.wallet_repo, "approve_withdraw", _fake_approve) + pushed: list[int] = [] + monkeypatch.setattr( + withdraw_router.push_notify, "notify_withdraw_approved_async", + lambda user_id: pushed.append(user_id), + ) + r = admin_client.post( + "/admin/api/withdraws/pushapprove0001/approve", headers=_auth(finance_token) + ) + assert r.status_code == 200, r.text + assert pushed == [uid] + + +def test_approve_failed_does_not_push( + admin_client: TestClient, finance_token: str, monkeypatch +) -> None: + """打款直接失败(status=failed, 已退款)不推"审核通过"。""" + uid = _seed_user("13900000022") + db = SessionLocal() + try: + db.add(WithdrawOrder( + user_id=uid, out_bill_no="pushapprove0002", amount_cents=100, status="reviewing" + )) + db.commit() + finally: + db.close() + + from app.admin.routers import withdraw as withdraw_router + + def _fake_approve_failed(db, obn): + o = db.execute( + select(WithdrawOrder).where(WithdrawOrder.out_bill_no == obn) + ).scalar_one() + o.status = "failed" + o.fail_reason = "余额不足" + db.commit() + return o + + monkeypatch.setattr(withdraw_router.wallet_repo, "approve_withdraw", _fake_approve_failed) + pushed: list[int] = [] + monkeypatch.setattr( + withdraw_router.push_notify, "notify_withdraw_approved_async", + lambda user_id: pushed.append(user_id), + ) + r = admin_client.post( + "/admin/api/withdraws/pushapprove0002/approve", headers=_auth(finance_token) + ) + assert r.status_code == 200, r.text + assert pushed == [] diff --git a/tests/test_oppo_push.py b/tests/test_oppo_push.py new file mode 100644 index 0000000..5cf5218 --- /dev/null +++ b/tests/test_oppo_push.py @@ -0,0 +1,98 @@ +"""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") diff --git a/tests/test_push_notify.py b/tests/test_push_notify.py new file mode 100644 index 0000000..30fbece --- /dev/null +++ b/tests/test_push_notify.py @@ -0,0 +1,70 @@ +"""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()