Padreug
a059e3f596
Fork of satmachineadmin's v2-bitspire line into its own repo. Renames
both identifiers so this extension is fully independent of the original
satmachineadmin install (which remains in service):
- extension id satmachineadmin -> spirekeeper
(router prefix, static path/static_url_for, module symbols, task
names, templates dir, config/manifest paths)
- database name satoshimachine -> spirekeeper
(Database(ext_spirekeeper), all schema-qualified table refs)
Also resets versioning to 0.1.0, sets the display name + manifest to
spirekeeper/aiolabs, and fixes the placeholder pyproject description.
Historical aiolabs/satmachineadmin#N issue references in comments are
left pointing at the original repo where those issues live.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
220 lines
9.3 KiB
Python
220 lines
9.3 KiB
Python
"""
|
|
Tests for the v1.1 cassette-config layer (aiolabs/satmachineadmin#29).
|
|
|
|
Covers the pure pieces that don't need a live DB:
|
|
- Pydantic validator behaviour on PublishCassettesPayload + the row /
|
|
upsert models (position key coercion, integer ranges, multiple-same-
|
|
denomination payloads, wire-format round-trip)
|
|
- _should_apply_bootstrap_state dedup helper (extracted from
|
|
apply_bootstrap_state so the relay-re-delivery decision is testable
|
|
without a database round-trip)
|
|
|
|
DB-touching tests (apply_bootstrap_state actually upserting, list-by-
|
|
machine ordering, etc.) follow the project convention from
|
|
test_deposit_currency.py: "Layer 2 is an endpoint-level behaviour better
|
|
covered by an integration test against a running LNbits; tracked in #26
|
|
as a follow-up." Smoke-tested manually via the dev container.
|
|
|
|
Wire shape pivot from m007 → m008 is the v1.1 coordination point per
|
|
coord-log 2026-05-30T18:30Z + 18:45Z: position is the row identity,
|
|
denomination + count are operator-editable per row, multiple same-denom
|
|
cassettes are valid.
|
|
"""
|
|
|
|
import pytest
|
|
|
|
from ..crud import _should_apply_bootstrap_state
|
|
from ..models import (
|
|
CassettePayloadRow,
|
|
PublishCassettesPayload,
|
|
UpsertCassetteConfigData,
|
|
)
|
|
|
|
# =============================================================================
|
|
# PublishCassettesPayload — wire-shape validators
|
|
# =============================================================================
|
|
|
|
|
|
class TestPublishCassettesPayload:
|
|
"""The kind-30078 content payload, bidirectional (operator→ATM and
|
|
ATM→operator share the shape). String JSON keys must coerce to int;
|
|
per-row int constraints enforced; multiple same-denom rows are valid."""
|
|
|
|
def test_happy_path_coerces_string_keys_to_int(self):
|
|
p = PublishCassettesPayload(
|
|
positions={
|
|
"1": {"denomination": 20, "count": 49},
|
|
"2": {"denomination": 50, "count": 100},
|
|
}
|
|
)
|
|
assert set(p.positions.keys()) == {1, 2}
|
|
assert p.positions[1].denomination == 20
|
|
assert p.positions[1].count == 49
|
|
assert p.positions[2].denomination == 50
|
|
assert p.positions[2].count == 100
|
|
|
|
def test_wire_dict_round_trip_restringifies_keys(self):
|
|
"""to_wire_dict() must restringify position keys so the resulting
|
|
JSON is parseable by clients (including the ATM-side nostr-tools
|
|
NIP-44 v2 consumer per the byte-compat cross-test)."""
|
|
original = PublishCassettesPayload(
|
|
positions={
|
|
"1": {"denomination": 20, "count": 49},
|
|
"2": {"denomination": 50, "count": 100},
|
|
}
|
|
)
|
|
wire = original.to_wire_dict()
|
|
assert wire == {
|
|
"positions": {
|
|
"1": {"denomination": 20, "count": 49},
|
|
"2": {"denomination": 50, "count": 100},
|
|
}
|
|
}
|
|
# And the wire form round-trips back through the parser cleanly.
|
|
reparsed = PublishCassettesPayload(**wire)
|
|
assert reparsed.positions == original.positions
|
|
|
|
def test_accepts_multiple_same_denomination_cassettes(self):
|
|
"""v1.1 operational case: real machines have N cassettes loaded
|
|
with the same denomination for cash-out throughput. The wire shape
|
|
must accept this, and we explicitly do NOT validate uniqueness on
|
|
denomination. Coord-log 2026-05-30T18:45Z bitspire response."""
|
|
p = PublishCassettesPayload(
|
|
positions={
|
|
"1": {"denomination": 20, "count": 100},
|
|
"2": {"denomination": 20, "count": 100},
|
|
"3": {"denomination": 50, "count": 50},
|
|
"4": {"denomination": 100, "count": 25},
|
|
}
|
|
)
|
|
assert len(p.positions) == 4
|
|
denoms = [row.denomination for row in p.positions.values()]
|
|
assert denoms.count(20) == 2 # two $20 cassettes
|
|
assert sorted(denoms) == [20, 20, 50, 100]
|
|
|
|
def test_rejects_non_int_position_key(self):
|
|
with pytest.raises(ValueError) as exc:
|
|
PublishCassettesPayload(positions={"abc": {"denomination": 20, "count": 1}})
|
|
assert "is not an int" in str(exc.value)
|
|
|
|
def test_rejects_non_positive_position(self):
|
|
with pytest.raises(ValueError) as exc:
|
|
PublishCassettesPayload(positions={"0": {"denomination": 20, "count": 1}})
|
|
assert "position must be > 0" in str(exc.value)
|
|
|
|
def test_rejects_negative_position(self):
|
|
with pytest.raises(ValueError) as exc:
|
|
PublishCassettesPayload(positions={"-1": {"denomination": 20, "count": 1}})
|
|
assert "position must be > 0" in str(exc.value)
|
|
|
|
def test_rejects_negative_count(self):
|
|
with pytest.raises(ValueError):
|
|
PublishCassettesPayload(positions={"1": {"denomination": 20, "count": -1}})
|
|
|
|
def test_rejects_zero_denomination(self):
|
|
with pytest.raises(ValueError):
|
|
PublishCassettesPayload(positions={"1": {"denomination": 0, "count": 49}})
|
|
|
|
def test_rejects_negative_denomination(self):
|
|
with pytest.raises(ValueError):
|
|
PublishCassettesPayload(positions={"1": {"denomination": -20, "count": 49}})
|
|
|
|
def test_allows_zero_count(self):
|
|
"""An empty cassette is a legal state — operator must be able to
|
|
record `count=0` after a dispatcher pulled the cassette mid-day."""
|
|
p = PublishCassettesPayload(positions={"1": {"denomination": 20, "count": 0}})
|
|
assert p.positions[1].count == 0
|
|
|
|
|
|
# =============================================================================
|
|
# CassettePayloadRow — per-row int constraints
|
|
# =============================================================================
|
|
|
|
|
|
class TestCassettePayloadRow:
|
|
def test_happy_path(self):
|
|
row = CassettePayloadRow(denomination=20, count=49)
|
|
assert row.denomination == 20
|
|
assert row.count == 49
|
|
|
|
@pytest.mark.parametrize("bad_denom", [0, -1, -100])
|
|
def test_rejects_non_positive_denomination(self, bad_denom):
|
|
with pytest.raises(ValueError):
|
|
CassettePayloadRow(denomination=bad_denom, count=1)
|
|
|
|
def test_rejects_negative_count(self):
|
|
with pytest.raises(ValueError):
|
|
CassettePayloadRow(denomination=20, count=-1)
|
|
|
|
|
|
# =============================================================================
|
|
# UpsertCassetteConfigData — operator-edit form
|
|
# =============================================================================
|
|
|
|
|
|
class TestUpsertCassetteConfigData:
|
|
"""Operator-driven row edit. Both fields optional; same int constraints
|
|
as the wire-format row but applied independently per-edit. Position is
|
|
NOT editable — it's the row's identity (the hardware bay number)."""
|
|
|
|
def test_partial_update_count_only(self):
|
|
d = UpsertCassetteConfigData(count=80)
|
|
assert d.count == 80
|
|
assert d.denomination is None
|
|
|
|
def test_partial_update_denomination_only(self):
|
|
"""v1.1 operational case: operator records a cartridge swap at
|
|
refill — slot 1 was $20, dispatcher replaced with $50."""
|
|
d = UpsertCassetteConfigData(denomination=50)
|
|
assert d.denomination == 50
|
|
assert d.count is None
|
|
|
|
def test_empty_update_is_legal(self):
|
|
"""An empty UpsertCassetteConfigData parses fine; the CRUD short-
|
|
circuits a no-op on empty payload (no SQL emitted)."""
|
|
d = UpsertCassetteConfigData()
|
|
assert d.count is None
|
|
assert d.denomination is None
|
|
|
|
def test_rejects_negative_count(self):
|
|
with pytest.raises(ValueError):
|
|
UpsertCassetteConfigData(count=-1)
|
|
|
|
def test_rejects_non_positive_denomination(self):
|
|
with pytest.raises(ValueError):
|
|
UpsertCassetteConfigData(denomination=0)
|
|
|
|
|
|
# =============================================================================
|
|
# _should_apply_bootstrap_state — relay re-delivery dedup
|
|
# =============================================================================
|
|
|
|
|
|
class TestShouldApplyBootstrapState:
|
|
"""Pure-function dedup gate extracted from apply_bootstrap_state so the
|
|
decision is testable without a DB. Logic: apply if-and-only-if the
|
|
existing row's state_event_id differs from the incoming event_id.
|
|
|
|
In v1.1 the ATM publishes the bootstrap event exactly once per machine,
|
|
so this is sufficient for replay protection. v2 will need a
|
|
`last_state_created_at` watermark in addition (per bitspire's
|
|
`meta.lastKnownConfigCreatedAt` on the ATM side) — flagged in #29's
|
|
v2 forward-look section.
|
|
"""
|
|
|
|
def test_applies_when_no_existing_row(self):
|
|
assert _should_apply_bootstrap_state(None, "new-event-id") is True
|
|
|
|
def test_applies_when_existing_event_id_differs(self):
|
|
assert _should_apply_bootstrap_state("old-event-id", "new-event-id") is True
|
|
|
|
def test_skips_when_existing_event_id_matches(self):
|
|
"""The same bootstrap event re-delivered after a relay reconnect
|
|
or spirekeeper restart should no-op, not re-upsert the same
|
|
rows (which would clobber any operator edits since)."""
|
|
assert _should_apply_bootstrap_state("same-event", "same-event") is False
|
|
|
|
def test_applies_when_existing_is_empty_string_and_incoming_is_id(self):
|
|
"""Defensive — a sentinel empty-string existing_state_event_id
|
|
shouldn't block a real incoming event from applying."""
|
|
assert _should_apply_bootstrap_state("", "real-event-id") is True
|