左辰勇
d6016c12f9
2024-11-20 后创建的 OPPO 应用发送通知必须携带 category(新规), 配置项 OPPO_PUSH_CHANNEL_ID / OPPO_PUSH_CATEGORY / OPPO_PUSH_NOTIFY_LEVEL(0=不传), 均留空时 payload 与原先完全一致。凭据已到位:荣耀/小米(含 channel 154219)/OPPO/vivo 四家配齐(.env,不入库),仅剩华为待补。 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
448 lines
18 KiB
Python
448 lines
18 KiB
Python
"""厂商推送(5 家)+ 推送测试三件套。
|
|
|
|
覆盖:华为 Push Kit 发送链路(OAuth + messages:send payload)、send_notification 通用入口
|
|
与 mock 模式、/push/vendors 配置状态、/push/templates 模板渲染、/push/test 的
|
|
mock/真发/变量覆盖/站内联动/设备反查/参数校验。厂商 HTTP 全部 monkeypatch,不真发。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.integrations import vendor_push
|
|
|
|
_ALL_VENDOR_SETTINGS = [key for keys in vendor_push.REQUIRED_SETTINGS.values() for key in keys]
|
|
|
|
|
|
class _Resp:
|
|
status_code = 200
|
|
text = "{}"
|
|
|
|
def __init__(self, data: dict) -> None:
|
|
self._data = data
|
|
|
|
def json(self) -> dict:
|
|
return self._data
|
|
|
|
|
|
@pytest.fixture()
|
|
def _no_vendor_creds(monkeypatch) -> None:
|
|
"""把 5 家厂商凭据全部清空(隔离本机 .env 里已填的真实密钥,保证用例确定性)。"""
|
|
for key in _ALL_VENDOR_SETTINGS:
|
|
monkeypatch.setattr(vendor_push.settings, key, "")
|
|
|
|
|
|
def _login(client: TestClient, phone: str) -> str:
|
|
client.post("/api/v1/auth/sms/send", json={"phone": phone})
|
|
r = client.post("/api/v1/auth/sms/login", json={"phone": phone, "code": "123456"})
|
|
assert r.status_code == 200, r.text
|
|
return r.json()["access_token"]
|
|
|
|
|
|
def _auth(token: str) -> dict[str, str]:
|
|
return {"Authorization": f"Bearer {token}"}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# integrations.vendor_push:华为链路 + 通用入口
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_huawei_auth_and_send_payload(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.HUAWEI_PUSH_TOKEN_ENDPOINT:
|
|
return _Resp({"access_token": "hw-access", "expires_in": 3600})
|
|
return _Resp({"code": "80000000", "msg": "Success", "requestId": "req-1"})
|
|
|
|
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)
|
|
|
|
data = vendor_push.send_notification(
|
|
"huawei",
|
|
"hw-token",
|
|
title="测试标题",
|
|
body="测试内容",
|
|
extras={"type": "withdraw_success", "notificationId": "90001"},
|
|
)
|
|
|
|
assert data["code"] == "80000000"
|
|
# OAuth:client_id 即 AppId
|
|
assert calls[0]["url"] == vendor_push.settings.HUAWEI_PUSH_TOKEN_ENDPOINT
|
|
assert calls[0]["data"]["grant_type"] == "client_credentials"
|
|
assert calls[0]["data"]["client_id"] == "10086001"
|
|
# 发送:v1 messages:send,Bearer 鉴权,token 数组 + data 透传 extras
|
|
assert calls[1]["url"].endswith("/v1/10086001/messages:send")
|
|
assert calls[1]["headers"]["Authorization"] == "Bearer hw-access"
|
|
message = calls[1]["json"]["message"]
|
|
assert message["token"] == ["hw-token"]
|
|
assert message["android"]["notification"]["title"] == "测试标题"
|
|
assert message["android"]["notification"]["click_action"] == {"type": 3}
|
|
assert json.loads(message["data"]) == {"type": "withdraw_success", "notificationId": "90001"}
|
|
|
|
|
|
def test_huawei_non_success_code_raises(monkeypatch) -> None:
|
|
vendor_push._token_cache.clear()
|
|
|
|
def _fake_request(method, url, **kwargs): # noqa: ANN001
|
|
if url == vendor_push.settings.HUAWEI_PUSH_TOKEN_ENDPOINT:
|
|
return _Resp({"access_token": "hw-access", "expires_in": 3600})
|
|
return _Resp({"code": "80300007", "msg": "all tokens are invalid"})
|
|
|
|
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)
|
|
|
|
with pytest.raises(vendor_push.VendorPushError, match="huawei push failed"):
|
|
vendor_push.send_notification("huawei", "bad-token", title="t", body="b")
|
|
|
|
|
|
def test_vendor_aliases_normalize() -> None:
|
|
assert vendor_push.normalize_vendor("华为") == "huawei"
|
|
assert vendor_push.normalize_vendor("HMS") == "huawei"
|
|
assert vendor_push.normalize_vendor("荣耀") == "honor"
|
|
assert vendor_push.normalize_vendor("小米") == "xiaomi"
|
|
assert vendor_push.SUPPORTED_VENDORS == {"honor", "huawei", "xiaomi", "oppo", "vivo"}
|
|
|
|
|
|
def test_send_notification_mock_skips_http(monkeypatch) -> None:
|
|
def _boom(*args, **kwargs): # noqa: ANN001, ANN002, ANN003
|
|
raise AssertionError("mock 模式不应发起任何 HTTP 请求")
|
|
|
|
monkeypatch.setattr(vendor_push.httpx, "request", _boom)
|
|
|
|
data = vendor_push.send_notification(
|
|
"oppo", "any-token", title="标题", body="正文", extras={"type": "push_test"}, mock=True
|
|
)
|
|
assert data == {
|
|
"mock": True,
|
|
"vendor": "oppo",
|
|
"title": "标题",
|
|
"body": "正文",
|
|
"extras": {"type": "push_test"},
|
|
}
|
|
|
|
|
|
def test_send_notification_rejects_unknown_vendor() -> None:
|
|
with pytest.raises(vendor_push.VendorPushError, match="unsupported push vendor"):
|
|
vendor_push.send_notification("nokia", "t", title="a", body="b", mock=True)
|
|
with pytest.raises(vendor_push.VendorPushError, match="token is empty"):
|
|
vendor_push.send_notification("huawei", " ", title="a", body="b", mock=True)
|
|
|
|
|
|
def test_accessibility_wrapper_keeps_legacy_extras(monkeypatch) -> None:
|
|
"""旧入口 send_accessibility_disabled 仍传 {"type":"accessibility_disabled"}(worker 兼容)。"""
|
|
captured: dict = {}
|
|
|
|
def _fake_request(method, url, **kwargs): # noqa: ANN001
|
|
captured.update(method=method, url=url, **kwargs)
|
|
return _Resp({"code": 0, "result": "ok"})
|
|
|
|
monkeypatch.setattr(vendor_push.settings, "XIAOMI_PUSH_APP_SECRET", "xiaomi-secret")
|
|
monkeypatch.setattr(vendor_push.settings, "XIAOMI_PUSH_TEMPLATE_PARAM_JSON", "")
|
|
monkeypatch.setattr(vendor_push.settings, "XIAOMI_PUSH_TEMPLATE_TITLE", "")
|
|
monkeypatch.setattr(vendor_push.settings, "XIAOMI_PUSH_TEMPLATE_DESCRIPTION", "")
|
|
monkeypatch.setattr(vendor_push.httpx, "request", _fake_request)
|
|
|
|
vendor_push.send_accessibility_disabled("xiaomi", "regid-1")
|
|
|
|
assert json.loads(captured["data"]["payload"]) == {"type": "accessibility_disabled"}
|
|
|
|
|
|
def test_oppo_payload_includes_new_message_category(monkeypatch) -> None:
|
|
"""OPPO 新消息分类:配置了 channel_id/category 时随通知体下发(2024-11 新规必带)。"""
|
|
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.OPPO_PUSH_AUTH_ENDPOINT:
|
|
return _Resp({"code": 0, "data": {"auth_token": "oppo-auth"}})
|
|
return _Resp({"code": 0, "data": {"message_id": "oppo-msg"}})
|
|
|
|
monkeypatch.setattr(vendor_push.settings, "OPPO_PUSH_APP_KEY", "oppo-key")
|
|
monkeypatch.setattr(vendor_push.settings, "OPPO_PUSH_MASTER_SECRET", "oppo-master")
|
|
monkeypatch.setattr(vendor_push.settings, "OPPO_PUSH_CHANNEL_ID", "push_oplus_category_content")
|
|
monkeypatch.setattr(vendor_push.settings, "OPPO_PUSH_CATEGORY", "MARKETING")
|
|
monkeypatch.setattr(vendor_push.settings, "OPPO_PUSH_NOTIFY_LEVEL", 0)
|
|
monkeypatch.setattr(vendor_push.httpx, "request", _fake_request)
|
|
|
|
vendor_push.send_notification(
|
|
"oppo", "oppo-regid", title="标题", body="正文", extras={"type": "push_test"}
|
|
)
|
|
|
|
notification = json.loads(calls[1]["data"]["message"])["notification"]
|
|
assert notification["channel_id"] == "push_oplus_category_content"
|
|
assert notification["category"] == "MARKETING"
|
|
assert "notify_level" not in notification # 0=不传,走 OPPO 默认
|
|
|
|
|
|
def test_oppo_payload_omits_category_when_unconfigured(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.OPPO_PUSH_AUTH_ENDPOINT:
|
|
return _Resp({"code": 0, "data": {"auth_token": "oppo-auth"}})
|
|
return _Resp({"code": 0, "data": {"message_id": "oppo-msg"}})
|
|
|
|
monkeypatch.setattr(vendor_push.settings, "OPPO_PUSH_APP_KEY", "oppo-key")
|
|
monkeypatch.setattr(vendor_push.settings, "OPPO_PUSH_MASTER_SECRET", "oppo-master")
|
|
monkeypatch.setattr(vendor_push.settings, "OPPO_PUSH_CHANNEL_ID", "")
|
|
monkeypatch.setattr(vendor_push.settings, "OPPO_PUSH_CATEGORY", "")
|
|
monkeypatch.setattr(vendor_push.settings, "OPPO_PUSH_NOTIFY_LEVEL", 0)
|
|
monkeypatch.setattr(vendor_push.httpx, "request", _fake_request)
|
|
|
|
vendor_push.send_notification(
|
|
"oppo", "oppo-regid", title="标题", body="正文", extras={"type": "push_test"}
|
|
)
|
|
|
|
notification = json.loads(calls[1]["data"]["message"])["notification"]
|
|
assert "channel_id" not in notification
|
|
assert "category" not in notification
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# /api/v1/push 三件套
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_vendors_status_reports_missing_keys(client: TestClient, monkeypatch, _no_vendor_creds) -> None:
|
|
monkeypatch.setattr(vendor_push.settings, "XIAOMI_PUSH_APP_SECRET", "xiaomi-secret")
|
|
token = _login(client, "13900011001")
|
|
|
|
r = client.get("/api/v1/push/vendors", headers=_auth(token))
|
|
assert r.status_code == 200
|
|
vendors = {v["vendor"]: v for v in r.json()["vendors"]}
|
|
assert list(vendors) == ["honor", "huawei", "xiaomi", "oppo", "vivo"]
|
|
|
|
assert vendors["xiaomi"]["configured"] is True
|
|
assert vendors["xiaomi"]["missingKeys"] == []
|
|
assert vendors["huawei"]["configured"] is False
|
|
assert vendors["huawei"]["missingKeys"] == ["HUAWEI_PUSH_APP_ID", "HUAWEI_PUSH_APP_SECRET"]
|
|
assert vendors["honor"]["label"] == "荣耀"
|
|
assert vendors["vivo"]["missingKeys"] == [
|
|
"VIVO_PUSH_APP_ID", "VIVO_PUSH_APP_KEY", "VIVO_PUSH_APP_SECRET",
|
|
]
|
|
|
|
|
|
def test_templates_render_all_13_types(client: TestClient) -> None:
|
|
token = _login(client, "13900011002")
|
|
r = client.get("/api/v1/push/templates", headers=_auth(token))
|
|
assert r.status_code == 200
|
|
templates = r.json()["templates"]
|
|
assert len(templates) == 13
|
|
|
|
by_type = {t["type"]: t for t in templates}
|
|
ws = by_type["withdraw_success"]
|
|
assert ws["pushTitle"] == "提现到账提醒"
|
|
assert ws["pushBodySample"] == "¥0.50已存入您的微信钱包，点击查看到账详情"
|
|
assert ws["variables"] == ["amount"]
|
|
|
|
expiring = by_type["reward_expiring"]
|
|
assert "86金币" in expiring["pushBodySample"]
|
|
assert "{coins}" in expiring["pushBodyTemplate"]
|
|
assert expiring["sampleVars"]["cash"] == "12.80"
|
|
|
|
# 权限类标题按类型写死功能名
|
|
assert by_type["perm_overlay"]["pushTitle"] == "检测到您的比价按钮已失效"
|
|
|
|
|
|
def test_push_test_mock_renders_template(client: TestClient, _no_vendor_creds) -> None:
|
|
token = _login(client, "13900011003")
|
|
r = client.post(
|
|
"/api/v1/push/test",
|
|
json={"vendor": "华为", "pushToken": "hw-token-1", "type": "withdraw_success"},
|
|
headers=_auth(token),
|
|
)
|
|
assert r.status_code == 200, r.text
|
|
body = r.json()
|
|
assert body["ok"] is True
|
|
assert body["mock"] is True
|
|
assert body["vendor"] == "huawei" # 中文别名已归一化
|
|
assert body["title"] == "提现到账提醒"
|
|
assert body["body"] == "¥0.50已存入您的微信钱包，点击查看到账详情"
|
|
assert body["extras"] == {"type": "withdraw_success"}
|
|
assert body["missingKeys"] == ["HUAWEI_PUSH_APP_ID", "HUAWEI_PUSH_APP_SECRET"]
|
|
assert body["vendorResponse"] is None
|
|
|
|
|
|
def test_push_test_vars_override(client: TestClient, _no_vendor_creds) -> None:
|
|
token = _login(client, "13900011004")
|
|
r = client.post(
|
|
"/api/v1/push/test",
|
|
json={
|
|
"vendor": "xiaomi",
|
|
"pushToken": "xm-1",
|
|
"type": "invite_order_reward",
|
|
"vars": {"nickname": "小王", "amount": "6.66"},
|
|
},
|
|
headers=_auth(token),
|
|
)
|
|
assert r.status_code == 200
|
|
assert r.json()["body"] == "您的好友「小王」完成首次下单，6.66元现金已到账"
|
|
|
|
|
|
def test_push_test_generic_copy_without_type(client: TestClient, _no_vendor_creds) -> None:
|
|
token = _login(client, "13900011005")
|
|
r = client.post(
|
|
"/api/v1/push/test",
|
|
json={"vendor": "oppo", "pushToken": "op-1"},
|
|
headers=_auth(token),
|
|
)
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
assert body["extras"]["type"] == "push_test"
|
|
assert "OPPO" in body["body"]
|
|
|
|
|
|
def test_push_test_create_notification_links_message_center(
|
|
client: TestClient, _no_vendor_creds
|
|
) -> None:
|
|
token = _login(client, "13900011006")
|
|
before = client.get("/api/v1/notifications/unread-count", headers=_auth(token)).json()["count"]
|
|
|
|
r = client.post(
|
|
"/api/v1/push/test",
|
|
json={
|
|
"vendor": "vivo",
|
|
"pushToken": "vv-1",
|
|
"type": "feedback_reward",
|
|
"createNotification": True,
|
|
},
|
|
headers=_auth(token),
|
|
)
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
nid = body["notificationId"]
|
|
assert isinstance(nid, int)
|
|
assert body["extras"]["notificationId"] == str(nid)
|
|
assert body["extras"]["type"] == "feedback_reward"
|
|
assert body["extras"]["feedbackId"] # 业务参数一并带上,客户端可直达反馈详情
|
|
|
|
# 站内多了一条未读;按 push extras 的 id 置读 → 闭环
|
|
after = client.get("/api/v1/notifications/unread-count", headers=_auth(token)).json()["count"]
|
|
assert after == before + 1
|
|
r = client.post("/api/v1/notifications/read", json={"ids": [nid]}, headers=_auth(token))
|
|
assert r.json()["markedCount"] == 1
|
|
|
|
|
|
def test_push_test_real_send_requires_credentials(client: TestClient, _no_vendor_creds) -> None:
|
|
token = _login(client, "13900011007")
|
|
r = client.post(
|
|
"/api/v1/push/test",
|
|
json={"vendor": "huawei", "pushToken": "hw-1", "mock": False},
|
|
headers=_auth(token),
|
|
)
|
|
assert r.status_code == 400
|
|
assert "HUAWEI_PUSH_APP_ID" in r.json()["detail"]
|
|
|
|
|
|
def test_push_test_real_send_xiaomi(client: TestClient, monkeypatch, _no_vendor_creds) -> 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-real"}})
|
|
|
|
monkeypatch.setattr(vendor_push.settings, "XIAOMI_PUSH_APP_SECRET", "xiaomi-secret")
|
|
monkeypatch.setattr(vendor_push.httpx, "request", _fake_request)
|
|
token = _login(client, "13900011008")
|
|
|
|
r = client.post(
|
|
"/api/v1/push/test",
|
|
json={
|
|
"vendor": "xiaomi",
|
|
"pushToken": "xm-regid-9",
|
|
"type": "report_approved",
|
|
"mock": False,
|
|
},
|
|
headers=_auth(token),
|
|
)
|
|
assert r.status_code == 200, r.text
|
|
body = r.json()
|
|
assert body["mock"] is False
|
|
assert body["missingKeys"] == []
|
|
assert body["vendorResponse"]["data"]["id"] == "xm-real"
|
|
assert captured["data"]["registration_id"] == "xm-regid-9"
|
|
assert captured["data"]["title"] == "爆料审核通过"
|
|
assert "蜀大侠火锅" in captured["data"]["description"]
|
|
assert json.loads(captured["data"]["payload"]) == {"type": "report_approved"}
|
|
|
|
|
|
def test_push_test_real_send_vendor_error_maps_502(
|
|
client: TestClient, monkeypatch, _no_vendor_creds
|
|
) -> None:
|
|
def _fake_request(method, url, **kwargs): # noqa: ANN001
|
|
return _Resp({"code": 500, "result": "error", "reason": "invalid regid"})
|
|
|
|
monkeypatch.setattr(vendor_push.settings, "XIAOMI_PUSH_APP_SECRET", "xiaomi-secret")
|
|
monkeypatch.setattr(vendor_push.httpx, "request", _fake_request)
|
|
token = _login(client, "13900011009")
|
|
|
|
r = client.post(
|
|
"/api/v1/push/test",
|
|
json={"vendor": "xiaomi", "pushToken": "bad", "mock": False},
|
|
headers=_auth(token),
|
|
)
|
|
assert r.status_code == 502
|
|
assert "厂商推送失败" in r.json()["detail"]
|
|
|
|
|
|
def test_push_test_resolves_token_from_registered_device(
|
|
client: TestClient, _no_vendor_creds
|
|
) -> None:
|
|
token = _login(client, "13900011010")
|
|
r = client.post(
|
|
"/api/v1/device/register",
|
|
json={"device_id": "dev-push-center-1", "push_vendor": "honor", "push_token": "honor-t1"},
|
|
headers=_auth(token),
|
|
)
|
|
assert r.status_code == 200, r.text
|
|
|
|
r = client.post(
|
|
"/api/v1/push/test",
|
|
json={"deviceId": "dev-push-center-1", "type": "perm_accessibility"},
|
|
headers=_auth(token),
|
|
)
|
|
assert r.status_code == 200, r.text
|
|
body = r.json()
|
|
assert body["vendor"] == "honor"
|
|
assert body["title"] == "检测到您的比价功能已失效"
|
|
|
|
|
|
def test_push_test_validation_errors(client: TestClient, _no_vendor_creds) -> None:
|
|
token = _login(client, "13900011011")
|
|
|
|
# 未知厂商
|
|
r = client.post(
|
|
"/api/v1/push/test",
|
|
json={"vendor": "nokia", "pushToken": "t"},
|
|
headers=_auth(token),
|
|
)
|
|
assert r.status_code == 400
|
|
|
|
# 未知类型
|
|
r = client.post(
|
|
"/api/v1/push/test",
|
|
json={"vendor": "xiaomi", "pushToken": "t", "type": "bogus"},
|
|
headers=_auth(token),
|
|
)
|
|
assert r.status_code == 400
|
|
assert "unknown notification type" in r.json()["detail"]
|
|
|
|
# 真发但没有 token 可用
|
|
r = client.post(
|
|
"/api/v1/push/test",
|
|
json={"vendor": "xiaomi", "mock": False},
|
|
headers=_auth(token),
|
|
)
|
|
assert r.status_code == 409
|