feat(v2): nostr-transport roster-resolver hook (#20 path-B)
Some checks failed
ci.yml / feat(v2): nostr-transport roster-resolver hook (#20 path-B) (pull_request) Failing after 0s

Exposes `resolve(sender_pubkey_hex) -> RouteHit | None` and a
`register_with_lnbits()` helper that lazily-imports + soft-fails on
lnbits versions without `register_roster_resolver`. Wired into
`satmachineadmin_start()`.

The hook delivers the path-B outcome ("cash-out sats go to the
operator's wallet, not an auto-created machine wallet") once the
lnbits side ships its half. Shape contract `(operator_user_id,
wallet_id, source_extension)` frozen per coord-log 2026-05-31T15:25Z.
Branch held local until lnbits lands the registry — no behaviour
change on the current lnbits version, just the future-ready handoff
+ a benign INFO log on boot.

Boot-smoked in dev container: extension loads, registration logs the
documented soft-fail message, invoice listener + cassette consumer
unchanged. 6 new unit tests cover happy path, miss, bech32 +
uppercase canonicalisation, fail-closed on malformed input, and the
soft-fail register branch.
This commit is contained in:
Padreug 2026-05-31 19:14:45 +02:00
commit 99efa52b69
3 changed files with 313 additions and 0 deletions

View file

@ -0,0 +1,163 @@
"""
Tests for `nostr_transport_roster.resolve` the lookup function
satmachineadmin hands lnbits' nostr-transport via
`register_roster_resolver` (path-B wallet-routing fix, #20 /
coord-log 2026-05-31T15:25Z).
Verifies:
- Known ATM npub RouteHit with operator_user_id + wallet_id from
the machine row
- Unknown sender None (lnbits falls back to its other resolvers,
or fail-closed rejection per the env-gated posture)
- bech32 input is normalised to hex before lookup
- Uppercase hex input is normalised to lowercase before lookup
- Malformed input raises (fail-closed sub-case per lnbits 15:15Z ack)
`register_with_lnbits` is also smoke-tested for the soft-fail branch
that fires on lnbits versions without `register_roster_resolver`. The
positive (lnbits hook present) branch needs a live lnbits import +
will be covered once the lnbits-side PR lands.
Coroutines driven via `asyncio.run` per project convention (no pytest-
asyncio plugin in CI; see test_cassette_state_consumer.py header).
"""
import asyncio
import sys
from types import SimpleNamespace
import coincurve
import pytest
from lnbits.utils.nostr import hex_to_npub
from .. import nostr_transport_roster as roster
from ..nostr_transport_roster import register_with_lnbits, resolve
_ATM_SEC = "00" * 31 + "02"
_ATM_PUB_HEX = (
coincurve.PrivateKey(bytes.fromhex(_ATM_SEC))
.public_key.format(compressed=True)[1:]
.hex()
)
_ATM_PUB_NPUB = hex_to_npub(_ATM_PUB_HEX)
def _fake_machine(operator_user_id: str, wallet_id: str, npub_hex: str):
return SimpleNamespace(
operator_user_id=operator_user_id,
wallet_id=wallet_id,
machine_npub=npub_hex,
)
def test_resolve_known_atm_returns_route_hit(monkeypatch):
captured: dict[str, str] = {}
async def _fake_lookup(pubkey_hex: str):
captured["pubkey_hex"] = pubkey_hex
return _fake_machine(
operator_user_id="op-123",
wallet_id="wallet-abc",
npub_hex=_ATM_PUB_HEX,
)
monkeypatch.setattr(roster, "get_machine_by_atm_pubkey_hex", _fake_lookup)
result = asyncio.run(resolve(_ATM_PUB_HEX))
# `_build_route_hit` prefers lnbits' canonical `RouteHit` when
# importable + falls back to our local class otherwise; assert on
# the frozen field-shape contract (coord-log 2026-05-31T15:25Z),
# not the specific class identity, so the test passes against
# both lnbits versions.
assert result is not None
assert result.operator_user_id == "op-123"
assert result.wallet_id == "wallet-abc"
assert result.source_extension == "satmachineadmin"
assert captured["pubkey_hex"] == _ATM_PUB_HEX
def test_resolve_unknown_sender_returns_none(monkeypatch):
async def _no_match(pubkey_hex: str):
return None
monkeypatch.setattr(roster, "get_machine_by_atm_pubkey_hex", _no_match)
result = asyncio.run(resolve(_ATM_PUB_HEX))
assert result is None
def test_resolve_canonicalises_bech32_to_hex(monkeypatch):
"""Sender pubkeys arrive lowercase-hex from lnbits PR #4, but the
resolver is paranoid: a bech32 input must still hit the hex-keyed
crud lookup."""
captured: dict[str, str] = {}
async def _fake_lookup(pubkey_hex: str):
captured["pubkey_hex"] = pubkey_hex
return _fake_machine(
operator_user_id="op-bech32",
wallet_id="wallet-bech32",
npub_hex=_ATM_PUB_HEX,
)
monkeypatch.setattr(roster, "get_machine_by_atm_pubkey_hex", _fake_lookup)
result = asyncio.run(resolve(_ATM_PUB_NPUB))
assert result is not None
assert captured["pubkey_hex"] == _ATM_PUB_HEX
def test_resolve_lowercases_uppercase_hex(monkeypatch):
captured: dict[str, str] = {}
async def _fake_lookup(pubkey_hex: str):
captured["pubkey_hex"] = pubkey_hex
return None
monkeypatch.setattr(roster, "get_machine_by_atm_pubkey_hex", _fake_lookup)
asyncio.run(resolve(_ATM_PUB_HEX.upper()))
assert captured["pubkey_hex"] == _ATM_PUB_HEX
def test_resolve_raises_on_malformed_input(monkeypatch):
"""Fail-closed sub-case per lnbits 15:15Z ack item 2: resolver
raising an exception surfaces to lnbits as a reject + ERROR log,
NOT a silent fall-through to auto-account creation."""
async def _unreachable(pubkey_hex: str):
raise AssertionError("crud must not be reached for malformed input")
monkeypatch.setattr(roster, "get_machine_by_atm_pubkey_hex", _unreachable)
with pytest.raises((ValueError, AssertionError)):
asyncio.run(resolve("not-a-pubkey"))
def test_register_with_lnbits_soft_fails_without_hook(monkeypatch):
"""Until the lnbits-side path-B PR lands, the registration call
must soft-fail cleanly (returns False, no exception) so
satmachineadmin keeps booting on every lnbits version."""
real_import = (
__builtins__["__import__"]
if isinstance(__builtins__, dict)
else __builtins__.__import__
)
def _faulty_import(name, *args, **kwargs):
if name == "lnbits.core.services.nostr_transport":
raise ImportError("simulated: pre-path-B lnbits")
return real_import(name, *args, **kwargs)
monkeypatch.setattr("builtins.__import__", _faulty_import)
# Drop any cached import so the lazy `from … import …` inside
# register_with_lnbits re-triggers the import statement.
monkeypatch.delitem(
sys.modules, "lnbits.core.services.nostr_transport", raising=False
)
assert register_with_lnbits() is False