thermograph/backend/tests/notifications/test_push.py

"""VAPID key resolution: the atomic first-writer-wins claim of the keypair file
(push.py's `_claim_file`), and that `_load()` caches whatever that settles on
rather than its own local generation when it loses the race. Also `send()`'s
mapping of push-service responses onto 'ok'/'gone'/'error'.

No network — where `send()` is exercised the pywebpush call is stubbed; the
dispatch path (notify.py's `test_push_dispatched_on_new_notification` etc.) stubs
`push.send` itself."""
import json

import pytest
from pywebpush import WebPushException

from notifications import push


def _reset_state(monkeypatch, path):
    monkeypatch.setattr(push, "_VAPID_PATH", str(path))
    monkeypatch.setattr(push, "_DATA_DIR", str(path.parent))
    monkeypatch.setattr(push, "_keys", None)
    monkeypatch.delenv("THERMOGRAPH_VAPID_PRIVATE_KEY", raising=False)
    monkeypatch.delenv("THERMOGRAPH_VAPID_PUBLIC_KEY", raising=False)


def test_claim_file_first_writer_wins(monkeypatch, tmp_path):
    path = tmp_path / "vapid.json"
    _reset_state(monkeypatch, path)
    data = {"private_key": "priv-a", "public_key": "pub-a"}

    result = push._claim_file(data)

    assert result == data
    assert json.loads(path.read_text()) == data
    # No leftover temp file.
    assert list(tmp_path.iterdir()) == [path]


def test_claim_file_reads_back_winner_on_race(monkeypatch, tmp_path):
    """A process that loses the os.link() race must discard its own freshly
    generated keypair and use whatever the winner actually persisted."""
    path = tmp_path / "vapid.json"
    _reset_state(monkeypatch, path)
    winner = {"private_key": "priv-winner", "public_key": "pub-winner"}
    path.write_text(json.dumps(winner))   # simulates another process winning first
    loser = {"private_key": "priv-loser", "public_key": "pub-loser"}

    result = push._claim_file(loser)

    assert result == winner
    assert json.loads(path.read_text()) == winner   # the loser never touched the file


def test_load_caches_winner_keys_not_its_own_generation(monkeypatch, tmp_path):
    """End-to-end through _load(): on a cold /state volume, a worker that hits the
    generate branch but loses the write race must end up with the SAME in-process
    cached keys the file actually holds — not the keys it generated locally,
    which would silently diverge from every other worker (a subscription signed
    against one worker's public key then fails to verify under another's
    private key)."""
    path = tmp_path / "vapid.json"
    _reset_state(monkeypatch, path)
    winner = {"private_key": "priv-winner", "public_key": "pub-winner"}

    def _generate_and_lose_the_race():
        # Between our cache-miss file read and our own _claim_file() call, a
        # concurrent worker wins and writes the real file first.
        path.write_text(json.dumps(winner))
        return {"private_key": "priv-mine", "public_key": "pub-mine"}

    monkeypatch.setattr(push, "_generate", _generate_and_lose_the_race)

    result = push._load()

    assert result == winner
    assert push._keys == winner            # the process-wide cache matches the file
    assert push.public_key() == "pub-winner"


def test_load_reads_existing_file_without_generating(monkeypatch, tmp_path):
    path = tmp_path / "vapid.json"
    _reset_state(monkeypatch, path)
    existing = {"private_key": "priv-x", "public_key": "pub-x"}
    path.write_text(json.dumps(existing))

    def boom():
        raise AssertionError("must not generate when a valid file already exists")
    monkeypatch.setattr(push, "_generate", boom)

    assert push._load() == existing


def test_load_prefers_env_over_file(monkeypatch, tmp_path):
    path = tmp_path / "vapid.json"
    _reset_state(monkeypatch, path)
    path.write_text(json.dumps({"private_key": "priv-file", "public_key": "pub-file"}))
    monkeypatch.setenv("THERMOGRAPH_VAPID_PRIVATE_KEY", " priv-env ")
    monkeypatch.setenv("THERMOGRAPH_VAPID_PUBLIC_KEY", " pub-env ")

    result = push._load()

    assert result == {"private_key": "priv-env", "public_key": "pub-env"}


# --- send() response mapping ------------------------------------------------
_SUB = {"endpoint": "https://push.example.com/ep-1", "keys": {"p256dh": "BKEY", "auth": "YXV0aA"}}


class _Resp:
    def __init__(self, status_code):
        self.status_code = status_code


def _raise_status(status):
    def _webpush(**kwargs):
        raise WebPushException("boom", response=_Resp(status))
    return _webpush


@pytest.mark.parametrize("status", [401, 403, 404, 410])
def test_send_prunes_permanently_dead_endpoints(monkeypatch, status):
    # 404/410 = endpoint retired; 401/403 = VAPID key mismatch (e.g. after a key
    # rotation). All are permanently dead, so send() reports 'gone' and the caller
    # deletes the row — otherwise a rotated key leaves dead subscriptions forever.
    monkeypatch.setattr(push, "webpush", _raise_status(status))
    assert push.send(_SUB, {"hello": "world"}) == "gone"


@pytest.mark.parametrize("status", [429, 500, 502])
def test_send_keeps_row_on_transient_failure(monkeypatch, status):
    # Rate limits / 5xx may recover; keep the row and report it as an error.
    monkeypatch.setattr(push, "webpush", _raise_status(status))
    assert push.send(_SUB, {"hello": "world"}) == "error"


def test_send_ok(monkeypatch):
    monkeypatch.setattr(push, "webpush", lambda **kwargs: None)
    assert push.send(_SUB, {"hello": "world"}) == "ok"


def test_default_contact_is_the_org_domain():
    # The VAPID "sub" contact must be a domain we actually own; .app was a typo.
    assert push._CONTACT == "mailto:admin@thermograph.org"