spirekeeper/tests/test_pair_endpoint.py

"""Wiring tests for POST /machines/{id}/pair (S0 / #9).

The pairing *service* is covered in test_pairing.py with a fake bunker;
here we only exercise the endpoint glue — ownership, the empty-relays
guard, the post-mint collision guard, persistence of the bunker-minted
hex npub, and error mapping — by monkeypatching the module-level deps.
"""

import asyncio
from datetime import datetime, timezone
from types import SimpleNamespace

import pytest
from fastapi import HTTPException
from lnbits.utils.nostr import hex_to_npub

from .. import views_api
from ..models import Machine, PairMachineData
from ..pairing import PairingError, PairResult

_NOW = datetime(2026, 6, 16, tzinfo=timezone.utc)
_SPIRE_HEX = "522a4538f1df96508d9ee8b14072344dd4a566acfe03c25a92a39179c6fca891"
_SPIRE_NPUB = hex_to_npub(_SPIRE_HEX)


def _machine(npub: str = "placeholder") -> Machine:
    return Machine(
        id="m1",
        operator_user_id="op1",
        machine_npub=npub,
        wallet_id="w1",
        name="sintra",
        location=None,
        fiat_code="EUR",
        is_active=True,
        created_at=_NOW,
        updated_at=_NOW,
    )


class _FakeAdmin:
    @classmethod
    def from_settings(cls):
        return cls()

    async def __aenter__(self):
        return self

    async def __aexit__(self, *exc):
        return False


def _result() -> PairResult:
    return PairResult(
        spire_npub=_SPIRE_NPUB,
        spire_pubkey_hex=_SPIRE_HEX,
        bunker_key_name="spire-m1",
        bunker_url="bunker://x?relay=r&secret=s",  # pragma: allowlist secret
        seed_url="spire-seed:v1:abc",
    )


def _wire(monkeypatch, *, pair="ok"):
    state: dict = {"persisted": None, "collision": None}

    async def fake_owned(machine_id, user_id):
        return _machine()

    async def fake_pair(machine, *, relays, admin_client, duration_hours=None):
        if pair == "error":
            raise PairingError("boom")
        return _result()

    async def fake_collision(npub):
        state["collision"] = npub

    async def fake_persist(
        machine_id, *, machine_npub, bunker_spire_key_name, paired_at
    ):
        state["persisted"] = (machine_id, machine_npub, bunker_spire_key_name)
        return _machine(npub=machine_npub)

    monkeypatch.setattr(views_api, "_machine_owned_by", fake_owned)
    monkeypatch.setattr(views_api, "NsecBunkerAdminClient", _FakeAdmin)
    monkeypatch.setattr(views_api, "pair_spire", fake_pair)
    monkeypatch.setattr(views_api, "_assert_no_pubkey_collision", fake_collision)
    monkeypatch.setattr(views_api, "set_machine_pairing", fake_persist)
    return state


def _call(relays):
    user = SimpleNamespace(id="op1")
    return asyncio.run(
        views_api.api_pair_machine("m1", PairMachineData(relays=relays), user)
    )


def test_pair_persists_hex_npub_and_returns_seed(monkeypatch):
    state = _wire(monkeypatch)
    result = _call(["wss://r"])
    assert result.seed_url == "spire-seed:v1:abc"
    # collision guard ran on the bunker-minted hex, and we persisted it as npub
    assert state["collision"] == _SPIRE_HEX
    assert state["persisted"] == ("m1", _SPIRE_HEX, "spire-m1")


def test_pair_empty_relays_rejected(monkeypatch):
    _wire(monkeypatch)
    with pytest.raises(HTTPException) as ei:
        _call([])
    assert ei.value.status_code == 400


def test_pair_failure_maps_to_bad_gateway(monkeypatch):
    state = _wire(monkeypatch, pair="error")
    with pytest.raises(HTTPException) as ei:
        _call(["wss://r"])
    assert ei.value.status_code == 502
    # nothing persisted on failure
    assert state["persisted"] is None


def _wire_revoke(monkeypatch, *, revoke="ok", count=2):
    state = {"unpaired": None}

    async def fake_owned(machine_id, user_id):
        return _machine()

    async def fake_revoke(machine, *, admin_client):
        if revoke == "error":
            raise PairingError("boom")
        return count

    async def fake_unpaired(machine_id):
        state["unpaired"] = machine_id
        return _machine()

    monkeypatch.setattr(views_api, "_machine_owned_by", fake_owned)
    monkeypatch.setattr(views_api, "NsecBunkerAdminClient", _FakeAdmin)
    monkeypatch.setattr(views_api, "revoke_spire", fake_revoke)
    monkeypatch.setattr(views_api, "set_machine_unpaired", fake_unpaired)
    return state


def _call_revoke():
    user = SimpleNamespace(id="op1")
    return asyncio.run(views_api.api_revoke_machine("m1", user))


def test_revoke_cuts_access_and_marks_unpaired(monkeypatch):
    state = _wire_revoke(monkeypatch, count=2)
    result = _call_revoke()
    assert result.revoked_count == 2
    assert state["unpaired"] == "m1"


def test_revoke_zero_when_nothing_bound(monkeypatch):
    _wire_revoke(monkeypatch, count=0)
    assert _call_revoke().revoked_count == 0


def test_revoke_failure_maps_to_bad_gateway(monkeypatch):
    state = _wire_revoke(monkeypatch, revoke="error")
    with pytest.raises(HTTPException) as ei:
        _call_revoke()
    assert ei.value.status_code == 502
    assert state["unpaired"] is None  # not persisted on failure