spirekeeper/tests/test_roster_resolver.py

"""
Tests for `nostr_transport_roster.resolve` — the lookup function
spirekeeper 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 == "spirekeeper"
    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
    spirekeeper 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