exinglang
444 lines
17 KiB
Python
444 lines
17 KiB
Python
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
|
|
from app.core import heartbeat_monitor_worker
|
|
from app.db.session import SessionLocal
|
|
from app.integrations import vendor_push
|
|
from app.models.device import DeviceLiveness
|
|
from app.repositories import user as user_repo
|
|
|
|
|
|
class _Resp:
|
|
status_code = 200
|
|
text = "{}"
|
|
|
|
def __init__(self, data: dict) -> None:
|
|
self._data = data
|
|
|
|
def json(self) -> dict:
|
|
return self._data
|
|
|
|
|
|
def test_xiaomi_accessibility_payload(monkeypatch) -> 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-msg"}})
|
|
|
|
monkeypatch.setattr(vendor_push.settings, "XIAOMI_PUSH_APP_SECRET", "xiaomi-secret")
|
|
monkeypatch.setattr(vendor_push.settings, "XIAOMI_PUSH_CHANNEL_ID", "")
|
|
monkeypatch.setattr(vendor_push.settings, "XIAOMI_PUSH_TEMPLATE_ID", "")
|
|
monkeypatch.setattr(vendor_push.settings, "XIAOMI_PUSH_TEMPLATE_TITLE", "")
|
|
monkeypatch.setattr(vendor_push.settings, "XIAOMI_PUSH_TEMPLATE_DESCRIPTION", "")
|
|
monkeypatch.setattr(vendor_push.settings, "XIAOMI_PUSH_TEMPLATE_PARAM_JSON", "")
|
|
monkeypatch.setattr(vendor_push.httpx, "request", _fake_request)
|
|
|
|
data = vendor_push.send_accessibility_disabled("xiaomi", "xm-regid")
|
|
|
|
assert data["data"]["id"] == "xm-msg"
|
|
assert captured["method"] == "POST"
|
|
assert captured["url"] == vendor_push.settings.XIAOMI_PUSH_SEND_ENDPOINT
|
|
assert captured["headers"]["Authorization"] == "key=xiaomi-secret"
|
|
body = captured["data"]
|
|
assert body["registration_id"] == "xm-regid"
|
|
assert body["restricted_package_name"] == "com.jishisongfu.shaguabijia"
|
|
assert json.loads(body["payload"]) == {"type": "accessibility_disabled"}
|
|
assert "extra.channel_id" not in body
|
|
|
|
|
|
def test_xiaomi_payload_with_channel_and_template(monkeypatch) -> 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-msg"}})
|
|
|
|
monkeypatch.setattr(vendor_push.settings, "XIAOMI_PUSH_APP_SECRET", "xiaomi-secret")
|
|
monkeypatch.setattr(vendor_push.settings, "XIAOMI_PUSH_CHANNEL_ID", "130")
|
|
monkeypatch.setattr(vendor_push.settings, "XIAOMI_PUSH_TEMPLATE_ID", "1001")
|
|
monkeypatch.setattr(vendor_push.settings, "XIAOMI_PUSH_TEMPLATE_TITLE", "{$app_name$}提醒")
|
|
monkeypatch.setattr(vendor_push.settings, "XIAOMI_PUSH_TEMPLATE_DESCRIPTION", "{$content$}")
|
|
monkeypatch.setattr(
|
|
vendor_push.settings,
|
|
"XIAOMI_PUSH_TEMPLATE_PARAM_JSON",
|
|
'{"app_name":"傻瓜比价","content":"{alert}"}',
|
|
)
|
|
monkeypatch.setattr(vendor_push.httpx, "request", _fake_request)
|
|
|
|
vendor_push.send_accessibility_disabled(
|
|
"xiaomi",
|
|
"xm-regid",
|
|
title="测试标题",
|
|
alert="测试内容",
|
|
)
|
|
|
|
body = captured["data"]
|
|
assert body["title"] == "{$app_name$}提醒"
|
|
assert body["description"] == "{$content$}"
|
|
assert body["extra.channel_id"] == "130"
|
|
assert body["extra.template_id"] == "1001"
|
|
assert body["extra.template_param"] == '{"app_name":"傻瓜比价","content":"测试内容"}'
|
|
|
|
|
|
def test_vivo_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.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)
|
|
|
|
data = vendor_push.send_accessibility_disabled("vivo", "vivo-regid")
|
|
|
|
assert data["taskId"] == "vivo-task"
|
|
assert calls[0]["url"] == vendor_push.settings.VIVO_PUSH_AUTH_ENDPOINT
|
|
assert calls[0]["json"]["appId"] == "106072775"
|
|
assert calls[0]["json"]["sign"]
|
|
assert calls[1]["url"] == vendor_push.settings.VIVO_PUSH_SEND_ENDPOINT
|
|
assert calls[1]["headers"]["authToken"] == "vivo-auth"
|
|
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] = []
|
|
|
|
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.httpx, "request", _fake_request)
|
|
|
|
data = vendor_push.send_accessibility_disabled("oppo", "oppo-regid")
|
|
|
|
assert data["data"]["message_id"] == "oppo-msg"
|
|
assert calls[0]["data"]["app_key"] == "oppo-key"
|
|
assert calls[0]["data"]["sign"]
|
|
message = json.loads(calls[1]["data"]["message"])
|
|
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()
|
|
calls: list[dict] = []
|
|
|
|
def _fake_request(method, url, **kwargs): # noqa: ANN001
|
|
calls.append({"method": method, "url": url, **kwargs})
|
|
if url == vendor_push.settings.HONOR_PUSH_TOKEN_ENDPOINT:
|
|
return _Resp({"access_token": "honor-access", "expires_in": 3600})
|
|
return _Resp({"code": 200, "message": "successful!", "data": {"sendResult": True}})
|
|
|
|
monkeypatch.setattr(vendor_push.settings, "HONOR_PUSH_APP_ID", "104559789")
|
|
monkeypatch.setattr(vendor_push.settings, "HONOR_PUSH_CLIENT_ID", "honor-client")
|
|
monkeypatch.setattr(vendor_push.settings, "HONOR_PUSH_CLIENT_SECRET", "honor-secret")
|
|
monkeypatch.setattr(vendor_push.httpx, "request", _fake_request)
|
|
|
|
data = vendor_push.send_accessibility_disabled("honor", "honor-token")
|
|
|
|
assert data["code"] == 200
|
|
assert calls[0]["data"]["client_id"] == "honor-client"
|
|
assert calls[1]["headers"]["Authorization"] == "Bearer honor-access"
|
|
assert calls[1]["headers"]["timestamp"]
|
|
assert calls[1]["url"].endswith("/api/v1/104559789/sendMessage")
|
|
body = calls[1]["json"]
|
|
assert body["token"] == ["honor-token"]
|
|
assert body["android"]["targetUserType"] == 1
|
|
assert body["android"]["notification"]["clickAction"] == {"type": 3}
|
|
assert json.loads(body["data"]) == {"type": "accessibility_disabled"}
|
|
|
|
|
|
def _seed_overdue_device(
|
|
*,
|
|
phone: str,
|
|
device_id: str,
|
|
push_vendor: str | None,
|
|
push_token: str | None,
|
|
) -> int:
|
|
with SessionLocal() as db:
|
|
user = user_repo.upsert_user_for_login(db, phone=phone, register_channel="sms")
|
|
device = DeviceLiveness(
|
|
user_id=user.id,
|
|
device_id=device_id,
|
|
push_vendor=push_vendor,
|
|
push_token=push_token,
|
|
platform="android",
|
|
ever_protected=True,
|
|
last_heartbeat_at=datetime.now(timezone.utc) - timedelta(minutes=30), # noqa: UP017
|
|
last_report_protection_on=True,
|
|
liveness_state="alive",
|
|
kill_alert_pending=False,
|
|
)
|
|
db.add(device)
|
|
db.commit()
|
|
db.refresh(device)
|
|
return device.id
|
|
|
|
|
|
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}"}
|
|
|
|
|
|
def test_heartbeat_monitor_pushes_overdue_device(monkeypatch) -> None:
|
|
device_pk = _seed_overdue_device(
|
|
phone="13900009001",
|
|
device_id="dev-push-honor",
|
|
push_vendor="honor",
|
|
push_token="honor-token-1",
|
|
)
|
|
calls: list[tuple[str, str]] = []
|
|
|
|
def _fake_send(push_vendor: str, push_token: str) -> dict:
|
|
calls.append((push_vendor, push_token))
|
|
return {"msg_id": "m1"}
|
|
|
|
monkeypatch.setattr(
|
|
heartbeat_monitor_worker.vendor_push,
|
|
"send_accessibility_disabled",
|
|
_fake_send,
|
|
)
|
|
|
|
result = heartbeat_monitor_worker._scan_once(timeout_minutes=10)
|
|
|
|
assert result["pushed"] >= 1
|
|
assert ("honor", "honor-token-1") in calls
|
|
with SessionLocal() as db:
|
|
device = db.get(DeviceLiveness, device_pk)
|
|
assert device is not None
|
|
assert device.liveness_state == "notified"
|
|
assert device.kill_alert_pending is True
|
|
|
|
|
|
def test_heartbeat_monitor_skips_push_without_vendor_token(monkeypatch) -> None:
|
|
device_pk = _seed_overdue_device(
|
|
phone="13900009002",
|
|
device_id="dev-push-no-token",
|
|
push_vendor=None,
|
|
push_token=None,
|
|
)
|
|
|
|
def _fake_send(push_vendor: str, push_token: str) -> dict:
|
|
raise AssertionError(f"should not push without token: {push_vendor}/{push_token}")
|
|
|
|
monkeypatch.setattr(
|
|
heartbeat_monitor_worker.vendor_push,
|
|
"send_accessibility_disabled",
|
|
_fake_send,
|
|
)
|
|
|
|
result = heartbeat_monitor_worker._scan_once(timeout_minutes=10)
|
|
|
|
assert result["checked"] >= 1
|
|
with SessionLocal() as db:
|
|
device = db.get(DeviceLiveness, device_pk)
|
|
assert device is not None
|
|
assert device.liveness_state == "notified"
|
|
assert device.kill_alert_pending is True
|
|
|
|
|
|
def test_push_test_endpoint_schedules_vendor_push(client: TestClient, monkeypatch) -> None:
|
|
token = _login(client, "13900009003")
|
|
calls: list[tuple[str, str, str, str]] = []
|
|
|
|
def _fake_send(push_vendor: str, push_token: str, *, title: str, alert: str) -> dict:
|
|
calls.append((push_vendor, push_token, title, alert))
|
|
return {"msg_id": "m-test"}
|
|
|
|
monkeypatch.setattr(device_api.vendor_push, "send_accessibility_disabled", _fake_send)
|
|
|
|
r = client.post(
|
|
"/api/v1/device/push-test",
|
|
json={
|
|
"device_id": "dev-push-test",
|
|
"push_vendor": "honor",
|
|
"push_token": "honor-test-token",
|
|
"delay_seconds": 0,
|
|
},
|
|
headers=_auth(token),
|
|
)
|
|
|
|
assert r.status_code == 200, r.text
|
|
assert r.json() == {
|
|
"ok": True,
|
|
"delay_seconds": 0,
|
|
"has_push_token": True,
|
|
}
|
|
assert calls == [
|
|
(
|
|
"honor",
|
|
"honor-test-token",
|
|
"测试推送",
|
|
"这是一条厂商通道测试推送。收到它说明 App 被划掉后仍可通过系统通知栏触达。",
|
|
)
|
|
]
|
|
|
|
|
|
def test_push_test_endpoint_requires_vendor_token(client: TestClient) -> None:
|
|
token = _login(client, "13900009004")
|
|
|
|
r = client.post(
|
|
"/api/v1/device/push-test",
|
|
json={"device_id": "dev-push-test-no-token", "delay_seconds": 0},
|
|
headers=_auth(token),
|
|
)
|
|
|
|
assert r.status_code == 409
|
|
assert r.json()["detail"] == "push vendor token not ready"
|