stackchain-dashboard/tests/test_push_notifications.py

import json
from types import SimpleNamespace

import pytest

from src import main
from src.push_notifications import PushConfiguration, dispatch_unread_updates
from src.push_subscription_store import PushSubscriptionStore


def test_subscription_store_delivers_each_unread_thread_once_per_device(tmp_path):
    store = PushSubscriptionStore(tmp_path / "push.sqlite3")
    subscription = {
        "endpoint": "https://push.example/device-a",
        "keys": {"p256dh": "public-key", "auth": "auth-secret"},
    }

    store.upsert("session-a", subscription)

    first = store.claim_unseen({41, 42})
    store.mark_delivered("session-a", first[0].thread_ids)
    second = store.claim_unseen({41, 42})

    assert [(item.session_id, item.thread_ids) for item in first] == [
        ("session-a", (41, 42))
    ]
    assert second == []


def test_replacing_subscription_resets_delivery_cursor_and_revocation_removes_device(tmp_path):
    store = PushSubscriptionStore(tmp_path / "push.sqlite3")
    original = {
        "endpoint": "https://push.example/original",
        "keys": {"p256dh": "old-key", "auth": "old-auth"},
    }
    replacement = {
        "endpoint": "https://push.example/replacement",
        "keys": {"p256dh": "new-key", "auth": "new-auth"},
    }
    store.upsert("session-a", original)
    store.claim_unseen({9})

    store.upsert("session-a", replacement)
    delivery = store.claim_unseen({9})
    store.delete_session("session-a")

    assert delivery[0].subscription == replacement
    assert delivery[0].thread_ids == (9,)
    assert store.claim_unseen({10}) == []


@pytest.mark.anyio
async def test_dispatch_sends_one_privacy_safe_deep_link_per_new_thread(tmp_path):
    store = PushSubscriptionStore(tmp_path / "push.sqlite3")
    store.upsert("session-a", {
        "endpoint": "https://push.example/device-a",
        "keys": {"p256dh": "public-key", "auth": "auth-secret"},
    })
    sent = []

    async def unread():
        return {"items": [
            {"notification_id": 42, "repository": "private/repo", "title": "Secret title"},
        ]}

    async def send(subscription, payload):
        sent.append((subscription, json.loads(payload)))

    config = PushConfiguration("public-vapid", "private-vapid", "mailto:ops@example.com")
    assert await dispatch_unread_updates(store, config, unread, send) == 1
    assert await dispatch_unread_updates(store, config, unread, send) == 0
    assert sent[0][1] == {
        "title": "New work update",
        "body": "Tap to review it in Stackchain.",
        "route": "#/my-work/update/42",
        "tag": "stackchain-update-42",
    }
    assert "private/repo" not in json.dumps(sent)
    assert "Secret title" not in json.dumps(sent)


@pytest.mark.anyio
async def test_dispatch_removes_an_expired_push_endpoint(tmp_path):
    store = PushSubscriptionStore(tmp_path / "push.sqlite3")
    store.upsert("session-a", {
        "endpoint": "https://push.example/expired",
        "keys": {"p256dh": "public-key", "auth": "auth-secret"},
    })

    class Expired(Exception):
        response = SimpleNamespace(status_code=410)

    async def unread():
        return {"items": [{"notification_id": 8}]}

    async def send(_subscription, _payload):
        raise Expired()

    config = PushConfiguration("public", "private", "mailto:ops@example.com")
    assert await dispatch_unread_updates(store, config, unread, send) == 0
    assert store.is_subscribed("session-a") is False


@pytest.mark.anyio
async def test_authenticated_device_can_subscribe_report_status_and_unsubscribe(tmp_path, monkeypatch):
    store = PushSubscriptionStore(tmp_path / "push.sqlite3")
    monkeypatch.setattr(main, "_push_subscription_store", store)
    monkeypatch.setattr(
        main,
        "_push_configuration",
        lambda: PushConfiguration("public-vapid", "private-vapid", "mailto:ops@example.com"),
    )
    async def management_id(_session):
        return "session-a"

    monkeypatch.setattr(main.dashboard_auth, "session_management_id", management_id)

    async def no_unread():
        return {"items": []}

    monkeypatch.setattr(main, "notifications", no_unread)
    request = SimpleNamespace(state=SimpleNamespace(
        dashboard_session=SimpleNamespace(session_id="session-a")
    ))
    payload = main.PushSubscriptionPayload(
        endpoint="https://push.example/device-a",
        keys={"p256dh": "public-key", "auth": "auth-secret"},
    )

    assert await main.push_status(request) == {
        "available": True, "subscribed": False, "public_key": "public-vapid"
    }
    assert await main.subscribe_push(payload, request) == {"subscribed": True}
    assert (await main.push_status(request))["subscribed"] is True
    assert await main.unsubscribe_push(request) == {"subscribed": False}
    assert (await main.push_status(request))["subscribed"] is False


@pytest.mark.anyio
async def test_subscription_fails_closed_when_existing_unread_baseline_is_unavailable(tmp_path, monkeypatch):
    store = PushSubscriptionStore(tmp_path / "push.sqlite3")
    monkeypatch.setattr(main, "_push_subscription_store", store)
    monkeypatch.setattr(
        main, "_push_configuration",
        lambda: PushConfiguration("public", "private", "mailto:ops@example.com"),
    )

    async def management_id(_session):
        return "device-a"

    async def unavailable():
        raise RuntimeError("Gitea unavailable")

    monkeypatch.setattr(main.dashboard_auth, "session_management_id", management_id)
    monkeypatch.setattr(main, "notifications", unavailable)
    request = SimpleNamespace(state=SimpleNamespace(dashboard_session=object()))
    payload = main.PushSubscriptionPayload(
        endpoint="https://push.example/device-a",
        keys={"p256dh": "public-key", "auth": "auth-secret"},
    )

    with pytest.raises(main.HTTPException) as raised:
        await main.subscribe_push(payload, request)

    assert raised.value.status_code == 503
    assert store.is_subscribed("device-a") is False