diff --git a/tests/test_device_push.py b/tests/test_device_push.py index 82f0569..a35ec24 100644 --- a/tests/test_device_push.py +++ b/tests/test_device_push.py @@ -3,7 +3,6 @@ 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 @@ -113,73 +112,9 @@ 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] = [] @@ -203,71 +138,10 @@ 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 1a3f498..9b44ac1 100644 --- a/tests/test_push_center.py +++ b/tests/test_push_center.py @@ -84,13 +84,8 @@ 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"] == "测试标题" - 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 message["android"]["notification"]["click_action"] == {"type": 3} assert json.loads(message["data"]) == { "type": "withdraw_success", "notificationId": "90001", @@ -99,61 +94,6 @@ 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() @@ -444,61 +384,6 @@ 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: