Padreug
8dad72a00d
Some checks failed
ci.yml / fix: complete the unpaired-machine sweep + regression test (pull_request) Failing after 0s
Full sweep of every machine_npub deref found one more reachable crash: _record_rejected (tasks.py) logs machine_npub[:12], and the assert_nostr_attribution guard now routes an unpaired machine there, so None[:12] -> TypeError. Fall back to machine.id. Every other deref is safe by the attribution-gate invariant: a settlement only flows past assert_nostr_attribution (now rejecting unpaired) for a paired machine, so the downstream distribution / parse-path / "landed" logs can't see None; the collision-loop display already uses `(m.machine_npub or m.id)`. Adds tests/test_unpaired_machine_guards.py: attribution rejects an unpaired machine with the domain SettlementAttributionError (not AttributeError), and build_state_d_tags skips it. New tests + every guard-affected suite pass. (Two pre-existing test_pair_endpoint failures — #29 drift: fake_pair lacks bunker_relay, and the test DB lacks super_config — are out of scope; filed separately.)
57 lines
2.1 KiB
Python
57 lines
2.1 KiB
Python
"""
|
|
Regression: `machine_npub` is nullable (#29/m011 register-unpaired flow), so
|
|
every consumer that derives a Nostr identity from it must handle `None` rather
|
|
than crash `normalize_public_key(None)` (AttributeError: 'NoneType' has no
|
|
'startswith') or `machine_npub[:12]` (TypeError). See PR #33 — an unpaired
|
|
machine on the demo broke the platform-fee update (500) and the cassette
|
|
consumer.
|
|
|
|
These cover the pure-function guards; the DB-backed loops
|
|
(get_machine_by_atm_pubkey_hex, the super-config republish loop) are exercised
|
|
on the dev stack with an unpaired active machine.
|
|
"""
|
|
|
|
from datetime import datetime, timezone
|
|
|
|
import pytest
|
|
|
|
from ..bitspire import SettlementAttributionError, assert_nostr_attribution
|
|
from ..cassette_transport import build_state_d_tags_for_machines
|
|
from ..models import Machine
|
|
|
|
_PAIRED_HEX = "82341f882b6eabcbd6b1c2da5cd14df14b8e91dd0e6da41a72b78ad8f3a7d3b9"
|
|
|
|
|
|
def _machine(npub: str | None) -> Machine:
|
|
now = datetime.now(timezone.utc)
|
|
return Machine(
|
|
id="unpaired1",
|
|
operator_user_id="op1",
|
|
machine_npub=npub,
|
|
wallet_id="w1",
|
|
name="unpaired",
|
|
location=None,
|
|
fiat_code="EUR",
|
|
is_active=True,
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
|
|
|
|
def test_attribution_rejects_unpaired_machine_cleanly():
|
|
"""An unpaired machine must raise the domain SettlementAttributionError
|
|
(which the listener records as 'rejected'), not an uncaught AttributeError
|
|
from normalize_public_key(None)."""
|
|
with pytest.raises(SettlementAttributionError):
|
|
assert_nostr_attribution(
|
|
_machine(None),
|
|
{"source": "bitspire", "nostr_sender_pubkey": _PAIRED_HEX},
|
|
)
|
|
|
|
|
|
def test_cassette_d_tags_skip_unpaired_machine():
|
|
"""build_state_d_tags_for_machines must skip unpaired machines rather than
|
|
crash _atm_hex_pubkey on a None npub — the cassette-consumer loop crash."""
|
|
tags = build_state_d_tags_for_machines([_machine(_PAIRED_HEX), _machine(None)])
|
|
assert len(tags) == 1 # only the paired machine contributes a d-tag
|
|
assert all("None" not in t for t in tags)
|