aiolabs/satmachineadmin
8
0
Fork 0
Code Issues Pull requests Projects Releases 4 Packages Wiki Activity Actions

refactor(v2): cassette transport — signer.nip44_* migration (#29 v1.1 / closes #21 partial)
Some checks failed
ci.yml / refactor(v2): cassette transport — signer.nip44_* migration (#29 v1.1 / closes #21 partial) (pull_request) Failing after 0s
Details

Browse source 
Migrates the cassette transport's encrypt/decrypt paths off direct
`account.prvkey` reads to `signer.nip44_encrypt` / `signer.nip44_decrypt`
on the NostrSigner ABC landed by aiolabs/lnbits PR #38 (phase 2.4). Closes
the operator-side regression flagged at coord-log 2026-05-31T06:50Z:
Greg's RemoteBunkerSigner-migrated account had `accounts.prvkey IS NULL`
post-bunker, which the old code couldn't handle — consumer was logging
WARN every poll cycle and skipping every inbound state event.

## What changed

### cassette_transport.py

- New imports: `resolve_signer`, `SignerError`, `SignerUnavailableError`,
  `NsecBunkerTimeoutError`, `NsecBunkerRpcError` from the post-#38 lnbits
  surface. (The `try: from lnbits.core.signers import SignerError` block
  in the old code was permanently failing because `SignerError` actually
  lives in `lnbits.core.signers.base`, not the package root — fixed.)
- New `_resolve_operator_signer(operator_user_id)`: single source of
  truth for "give me the operator's account + NostrSigner, or raise an
  operator-facing error." Used by both the publish path and the consumer
  task.
- New `_nip44_encrypt_via_signer(account, signer, plaintext, peer)`
  and `_nip44_decrypt_via_signer(...)`: route through `signer.nip44_*`
  first; on `SignerUnavailableError` from a LocalSigner stub (the
  post-#38 ABC has LocalSigner raise on nip44_* explicitly — bunker
  migration required for NIP-44 v2), fall back to the hand-rolled impl
  against `account.prvkey`. Transitional until every operator on the
  instance is bunker-backed (S7).
- `_sign_as_operator` simplified: now `await signer.sign_event(event)`
  (the ABC is async; the old code passed `signer.sign_event` to the
  caller without await, returning a coroutine — also broken but never
  hit because the ImportError fallback fired first).
- `publish_to_atm` flow: `_resolve_operator_signer` → `_nip44_encrypt_
  via_signer` → `_sign_as_operator` → publish. Each step maps bunker /
  signer errors to `OperatorIdentityMissing` (400) / `SignerUnavailable`
  (503) / `CassetteTransportError` (500) for the API handler.
- `decrypt_and_parse_state_event` now `async` and takes `(event, account,
  signer)` instead of `(event, operator_privkey_hex)`. Maps
  `NsecBunkerTimeoutError` → `CassetteEventTransientError` (caller
  should retry on next poll, NOT advance `state_event_id`).
  `NsecBunkerRpcError` / `SignerUnavailableError` / `Nip44Error` / etc.
  → `CassetteEventDecodeError` (terminal — caller logs + skips).
- New `CassetteEventTransientError` class for the bunker-timeout case.
  Distinct from `CassetteEventDecodeError` so the consumer can log at
  INFO + retry vs WARNING + advance.
- Deleted `_get_operator_privkey_hex` (no longer needed).

### tasks.py — _handle_cassette_state_event

- Resolves the signer via `_resolve_operator_signer(machine.operator_
  user_id)`. On `CassetteTransportError` (OperatorIdentityMissing /
  SignerUnavailable), logs + skips.
- Awaits `decrypt_and_parse_state_event(event_obj, account, signer)`.
  On `CassetteEventTransientError`, logs at INFO + returns (state_event_
  id NOT advanced → consumer retries on next poll cycle).
  On `CassetteEventDecodeError`, logs at WARNING + returns (still
  state_event_id NOT advanced for v1; the WARN log surfaces the
  underlying issue for operator triage).

### tests/test_cassette_state_consumer.py — rewritten

- Three test doubles: `_FakeBunkerSigner` (working nip44_decrypt via
  hand-rolled impl), `_FakeLocalSignerStub` (raises like the post-#38
  LocalSigner stub), `_FakeRaisingSigner` (configurable exception).
- `_fake_account` helper using SimpleNamespace — the code under test
  only reads `.signer_type` + `.prvkey`.
- Five test classes covering: bunker-signer happy path (incl. multi-
  same-denom round-trip), LocalSigner transitional fallback,
  bunker-error mapping (timeout → transient, rpc reject → decode),
  payload validation (tamper / wrong-key / missing-fields / garbage
  JSON / wrong shape), d-tag construction (unchanged, kept as
  regression guard).
- Async coroutines driven via `asyncio.run` — matches the existing
  project pattern (no pytest-asyncio plugin in CI; see test_init.py
  failure mode).

### nip44.py — docstring update

Added a "Runtime status (post lnbits PR #38, 2026-05-31)" section
documenting that runtime usage moved to `signer.nip44_*` and this
module's role narrowed to (a) the LocalSigner transitional fallback
called from `cassette_transport`, and (b) test-only fixtures in
test_nip44_v2.py for spec-vector + bitspire cross-test validation.
"Don't add new runtime call sites here. The signer abstraction is
the path."

## Verification

- 155 passed, 1 pre-existing async-plugin failure unchanged. The 19
  consumer tests cover bunker happy path + LocalSigner fallback +
  bunker error mapping + payload validation + d-tag construction.
- Live smoke against Greg's RemoteBunkerSigner-migrated account
  on the regtest container: consumer correctly resolves the bunker
  signer, fires `NIP-46 rpc -> method=nip44_decrypt`, catches the
  resulting `NsecBunkerTimeoutError` (the local nsecbunkerd is not
  responding within 15s — separate operational concern), maps to
  `CassetteEventTransientError`, logs at INFO with "will retry next
  poll", and crucially does NOT advance `state_event_id` on the
  cassette_configs rows. Retry semantics preserved.

## Outstanding

- The bunker timeout itself is an operational issue (nsecbunkerd
  config / policy / process state for kind-less nip44_decrypt RPC) —
  not a satmachineadmin code concern; surface to the nsecbunkerd /
  lnbits sessions if it persists.
- Once every operator on the instance is on RemoteBunkerSigner (S7
  fully landed), the `_nip44_*_via_signer` helpers collapse to a
  direct `await signer.nip44_*` call, the LocalSigner fallback can
  be deleted, and `nip44.py`'s runtime exports retire (test-only).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Padreug 2026-05-31 09:21:43 +02:00
commit dcb7de0c27
4 changed files with 573 additions and 199 deletions
Download patch file Download diff file Expand all files Collapse all files

394
tests/test_cassette_state_consumer.py
View file

@ -1,32 +1,49 @@
"""
Tests for the cassette bootstrap consumer (`tasks._handle_cassette_state_event`
and `cassette_transport.decrypt_and_parse_state_event`).
Tests for the cassette bootstrap consumer's transport-decrypt path
(`cassette_transport.decrypt_and_parse_state_event`) and d-tag construction.
Covers the consumer-side validation path end-to-end without standing up
the full nostrclient relay subscription:
- happy path: signed event from a known ATM → decrypt → parse → returns
a position-keyed PublishCassettesPayload
- multiple same-denom cassettes (v1.1 operational case) — round-trips
- tampered ciphertext → CassetteEventDecodeError
- wrong operator privkey → CassetteEventDecodeError (well-formed but
decrypt fails because conversation key is wrong)
- malformed pubkey → CassetteEventDecodeError
- missing fields → CassetteEventDecodeError
- decrypted garbage / wrong-shape JSON → CassetteEventDecodeError
Post-PR-#38 migration (2026-05-31): the function takes an Account +
NostrSigner instead of a raw privkey, and is async. Tests use:
- `_FakeBunkerSigner` — implements async `nip44_decrypt/encrypt` against
the hand-rolled `nip44` impl so tests don't need a live bunker.
Exercises the "happy" RemoteBunkerSigner path.
- `_FakeLocalSignerStub` — raises `SignerUnavailableError` from
`nip44_decrypt`, mimicking the post-#38 `LocalSigner` stub. Combined
with an Account that has `signer_type="LocalSigner"` + `prvkey`,
exercises the transitional fallback path in
`_nip44_decrypt_via_signer`.
- `_FakeRaisingSigner` — raises an arbitrary exception, used to
exercise the `NsecBunkerTimeoutError` → `CassetteEventTransientError`
and `NsecBunkerRpcError` → `CassetteEventDecodeError` mappings.
Full handler tests (the dispatch through verify_event → get_machine_by_atm_
pubkey_hex → apply_bootstrap_state) need a live LNbits DB; they're
smoke-tested manually via the dev container per the project's existing
convention (see test_deposit_currency.py).
Coroutines are driven via `asyncio.run` so no pytest-asyncio config is
required. Matches the existing project test pattern (test_init.py
demonstrates the project lacks an asyncio plugin in CI; using asyncio.run
inside the test body sidesteps that without changing project config).
Full handler tests (the dispatch through verify_event →
get_machine_by_atm_pubkey_hex → apply_bootstrap_state) need a live LNbits
DB; smoke-tested manually via the dev container per the project
convention (see test_deposit_currency.py rationale).
"""
import asyncio
import json
from types import SimpleNamespace
from typing import Optional
import coincurve
import pytest
from lnbits.core.services.nip46_bunker_client import (
NsecBunkerRpcError,
NsecBunkerTimeoutError,
)
from lnbits.core.signers.base import SignerUnavailableError
from ..cassette_transport import (
CassetteEventDecodeError,
CassetteEventTransientError,
_atm_hex_pubkey,
_config_d_tag,
_state_d_tag,
@ -34,7 +51,13 @@ from ..cassette_transport import (
decrypt_and_parse_state_event,
)
from ..models import Machine, PublishCassettesPayload
from ..nip44 import encrypt_with_conversation_key, get_conversation_key
from ..nip44 import (
decrypt_from as _nip44_decrypt,
)
from ..nip44 import (
encrypt_with_conversation_key,
get_conversation_key,
)
# Canonical keys (integer 1 + integer 2, the paulmillr/nip44 reference pair).
@ -54,6 +77,91 @@ _OP_PUB = _pub_hex(_OP_SEC)
_ATM_PUB = _pub_hex(_ATM_SEC)
# =============================================================================
# Fake signers + account-shaped helper
# =============================================================================
class _FakeBunkerSigner:
"""Test double for RemoteBunkerSigner — implements async nip44_*
against the hand-rolled `nip44` impl. Used to exercise the
"signer.nip44_decrypt returns successfully" path without standing up
a live bunker process."""
def __init__(self, privkey_hex: str):
self._privkey_hex = privkey_hex
@property
def pubkey(self) -> str:
return _pub_hex(self._privkey_hex)
def can_sign(self) -> bool:
return True
async def nip44_encrypt(self, plaintext: str, peer_pubkey_hex: str) -> str:
ck = get_conversation_key(self._privkey_hex, peer_pubkey_hex)
return encrypt_with_conversation_key(plaintext, ck)
async def nip44_decrypt(self, ciphertext: str, peer_pubkey_hex: str) -> str:
return _nip44_decrypt(ciphertext, self._privkey_hex, peer_pubkey_hex)
class _FakeLocalSignerStub:
"""Test double for the post-#38 LocalSigner stub — its nip44_* always
raises SignerUnavailableError. Combined with an Account that has
`signer_type='LocalSigner'` + `prvkey` populated, exercises the
transitional fallback in `_nip44_decrypt_via_signer` (which catches
the SignerUnavailableError and falls back to direct-prvkey via the
hand-rolled impl)."""
def can_sign(self) -> bool:
return True
async def nip44_encrypt(self, plaintext: str, peer_pubkey_hex: str) -> str:
raise SignerUnavailableError(
"LocalSigner does not implement nip44_encrypt"
)
async def nip44_decrypt(self, ciphertext: str, peer_pubkey_hex: str) -> str:
raise SignerUnavailableError(
"LocalSigner does not implement nip44_decrypt"
)
class _FakeRaisingSigner:
"""Test double that raises a configurable exception on nip44_decrypt.
Used to validate the bunker-error-mapping branches in
decrypt_and_parse_state_event."""
def __init__(self, exc):
self._exc = exc
def can_sign(self) -> bool:
return True
async def nip44_encrypt(self, plaintext: str, peer_pubkey_hex: str) -> str:
raise self._exc
async def nip44_decrypt(self, ciphertext: str, peer_pubkey_hex: str) -> str:
raise self._exc
def _fake_account(
signer_type: str = "RemoteBunkerSigner",
prvkey: Optional[str] = None,
):
"""Account-shaped duck-typed object. decrypt_and_parse_state_event +
_nip44_decrypt_via_signer only read `.signer_type` and `.prvkey`; the
rest is irrelevant."""
return SimpleNamespace(
id="test-operator",
pubkey=_OP_PUB,
prvkey=prvkey,
signer_type=signer_type,
signer_config=None,
)
def _make_state_event(
payload: PublishCassettesPayload,
*,
@ -63,10 +171,9 @@ def _make_state_event(
event_id: str = "fake-event-id",
created_at: int = 1234567890,
) -> dict:
"""Build a state event the way bitspire's ATM publisher would.
Skips the actual sig-verify step (the handler-level test does
that against verify_event); the transport-level decrypt path
doesn't care about sig validity, only about the conversation key."""
"""Build a state event the way bitspire's ATM publisher would. Skips
the sig-verify step (handler-level concern); the transport-decrypt
path doesn't depend on sig validity, only on conversation-key match."""
plaintext = json.dumps(payload.to_wire_dict(), separators=(",", ":"))
ck = get_conversation_key(atm_sec, op_pub)
content = encrypt_with_conversation_key(plaintext, ck)
@ -84,17 +191,16 @@ def _make_state_event(
# =============================================================================
# decrypt_and_parse_state_event — transport-decrypt path
# decrypt_and_parse_state_event — RemoteBunkerSigner happy path
# =============================================================================
class TestDecryptAndParseStateEvent:
"""The function the consumer task calls per inbound event. Verifies
NIP-44 v2 decrypt + JSON-parse + PublishCassettesPayload validation.
Sig verification is the caller's responsibility (the handler does it
before reaching here)."""
class TestDecryptViaBunkerSigner:
"""The expected production path post-#38: operator account is bunker-
backed, signer.nip44_decrypt routes through the bunker (mocked here
via _FakeBunkerSigner), and the wire payload round-trips cleanly."""
def test_happy_path(self):
def test_happy_path_recovers_positions_keyed_payload(self):
payload = PublishCassettesPayload(
positions={
"1": {"denomination": 20, "count": 49},
@ -102,7 +208,12 @@ class TestDecryptAndParseStateEvent:
}
)
event = _make_state_event(payload)
recovered = decrypt_and_parse_state_event(event, _OP_SEC)
account = _fake_account(signer_type="RemoteBunkerSigner")
signer = _FakeBunkerSigner(_OP_SEC)
recovered = asyncio.run(
decrypt_and_parse_state_event(event, account, signer)
)
assert sorted(recovered.positions.keys()) == [1, 2]
assert recovered.positions[1].denomination == 20
assert recovered.positions[1].count == 49
@ -110,8 +221,8 @@ class TestDecryptAndParseStateEvent:
assert recovered.positions[2].count == 100
def test_round_trips_multiple_same_denomination(self):
"""v1.1 operational case from coord-log 18:45Z: real machines
load multiple cassettes with the same denomination."""
"""v1.1 operational case (coord-log 2026-05-30T18:45Z) — multiple
bays carrying the same denomination."""
payload = PublishCassettesPayload(
positions={
"1": {"denomination": 20, "count": 100},
@ -121,44 +232,166 @@ class TestDecryptAndParseStateEvent:
}
)
event = _make_state_event(payload)
recovered = decrypt_and_parse_state_event(event, _OP_SEC)
account = _fake_account()
signer = _FakeBunkerSigner(_OP_SEC)
recovered = asyncio.run(
decrypt_and_parse_state_event(event, account, signer)
)
assert len(recovered.positions) == 4
for pos in (1, 2, 3, 4):
assert recovered.positions[pos].denomination == 20
assert recovered.positions[pos].count == 100
# =============================================================================
# decrypt_and_parse_state_event — LocalSigner transitional fallback
# =============================================================================
class TestDecryptViaLocalSignerFallback:
"""When the operator account is still on LocalSigner (pre-bunker
migration), the LocalSigner stub raises SignerUnavailableError from
nip44_decrypt. `_nip44_decrypt_via_signer` catches that and falls
back to the hand-rolled impl using `account.prvkey`. Same wire
output; transitional until S7 retires LocalSigner accounts entirely."""
def test_localsigner_with_prvkey_decrypts_via_fallback(self):
payload = PublishCassettesPayload(
positions={"1": {"denomination": 20, "count": 49}}
)
event = _make_state_event(payload)
account = _fake_account(signer_type="LocalSigner", prvkey=_OP_SEC)
signer = _FakeLocalSignerStub()
recovered = asyncio.run(
decrypt_and_parse_state_event(event, account, signer)
)
assert recovered.positions[1].denomination == 20
assert recovered.positions[1].count == 49
def test_localsigner_without_prvkey_raises_decode_error(self):
"""A LocalSigner account whose prvkey field is None (impossible
in practice — LocalSigner requires prvkey at provision time, but
defensive in case the row is corrupt) should surface as a
decode error, not silently succeed."""
payload = PublishCassettesPayload(
positions={"1": {"denomination": 20, "count": 49}}
)
event = _make_state_event(payload)
account = _fake_account(signer_type="LocalSigner", prvkey=None)
signer = _FakeLocalSignerStub()
with pytest.raises(CassetteEventDecodeError):
asyncio.run(
decrypt_and_parse_state_event(event, account, signer)
)
def test_clientonlysigner_raises_decode_error(self):
"""ClientSideOnlySigner has no server-side decrypt path at all;
falling back to direct-prvkey is also impossible (no prvkey).
Surface as a decode error so the consumer logs + skips."""
payload = PublishCassettesPayload(
positions={"1": {"denomination": 20, "count": 49}}
)
event = _make_state_event(payload)
account = _fake_account(
signer_type="ClientSideOnlySigner", prvkey=None
)
signer = _FakeLocalSignerStub() # behaves the same way: raises
with pytest.raises(CassetteEventDecodeError):
asyncio.run(
decrypt_and_parse_state_event(event, account, signer)
)
# =============================================================================
# decrypt_and_parse_state_event — bunker error mapping
# =============================================================================
class TestBunkerErrorMapping:
"""The post-#38 error hierarchy splits transient (bunker partitioned)
from terminal (bunker policy reject, MAC failure). Consumer behaves
differently — transient retries, terminal logs + skips. Validate the
mapping from NsecBunker* exceptions to our CassetteEvent* types."""
def test_timeout_maps_to_transient_error(self):
"""Bunker unreachable → NsecBunkerTimeoutError → caller-visible
CassetteEventTransientError. Consumer treats this as retry-
eligible (don't advance state_event_id)."""
payload = PublishCassettesPayload(
positions={"1": {"denomination": 20, "count": 49}}
)
event = _make_state_event(payload)
account = _fake_account()
signer = _FakeRaisingSigner(
NsecBunkerTimeoutError("bunker unreachable")
)
with pytest.raises(CassetteEventTransientError):
asyncio.run(
decrypt_and_parse_state_event(event, account, signer)
)
def test_rpc_reject_maps_to_decode_error(self):
"""Bunker rejected the RPC (policy / MAC / config) →
NsecBunkerRpcError → caller-visible CassetteEventDecodeError.
Terminal — retrying won't help."""
payload = PublishCassettesPayload(
positions={"1": {"denomination": 20, "count": 49}}
)
event = _make_state_event(payload)
account = _fake_account()
signer = _FakeRaisingSigner(
NsecBunkerRpcError("bunker policy reject: kind 30078 not authorised")
)
with pytest.raises(CassetteEventDecodeError):
asyncio.run(
decrypt_and_parse_state_event(event, account, signer)
)
# =============================================================================
# decrypt_and_parse_state_event — payload + envelope validation
# =============================================================================
class TestPayloadValidation:
"""Errors that originate at the parse layer (post-decrypt), not the
signer. Same set as pre-migration — covered through the bunker-signer
path since LocalSigner is going away."""
def test_tampered_content_rejected(self):
payload = PublishCassettesPayload(
positions={"1": {"denomination": 20, "count": 49}}
)
event = _make_state_event(payload)
# Flip a base64 character — corrupts the ciphertext or MAC
# depending on where the flip lands.
event["content"] = event["content"][:-2] + "AA"
account = _fake_account()
signer = _FakeBunkerSigner(_OP_SEC)
with pytest.raises(CassetteEventDecodeError):
decrypt_and_parse_state_event(event, _OP_SEC)
asyncio.run(
decrypt_and_parse_state_event(event, account, signer)
)
def test_wrong_operator_privkey_rejected(self):
"""The conversation key derives from operator-privkey + sender-pubkey.
A wrong privkey gives a different conversation key, which yields a
different hmac_key, so MAC verification inside NIP-44 v2 decrypt
fails — surfaced as CassetteEventDecodeError."""
def test_wrong_signer_privkey_rejected(self):
"""Wrong privkey on the signer → wrong conversation key → MAC
verification fails inside nip44_decrypt → surfaces as decode
error (via the hand-rolled Nip44Error since this is the fake
bunker signer; in production the bunker would raise
NsecBunkerRpcError which also maps to CassetteEventDecodeError)."""
payload = PublishCassettesPayload(
positions={"1": {"denomination": 20, "count": 49}}
)
event = _make_state_event(payload)
account = _fake_account()
wrong_sec = "00" * 31 + "03"
signer = _FakeBunkerSigner(wrong_sec)
with pytest.raises(CassetteEventDecodeError):
decrypt_and_parse_state_event(event, wrong_sec)
def test_malformed_sender_pubkey_rejected(self):
payload = PublishCassettesPayload(
positions={"1": {"denomination": 20, "count": 49}}
)
event = _make_state_event(payload)
event["pubkey"] = "not-a-real-pubkey"
with pytest.raises(CassetteEventDecodeError):
decrypt_and_parse_state_event(event, _OP_SEC)
asyncio.run(
decrypt_and_parse_state_event(event, account, signer)
)
def test_missing_content_rejected(self):
event = _make_state_event(
@ -167,8 +400,12 @@ class TestDecryptAndParseStateEvent:
)
)
del event["content"]
account = _fake_account()
signer = _FakeBunkerSigner(_OP_SEC)
with pytest.raises(CassetteEventDecodeError):
decrypt_and_parse_state_event(event, _OP_SEC)
asyncio.run(
decrypt_and_parse_state_event(event, account, signer)
)
def test_missing_pubkey_rejected(self):
event = _make_state_event(
@ -177,14 +414,18 @@ class TestDecryptAndParseStateEvent:
)
)
del event["pubkey"]
account = _fake_account()
signer = _FakeBunkerSigner(_OP_SEC)
with pytest.raises(CassetteEventDecodeError):
decrypt_and_parse_state_event(event, _OP_SEC)
asyncio.run(
decrypt_and_parse_state_event(event, account, signer)
)
def test_decrypted_garbage_json_rejected(self):
"""If the plaintext decrypts but isn't JSON, we surface an error
rather than crashing the consumer loop."""
"""If plaintext decrypts cleanly but isn't valid JSON, surface
as decode error (not crash the consumer loop)."""
ck = get_conversation_key(_ATM_SEC, _OP_PUB)
bad_plaintext_event = {
event = {
"kind": 30078,
"pubkey": _ATM_PUB,
"content": encrypt_with_conversation_key(
@ -194,38 +435,42 @@ class TestDecryptAndParseStateEvent:
"created_at": 0,
"id": "x",
}
with pytest.raises(CassetteEventDecodeError) as exc:
decrypt_and_parse_state_event(bad_plaintext_event, _OP_SEC)
assert "JSON" in str(exc.value) or "didn't validate" in str(exc.value)
account = _fake_account()
signer = _FakeBunkerSigner(_OP_SEC)
with pytest.raises(CassetteEventDecodeError):
asyncio.run(
decrypt_and_parse_state_event(event, account, signer)
)
def test_decrypted_json_with_wrong_shape_rejected(self):
"""Well-formed JSON but missing the 'positions' field is
a payload-shape failure, not a decrypt failure."""
def test_decrypted_wrong_shape_rejected(self):
"""Well-formed JSON but missing 'positions' → payload-shape
validation failure."""
ck = get_conversation_key(_ATM_SEC, _OP_PUB)
bad_shape_event = {
event = {
"kind": 30078,
"pubkey": _ATM_PUB,
"content": encrypt_with_conversation_key(
'{"wrong_field": 42}', ck
),
"content": encrypt_with_conversation_key('{"wrong_field": 42}', ck),
"tags": [],
"created_at": 0,
"id": "x",
}
with pytest.raises(CassetteEventDecodeError) as exc:
decrypt_and_parse_state_event(bad_shape_event, _OP_SEC)
assert "didn't validate" in str(exc.value)
account = _fake_account()
signer = _FakeBunkerSigner(_OP_SEC)
with pytest.raises(CassetteEventDecodeError):
asyncio.run(
decrypt_and_parse_state_event(event, account, signer)
)
# =============================================================================
# d-tag construction — _atm_hex_pubkey, _config_d_tag, _state_d_tag, helper
# d-tag construction — unchanged by the migration, kept as regression guard
# =============================================================================
class TestDTagConstruction:
"""The `<m>` placeholder in d-tags = ATM hex pubkey (load-bearing per
coord-log 11:50Z). These tests pin the canonical substitution so a
refactor can't silently break wire compatibility."""
coord-log 2026-05-30T11:50Z). These tests pin the canonical
substitution so a refactor can't silently break wire compatibility."""
def _machine(self, npub: str, id_: str = "m1") -> Machine:
from datetime import datetime, timezone
@ -251,8 +496,6 @@ class TestDTagConstruction:
assert _atm_hex_pubkey(self._machine(_ATM_PUB.upper())) == _ATM_PUB
def test_atm_hex_pubkey_canonicalises_bech32_to_hex(self):
"""Operator may have entered npub1... in the UI; canonical d-tag
substitution is always the hex form."""
from lnbits.utils.nostr import hex_to_npub
npub_bech32 = hex_to_npub(_ATM_PUB)
@ -260,8 +503,7 @@ class TestDTagConstruction:
def test_config_d_tag_uses_hex_pubkey_not_id(self):
"""REGRESSION GUARD: d-tag must contain the ATM hex pubkey, NOT
the internal machine UUID. If this test fails, bitspire's ATM
won't see our publishes."""
the internal machine UUID."""
m = self._machine(_ATM_PUB, id_="some-uuid-not-the-pubkey")
d_tag = _config_d_tag(_atm_hex_pubkey(m))
assert d_tag == f"bitspire-cassettes:{_ATM_PUB}"