feat(pairing): POST /machines/{id}/pair endpoint (#9)
Some checks failed
ci.yml / feat(pairing): POST /machines/{id}/pair endpoint (#9) (pull_request) Failing after 0s

Wires the pairing service into the operator API. api_pair_machine:
  - _machine_owned_by ownership guard (404 on miss)
  - opens NsecBunkerAdminClient.from_settings() and runs pair_spire
  - maps bunker failures: not-configured -> 503, PairingError/NsecBunkerError
    -> 502 (nothing persisted on failure)
  - runs _assert_no_pubkey_collision on the bunker-minted hex, then
    set_machine_pairing persists machine_npub (= minted spire identity, so
    path-B roster routes it), bunker_spire_key_name, paired_at.

Re-pair supported; revoke/expiry gated on aiolabs/lnbits#54.

Adds Create... PairMachineData {relays} body, set_machine_pairing CRUD,
and 3 endpoint wiring tests (persist+collision, empty-relays 400, failure
502). 203 tests green. Pre-existing black/ruff debt in crud/views_api left
untouched (version-drift churn avoided); new code is lint-clean.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Padreug 2026-06-16 23:39:18 +02:00
commit 761f078053
4 changed files with 213 additions and 0 deletions

120
tests/test_pair_endpoint.py Normal file
View file

@ -0,0 +1,120 @@
"""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):
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