diff --git a/.env.example b/.env.example index 9ffc4a0..b77fd01 100644 --- a/.env.example +++ b/.env.example @@ -175,3 +175,9 @@ PANGLE_REPORT_SITE_ID_TEST=5832303 # APPLOG_MAX_BATCH=500 # 单批最大条数(超 → 422;导入期常量,改需重启) # APPLOG_MAX_BODY_BYTES=2097152 # 请求体上限 2MB(超 → 413;运行期可调) # APPLOG_MAX_MSG_BYTES=8192 # 单条 msg 超此字节数截断 + +# ===== 华为 Push ===== +HUAWEI_PUSH_APP_ID= +HUAWEI_PUSH_APP_SECRET= +# 0=正式消息(默认)；1=测试消息(仅开发联调，生产必须保持 0) +HUAWEI_PUSH_TARGET_USER_TYPE=0 diff --git a/app/core/config.py b/app/core/config.py index e020307..f35f2e9 100644 --- a/app/core/config.py +++ b/app/core/config.py @@ -88,6 +88,8 @@ class Settings(BaseSettings): # (OAuth 换 token 时 client_id 即 AppId)。发送走 v1 messages:send,成功码 80000000。 HUAWEI_PUSH_APP_ID: str = "" HUAWEI_PUSH_APP_SECRET: str = "" + # 0=正式消息(默认，受正式消息频控)；1=测试消息(仅开发联调，勿用于生产)。 + HUAWEI_PUSH_TARGET_USER_TYPE: int = Field(default=0, ge=0, le=1) HUAWEI_PUSH_TOKEN_ENDPOINT: str = "https://oauth-login.cloud.huawei.com/oauth2/v3/token" HUAWEI_PUSH_SEND_ENDPOINT_TEMPLATE: str = ( "https://push-api.cloud.huawei.com/v1/{app_id}/messages:send" diff --git a/app/integrations/vendor_push.py b/app/integrations/vendor_push.py index e7167b1..900b8dc 100644 --- a/app/integrations/vendor_push.py +++ b/app/integrations/vendor_push.py @@ -31,6 +31,7 @@ from app.core.config import settings logger = logging.getLogger("shagua.vendor_push") TYPE_ACCESSIBILITY_DISABLED = "accessibility_disabled" +DATA_EVENT_NOTIFICATION_CREATED = "notification_created" SUPPORTED_VENDORS = frozenset({"honor", "huawei", "vivo", "xiaomi", "oppo"}) # vendor key → 中文名(测试/配置状态接口展示用) @@ -194,6 +195,49 @@ def send_accessibility_disabled( ) +def send_data_event( + push_vendor: str, + push_token: str, + *, + event: str, + notification_id: str, + mock: bool = False, +) -> dict[str, Any]: + """发送无界面的轻量事件；当前只允许站内消息创建事件。""" + vendor = normalize_vendor(push_vendor) + token = push_token.strip() if push_token else "" + if not vendor or vendor not in SUPPORTED_VENDORS: + raise VendorPushError(f"unsupported push vendor: {push_vendor}") + if not token: + raise VendorPushError("push token is empty") + if event != DATA_EVENT_NOTIFICATION_CREATED: + raise VendorPushError(f"unsupported data event: {event}") + if not notification_id: + raise VendorPushError("notification_id is empty") + + payload = {"event": event, "notificationId": str(notification_id)} + if mock: + return {"mock": True, "vendor": vendor, "payload": payload} + + # vivo 的通知已设置 foregroundShow=false：App 在前台时必走 + # onForegroundMessageArrived，正好就是本事件需要的刷新信号；不重复占用一次推送配额。 + if vendor == "vivo": + return {"skipped": True, "reason": "foreground notification callback"} + + # OPush 当前只支持通知栏消息，没有服务端透传单推接口。 + # 客户端在 OPPO 首页可见时轮询未读数；这里必须安全跳过，不能请求不存在的 + # /message/transparent/unicast（该地址会稳定返回 HTTP 404）。 + if vendor == "oppo": + return {"skipped": True, "reason": "oppo does not support data messages"} + + dispatch: dict[str, Callable[[str, dict[str, str]], dict[str, Any]]] = { + "honor": _send_honor_data, + "huawei": _send_huawei_data, + "xiaomi": _send_xiaomi_data, + } + return dispatch[vendor](token, payload) + + def _require(value: str, name: str) -> str: if not value: raise VendorPushError(f"{name} not configured") @@ -441,9 +485,15 @@ def _honor_access_token() -> str: def _send_honor(token: str, title: str, body: str, extras: dict[str, str]) -> dict[str, Any]: app_id = _require(settings.HONOR_PUSH_APP_ID, "HONOR_PUSH_APP_ID") access_token = _honor_access_token() + click_action: dict[str, Any] + if extras.get("notificationId"): + # type=3 只负责打开首页，不保证 data 会变成目标 Activity 的 extras。消息中心通知必须 + # 用自定义页面(type=1)+ intent URI，把 notificationId/type/feedbackId 等直接带给 + # MainActivity；否则华为/荣耀系统能展示通知，但用户点击后客户端收不到任何导航参数。 + click_action = {"type": 1, "intent": _click_intent_uri(extras)} + else: + click_action = {"type": 3} payload = { - # clickAction type=3(打开应用首页)时,荣耀点击会把 data JSON 的键值对注入启动 intent 的 - # extras(与 HMS 同机制)→ MainActivity.consumeNavTarget 读 notif_id/notif_type 直达落地。 "data": json.dumps(_click_extras(extras), ensure_ascii=False), "notification": {"title": title, "body": body}, "android": { @@ -452,7 +502,7 @@ def _send_honor(token: str, title: str, body: str, extras: dict[str, str]) -> di "notification": { "title": title, "body": body, - "clickAction": {"type": 3}, + "clickAction": click_action, "importance": "NORMAL", }, }, @@ -476,6 +526,25 @@ def _send_honor(token: str, title: str, body: str, extras: dict[str, str]) -> di return data +def _send_honor_data(token: str, payload: dict[str, str]) -> dict[str, Any]: + app_id = _require(settings.HONOR_PUSH_APP_ID, "HONOR_PUSH_APP_ID") + access_token = _honor_access_token() + data = _request_json( + "POST", + settings.HONOR_PUSH_SEND_ENDPOINT_TEMPLATE.format(app_id=app_id), + json={"data": json.dumps(payload, ensure_ascii=False), "token": [token]}, + headers={ + "Content-Type": "application/json; charset=UTF-8", + "Authorization": f"Bearer {access_token}", + "timestamp": str(int(time.time() * 1000)), + }, + ) + code = data.get("code") + if code is not None and int(code) != 200: + raise VendorPushError(f"honor data push failed: {data}") + return data + + def _huawei_access_token() -> str: """华为 OAuth2 client_credentials 换 access_token(client_id 即 AGC 应用的 AppId)。""" cache_key = "huawei" @@ -507,18 +576,24 @@ def _send_huawei(token: str, title: str, body: str, extras: dict[str, str]) -> d '80100000' 为部分成功(单 token 场景仍视为失败,错误里带原始响应便于排障)。""" app_id = _require(settings.HUAWEI_PUSH_APP_ID, "HUAWEI_PUSH_APP_ID") access_token = _huawei_access_token() + click_action: dict[str, Any] + if extras.get("notificationId"): + # type=3 只打开首页，Mate 20/EMUI 不会把 message.data 自动拆成启动 Intent extras。 + # type=1 的自定义 intent 才能稳定携带反馈记录 id，且冷启动/onNewIntent 都走同一路由。 + click_action = {"type": 1, "intent": _click_intent_uri(extras)} + else: + click_action = {"type": 3} payload = { "validate_only": False, "message": { - # click_action type=3(打开应用首页)时,HMS 点击会把 data JSON 的键值对注入启动 intent - # 的 extras → MainActivity.consumeNavTarget 读 notif_id/notif_type 直达落地。 "data": json.dumps(_click_extras(extras), ensure_ascii=False), "android": { + "target_user_type": settings.HUAWEI_PUSH_TARGET_USER_TYPE, "ttl": f"{settings.PUSH_TIME_TO_LIVE_SEC}s", "notification": { "title": title, "body": body, - "click_action": {"type": 3}, + "click_action": click_action, "importance": "NORMAL", }, }, @@ -541,6 +616,29 @@ def _send_huawei(token: str, title: str, body: str, extras: dict[str, str]) -> d return data +def _send_huawei_data(token: str, payload: dict[str, str]) -> dict[str, Any]: + app_id = _require(settings.HUAWEI_PUSH_APP_ID, "HUAWEI_PUSH_APP_ID") + access_token = _huawei_access_token() + data = _request_json( + "POST", + settings.HUAWEI_PUSH_SEND_ENDPOINT_TEMPLATE.format(app_id=app_id), + json={ + "validate_only": False, + "message": { + "data": json.dumps(payload, ensure_ascii=False), + "token": [token], + }, + }, + headers={ + "Content-Type": "application/json; charset=UTF-8", + "Authorization": f"Bearer {access_token}", + }, + ) + if str(data.get("code", "")) != "80000000": + raise VendorPushError(f"huawei data push failed: {data}") + return data + + def _vivo_auth_token() -> str: cache_key = "vivo" cached = _cache_get(cache_key) @@ -584,15 +682,22 @@ def _send_vivo(token: str, title: str, body: str, extras: dict[str, str]) -> dic "timeToLive": settings.PUSH_TIME_TO_LIVE_SEC, "requestId": uuid.uuid4().hex, "pushMode": settings.VIVO_PUSH_MODE, + # vivo SDK 4.1.5 只有在前台展示关闭时才调用 + # OpenClientPushMessageReceiver.onForegroundMessageArrived。 + # App 在该回调中刷新服务端未读数并补发一条本地通知；后台/锁屏时仍由 + # vivo 系统展示，因而任何场景都只会出现一条通知。 + "foregroundShow": False, + # vivo 官方 VPush 角标字段：离线收到通知时先由桌面自动 +1； + # App 启动/同步后再由统一未读数通过系统 API 精确校准。 + "addBadge": True, "clientCustomMap": extras, } - # 点击落地:消息中心推送(带 notificationId)→ skipType=4 + skipContent=intent uri,由 vivo - # 系统直启 MainActivity 并携带 S. extras(与小米 notify_effect=2 同机制)。不依赖客户端 - # VivoPushReceiver.onNotificationMessageClicked 里的后台 startActivity——Android 10+ BAL - # 会静默拦掉,receiver 路径仅作兜底。无 notificationId 的召回类保持 skipType=1 仅打开首页。 + # vivo Push SDK 480+ 已不再回调 skipType=3；必须使用 skipType=4 的完整 Intent URI。 + # URI 同时包含 data/scheme、显式 component 和 S. extras，OriginOS 才会把业务参数 + # 原样交给 MainActivity（仅靠隐式 deeplink 会打开 App，但可能剥掉 extras）。 if extras.get("notificationId"): payload["skipType"] = 4 - payload["skipContent"] = _click_intent_uri(extras) + payload["skipContent"] = _vivo_click_intent_uri(extras) else: payload["skipType"] = 1 if settings.VIVO_PUSH_CATEGORY: @@ -608,6 +713,25 @@ def _send_vivo(token: str, title: str, body: str, extras: dict[str, str]) -> dic "authToken": auth_token, }, ) + # 10089 = 当前 vivo 应用尚未开通 VPush「离线自动角标」能力。 + # 角标只是通知中心未读数的镜像，不能因此让整条通知发送失败：去掉 addBadge + # 原样重试，通知到达/应用同步后仍由客户端系统 API 按统一未读数精确设置。 + if int(data.get("result", -1)) == 10089: + logger.warning("vivo VPush badge not enabled (10089); retrying without addBadge") + fallback_payload = dict(payload) + fallback_payload.pop("addBadge", None) + fallback_payload["requestId"] = uuid.uuid4().hex + data = _request_json( + "POST", + settings.VIVO_PUSH_SEND_ENDPOINT, + json=fallback_payload, + headers={ + "Content-Type": "application/json", + "authToken": auth_token, + }, + ) + if int(data.get("result", -1)) == 0: + data = {**data, "badgeFallback": True} if int(data.get("result", -1)) != 0: raise VendorPushError(f"vivo push failed: {_raw_log_summary(data)}") return data @@ -660,6 +784,28 @@ def _send_xiaomi(token: str, title: str, body: str, extras: dict[str, str]) -> d return data +def _send_xiaomi_data(token: str, payload: dict[str, str]) -> dict[str, Any]: + app_secret = _require(settings.XIAOMI_PUSH_APP_SECRET, "XIAOMI_PUSH_APP_SECRET") + data = _request_form( + "POST", + settings.XIAOMI_PUSH_SEND_ENDPOINT, + data={ + "registration_id": token, + "restricted_package_name": settings.ANDROID_PACKAGE_NAME, + "payload": json.dumps(payload, ensure_ascii=False), + "pass_through": "1", + "time_to_live": str(settings.PUSH_TIME_TO_LIVE_SEC * 1000), + }, + headers={"Authorization": f"key={app_secret}"}, + ) + code = data.get("code") + if code not in (0, "0", None): + raise VendorPushError(f"xiaomi data push failed: {data}") + if str(data.get("result", "ok")).lower() not in ("ok", "success"): + raise VendorPushError(f"xiaomi data push failed: {data}") + return data + + def _click_extras(extras: dict[str, str]) -> dict[str, str]: """点击落地参数:消息中心推送(extras 带 notificationId)补 notif_id/notif_type 别名—— 客户端 MainActivity.consumeNavTarget 首选这两个键(厂商 receiver 路径的历史约定),原始键 @@ -693,6 +839,24 @@ def _click_intent_uri(extras: dict[str, str]) -> str: return ";".join(parts) +def _vivo_click_intent_uri(extras: dict[str, str]) -> str: + """vivo skipType=4 使用官方要求的完整 intent deeplink 形态。 + + `?#Intent` 中的问号不能省；OriginOS 对缺少它的 URI 会退化为“仅打开应用”， + 不把 S. 参数交给目标 Activity。 + """ + pkg = settings.ANDROID_PACKAGE_NAME + parts = [ + "intent://push/detail?#Intent", + "scheme=shaguabijia", + f"component={pkg}/{pkg}.MainActivity", + "launchFlags=0x14000000", + ] + parts += [f"S.{key}={quote(str(value), safe='')}" for key, value in _click_extras(extras).items()] + parts.append("end") + return ";".join(parts) + + def _xiaomi_template_param(title: str, alert: str) -> str: rendered = ( settings.XIAOMI_PUSH_TEMPLATE_PARAM_JSON @@ -753,6 +917,11 @@ def _send_oppo(token: str, title: str, body: str, extras: dict[str, str]) -> dic "off_line_ttl": ttl_hours, "action_parameters": json.dumps(_click_extras(extras), ensure_ascii=False), } + if extras.get("notificationId"): + # 只有已落库的站内消息才计入角标；保活提醒等临时推送不应留下无法消除的未读数。 + # 客户端打开/已读后会按服务端真实未读总数覆盖，避免长期漂移。 + notification["badge_operation_type"] = 1 + notification["badge_message_count"] = 1 # 点击落地:OPPO SDK 没有点击回调,参数只能靠服务端点击动作配置送达——action_parameters 的 # 键值对仅在 click_action_type=1/4 时才会注入目标 Activity 的 intent extras(type=0「启动应用」 # 会忽略它,extras 全丢 → 点了没反应,与小米 notify_effect=1 同款坑)。 diff --git a/app/services/notification_events.py b/app/services/notification_events.py index 434fced..a4d4f23 100644 --- a/app/services/notification_events.py +++ b/app/services/notification_events.py @@ -136,9 +136,17 @@ def _push_to_user_devices(db: Session, row: Notification, push_vars: dict[str, s ) continue try: - vendor_push.send_notification( + response = vendor_push.send_notification( vendor, dev.push_token, title=title, body=body, extras=extras ) + if vendor == "huawei" and row.type in {"feedback_reply", "feedback_reward"}: + # 华为反馈推送联调日志：保留厂商返回码/requestId 等排障信息， + # 请求本身的 token、Authorization 和应用密钥不会进入 response。 + logger.info( + "huawei feedback push response user_id=%s type=%s " + "notification_id=%s response=%s", + row.user_id, row.type, row.id, response, + ) logger.info( "push sent user_id=%s type=%s vendor=%s notification_id=%s", row.user_id, row.type, vendor, row.id, @@ -147,6 +155,19 @@ def _push_to_user_devices(db: Session, row: Notification, push_vars: dict[str, s logger.warning( "push failed user_id=%s type=%s vendor=%s: %s", row.user_id, row.type, vendor, e ) + try: + vendor_push.send_data_event( + vendor, + dev.push_token, + event=vendor_push.DATA_EVENT_NOTIFICATION_CREATED, + notification_id=str(row.id), + ) + except vendor_push.VendorPushError as e: + # 透传只负责前台铃铛实时刷新，失败不能影响通知栏消息或站内消息。 + logger.warning( + "push data event failed user_id=%s type=%s vendor=%s: %s", + row.user_id, row.type, vendor, e, + ) except Exception: # noqa: BLE001 — 渲染/查设备等意外失败同样不外抛 logger.exception("push notification failed user_id=%s type=%s", row.user_id, row.type) diff --git a/tests/test_device_push.py b/tests/test_device_push.py index a35ec24..82f0569 100644 --- a/tests/test_device_push.py +++ b/tests/test_device_push.py @@ -3,6 +3,7 @@ from __future__ import annotations import json from datetime import datetime, timedelta, timezone +import pytest from fastapi.testclient import TestClient from app.api.v1 import device as device_api @@ -112,9 +113,73 @@ def test_vivo_auth_and_send_payload(monkeypatch) -> None: body = calls[1]["json"] assert body["regId"] == "vivo-regid" assert body["pushMode"] == vendor_push.settings.VIVO_PUSH_MODE + assert body["foregroundShow"] is False + assert body["addBadge"] is True assert body["clientCustomMap"] == {"type": "accessibility_disabled"} +def test_vivo_retries_without_add_badge_when_capability_is_not_enabled(monkeypatch) -> None: + vendor_push._token_cache.clear() + calls: list[dict] = [] + + def _fake_request(method, url, **kwargs): # noqa: ANN001 + calls.append({"method": method, "url": url, **kwargs}) + if url == vendor_push.settings.VIVO_PUSH_AUTH_ENDPOINT: + return _Resp({"result": 0, "authToken": "vivo-auth"}) + if len(calls) == 2: + return _Resp({"result": 10089, "desc": "VPUSH角标功能没开通"}) + return _Resp({"result": 0, "taskId": "vivo-task"}) + + monkeypatch.setattr(vendor_push.settings, "VIVO_PUSH_APP_ID", "106072775") + monkeypatch.setattr(vendor_push.settings, "VIVO_PUSH_APP_KEY", "vivo-key") + monkeypatch.setattr(vendor_push.settings, "VIVO_PUSH_APP_SECRET", "vivo-secret") + monkeypatch.setattr(vendor_push.httpx, "request", _fake_request) + + data = vendor_push.send_accessibility_disabled("vivo", "vivo-regid") + + assert data["taskId"] == "vivo-task" + assert data["badgeFallback"] is True + assert calls[1]["json"]["addBadge"] is True + assert calls[1]["json"]["foregroundShow"] is False + assert "addBadge" not in calls[2]["json"] + assert calls[2]["json"]["foregroundShow"] is False + assert calls[1]["json"]["requestId"] != calls[2]["json"]["requestId"] + + +def test_vivo_notification_click_uses_explicit_intent_uri(monkeypatch) -> None: + """SDK 480+ 需 skipType=4；显式组件与清栈标志保证冷/热启动均携参落地。""" + vendor_push._token_cache.clear() + calls: list[dict] = [] + + def _fake_request(method, url, **kwargs): # noqa: ANN001 + calls.append({"method": method, "url": url, **kwargs}) + if url == vendor_push.settings.VIVO_PUSH_AUTH_ENDPOINT: + return _Resp({"result": 0, "authToken": "vivo-auth"}) + return _Resp({"result": 0, "taskId": "vivo-task"}) + + monkeypatch.setattr(vendor_push.settings, "VIVO_PUSH_APP_ID", "106072775") + monkeypatch.setattr(vendor_push.settings, "VIVO_PUSH_APP_KEY", "vivo-key") + monkeypatch.setattr(vendor_push.settings, "VIVO_PUSH_APP_SECRET", "vivo-secret") + monkeypatch.setattr(vendor_push.httpx, "request", _fake_request) + + vendor_push.send_notification( + "vivo", + "vivo-regid", + title="提现失败", + body="点击查看", + extras={"type": "withdraw_failed", "notificationId": "10"}, + ) + + body = calls[1]["json"] + intent_uri = body["skipContent"] + assert body["skipType"] == 4 + assert "component=com.jishisongfu.shaguabijia/com.jishisongfu.shaguabijia.MainActivity" in intent_uri + assert "launchFlags=0x14000000" in intent_uri + assert "S.notificationId=10" in intent_uri + assert "S.notif_id=10" in intent_uri + assert "S.notif_type=withdraw_failed" in intent_uri + + def test_oppo_auth_and_send_payload(monkeypatch) -> None: vendor_push._token_cache.clear() calls: list[dict] = [] @@ -138,10 +203,71 @@ def test_oppo_auth_and_send_payload(monkeypatch) -> None: assert calls[1]["data"]["auth_token"] == "oppo-auth" assert message["target_type"] == 2 assert message["target_value"] == "oppo-regid" + assert "badge_operation_type" not in message["notification"] + assert "badge_message_count" not in message["notification"] assert json.loads(message["notification"]["action_parameters"]) == { "type": "accessibility_disabled" } + vendor_push.send_notification( + "oppo", + "oppo-regid", + title="提现失败", + body="点击查看", + extras={"type": "withdraw_failed", "notificationId": "10"}, + ) + notification = json.loads(calls[2]["data"]["message"])["notification"] + assert notification["badge_operation_type"] == 1 + assert notification["badge_message_count"] == 1 + + +def test_data_event_dispatches_silent_payload_to_all_vendors(monkeypatch) -> None: + calls: list[tuple[str, str, dict[str, str]]] = [] + + def _capture(vendor: str): + def _send(token: str, payload: dict[str, str]) -> dict: + calls.append((vendor, token, payload)) + return {"ok": True} + + return _send + + monkeypatch.setattr(vendor_push, "_send_honor_data", _capture("honor")) + monkeypatch.setattr(vendor_push, "_send_huawei_data", _capture("huawei")) + monkeypatch.setattr(vendor_push, "_send_xiaomi_data", _capture("xiaomi")) + for vendor in ("honor", "huawei", "xiaomi", "oppo", "vivo"): + result = vendor_push.send_data_event( + vendor, + f"{vendor}-token", + event=vendor_push.DATA_EVENT_NOTIFICATION_CREATED, + notification_id="10", + ) + if vendor in {"oppo", "vivo"}: + assert result["skipped"] is True + + expected_payload = {"event": "notification_created", "notificationId": "10"} + assert calls == [ + ("honor", "honor-token", expected_payload), + ("huawei", "huawei-token", expected_payload), + ("xiaomi", "xiaomi-token", expected_payload), + ] + + +def test_data_event_rejects_unknown_or_missing_notification() -> None: + with pytest.raises(vendor_push.VendorPushError): + vendor_push.send_data_event( + "oppo", + "token", + event="other", + notification_id="10", + ) + with pytest.raises(vendor_push.VendorPushError): + vendor_push.send_data_event( + "oppo", + "token", + event=vendor_push.DATA_EVENT_NOTIFICATION_CREATED, + notification_id="", + ) + def test_honor_auth_and_send_payload(monkeypatch) -> None: vendor_push._token_cache.clear() diff --git a/tests/test_push_center.py b/tests/test_push_center.py index 585f8dc..3108857 100644 --- a/tests/test_push_center.py +++ b/tests/test_push_center.py @@ -86,8 +86,13 @@ def test_huawei_auth_and_send_payload(monkeypatch) -> None: assert calls[1]["headers"]["Authorization"] == "Bearer hw-access" message = calls[1]["json"]["message"] assert message["token"] == ["hw-token"] + assert message["android"]["target_user_type"] == 1 assert message["android"]["notification"]["title"] == "测试标题" - assert message["android"]["notification"]["click_action"] == {"type": 3} + click_action = message["android"]["notification"]["click_action"] + assert click_action["type"] == 1 + assert "component=com.jishisongfu.shaguabijia/com.jishisongfu.shaguabijia.MainActivity" in click_action["intent"] + assert "S.notif_id=90001" in click_action["intent"] + assert "S.notif_type=withdraw_success" in click_action["intent"] assert json.loads(message["data"]) == { "type": "withdraw_success", "notificationId": "90001", @@ -96,6 +101,61 @@ def test_huawei_auth_and_send_payload(monkeypatch) -> None: } +def test_huawei_feedback_click_intent_keeps_highlight_id(monkeypatch) -> None: + """华为消息点击必须把 feedbackId 放进自定义 intent，不能只留在 message.data。""" + vendor_push._token_cache.clear() + calls: list[dict] = [] + + def _fake_request(method, url, **kwargs): # noqa: ANN001 + calls.append({"method": method, "url": url, **kwargs}) + if url == vendor_push.settings.HUAWEI_PUSH_TOKEN_ENDPOINT: + return _Resp({"access_token": "hw-access", "expires_in": 3600}) + return _Resp({"code": "80000000", "msg": "Success"}) + + monkeypatch.setattr(vendor_push.settings, "HUAWEI_PUSH_APP_ID", "10086001") + monkeypatch.setattr(vendor_push.settings, "HUAWEI_PUSH_APP_SECRET", "hw-secret") + monkeypatch.setattr(vendor_push.settings, "HUAWEI_PUSH_TARGET_USER_TYPE", 1) + monkeypatch.setattr(vendor_push.httpx, "request", _fake_request) + + vendor_push.send_notification( + "huawei", + "hw-token", + title="反馈奖励已到账", + body="4金币已到账", + extras={ + "type": "feedback_reward", + "notificationId": "90002", + "feedbackId": "3002", + }, + ) + + click_action = calls[1]["json"]["message"]["android"]["notification"]["click_action"] + assert click_action["type"] == 1 + assert "S.feedbackId=3002" in click_action["intent"] + assert "S.notif_type=feedback_reward" in click_action["intent"] + + +def test_huawei_recall_without_notification_id_still_opens_home(monkeypatch) -> None: + """无站内消息 id 的旧召回通知维持 type=3 首页行为。""" + vendor_push._token_cache.clear() + calls: list[dict] = [] + + def _fake_request(method, url, **kwargs): # noqa: ANN001 + calls.append({"method": method, "url": url, **kwargs}) + if url == vendor_push.settings.HUAWEI_PUSH_TOKEN_ENDPOINT: + return _Resp({"access_token": "hw-access", "expires_in": 3600}) + return _Resp({"code": "80000000", "msg": "Success"}) + + monkeypatch.setattr(vendor_push.settings, "HUAWEI_PUSH_APP_ID", "10086001") + monkeypatch.setattr(vendor_push.settings, "HUAWEI_PUSH_APP_SECRET", "hw-secret") + monkeypatch.setattr(vendor_push.httpx, "request", _fake_request) + + vendor_push.send_accessibility_disabled("huawei", "hw-token") + + click_action = calls[1]["json"]["message"]["android"]["notification"]["click_action"] + assert click_action == {"type": 3} + + def test_huawei_non_success_code_raises(monkeypatch) -> None: vendor_push._token_cache.clear() @@ -479,6 +539,61 @@ def test_push_test_real_send_xiaomi(client: TestClient, monkeypatch, _no_vendor_ assert json.loads(captured["data"]["payload"]) == {"type": "report_approved"} +@pytest.mark.parametrize( + ("type_key", "target_key", "target_value"), + [ + ("withdraw_success", None, None), + ("withdraw_failed", None, None), + ("feedback_reply", "feedbackId", "3001"), + ("feedback_reward", "feedbackId", "3002"), + ("report_approved", "reportId", "4001"), + ], +) +def test_xiaomi_business_notification_click_carries_landing_extras( + monkeypatch, + type_key: str, + target_key: str | None, + target_value: str | None, +) -> None: + captured: dict = {} + + def _fake_request(method, url, **kwargs): # noqa: ANN001 + captured.update(method=method, url=url, **kwargs) + return _Resp({"code": 0, "result": "ok", "data": {"id": "xm-click"}}) + + extras = { + "type": type_key, + "notificationId": "90017", + } + if target_key and target_value: + extras[target_key] = target_value + + monkeypatch.setattr(vendor_push.settings, "XIAOMI_PUSH_APP_SECRET", "xiaomi-secret") + monkeypatch.setattr(vendor_push.httpx, "request", _fake_request) + + vendor_push.send_notification( + "xiaomi", + "xm-regid-click", + title="测试通知", + body="测试点击落地", + extras=extras, + ) + + form = captured["data"] + assert form["extra.notify_effect"] == "2" + intent_uri = form["extra.intent_uri"] + assert ( + "component=com.jishisongfu.shaguabijia/" + "com.jishisongfu.shaguabijia.MainActivity" + ) in intent_uri + assert "S.notificationId=90017" in intent_uri + assert "S.notif_id=90017" in intent_uri + assert f"S.type={type_key}" in intent_uri + assert f"S.notif_type={type_key}" in intent_uri + if target_key and target_value: + assert f"S.{target_key}={target_value}" in intent_uri + + def test_push_test_real_send_vendor_error_maps_502( client: TestClient, monkeypatch, _no_vendor_creds ) -> None: