shaguabijia-app-server/tests/test_device_push.py

from __future__ import annotations

import json
from datetime import datetime, timedelta, timezone

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["clientCustomMap"] == {"type": "accessibility_disabled"}


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 json.loads(message["notification"]["action_parameters"]) == {
        "type": "accessibility_disabled"
    }


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"