aiolabs/spirekeeper
8
0
Fork 0
Code Issues 23 Pull requests Projects Releases Packages Wiki Activity Actions

test(v2): rewrite cassette tests for v1.1 position-keyed wire shape (#29 v1.1)

Browse source 
The wire-shape pivot (m007 denomination-keyed → m008 position-keyed)
needs the unit test surface re-written to match:

  test_cassette_configs.py
    - PublishCassettesPayload tests pivot to positions-keyed input.
      Validators reject non-int / non-positive position keys, negative
      denom, negative count. Zero count allowed (empty cassette).
    - NEW: test_accepts_multiple_same_denomination_cassettes — pins the
      v1.1 operational requirement (real machines load 4×$20 for cash-out
      throughput) per coord-log 18:45Z. No denom-unique validator.
    - CassettePayloadRow tests pivot to the new field shape
      (denomination + count, no position).
    - UpsertCassetteConfigData tests cover edit-denomination (the v1.1
      "operator swaps a cartridge during refill" scenario) and edit-count.
      Position no longer in the model.

  test_cassette_state_consumer.py
    - _make_state_event helper builds {"positions": {...}} ciphertext.
    - Happy-path assertion checks p.positions keys + denomination/count
      per row.
    - NEW: test_round_trips_multiple_same_denomination — covers the v1.1
      four-of-the-same case through encrypt → decrypt → parse.
    - All negative paths (tamper, wrong privkey, malformed pubkey,
      missing fields, garbage JSON, wrong shape) carry over with the new
      payload shape.
    - d-tag tests unchanged (position vs denomination isn't on the d-tag).

  test_nip44_v2.py
    - TestBitspireCrossTest temporarily re-skipped at the class level: the
      13:15Z fixture is encoded with the v1 denomination-keyed shape;
      bitspire's posting a v1.1 fixture and commit g will swap +
      unskip.

Total: 148 passed, 3 skipped (bitspire cross-test pending the v1.1
fixture from bitspire), 1 pre-existing async-plugin failure unchanged.

Branch tip is now functionally green (the pre-existing async failure
predates this PR + can't be addressed without a pytest plugin install).
Pending commit g for the cross-test fixture re-wire when bitspire posts.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Padreug 2026-05-30 22:31:08 +02:00
commit 1cebefcde5
3 changed files with 132 additions and 80 deletions
Download patch file Download diff file Expand all files Collapse all files

147
tests/test_cassette_configs.py
View file

@ -1,10 +1,10 @@
"""
Tests for the v1 cassette-config layer (aiolabs/satmachineadmin#29).
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 (denomination key coercion, integer ranges, no-duplicate-
positions, wire-format round-trip)
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)
@ -14,6 +14,11 @@ 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
@ -34,116 +39,127 @@ from ..models import (
class TestPublishCassettesPayload:
"""The kind-30078 content payload, bidirectional (operator→ATM and
ATM→operator share the shape). String JSON keys must coerce to int;
duplicate positions must reject; per-row int constraints enforced."""
per-row int constraints enforced; multiple same-denom rows are valid."""
def test_happy_path_coerces_string_keys_to_int(self):
p = PublishCassettesPayload(
denominations={
"20": {"position": 1, "count": 49},
"50": {"position": 2, "count": 100},
positions={
"1": {"denomination": 20, "count": 49},
"2": {"denomination": 50, "count": 100},
}
)
assert set(p.denominations.keys()) == {20, 50}
assert p.denominations[20].position == 1
assert p.denominations[20].count == 49
assert p.denominations[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 denomination 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)."""
"""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(
denominations={
"20": {"position": 1, "count": 49},
"50": {"position": 2, "count": 100},
positions={
"1": {"denomination": 20, "count": 49},
"2": {"denomination": 50, "count": 100},
}
)
wire = original.to_wire_dict()
assert wire == {
"denominations": {
"20": {"position": 1, "count": 49},
"50": {"position": 2, "count": 100},
"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.denominations == original.denominations
assert reparsed.positions == original.positions
def test_rejects_non_int_key(self):
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(
denominations={"abc": {"position": 1, "count": 1}}
positions={"abc": {"denomination": 20, "count": 1}}
)
assert "is not an int" in str(exc.value)
def test_rejects_non_positive_denomination(self):
def test_rejects_non_positive_position(self):
with pytest.raises(ValueError) as exc:
PublishCassettesPayload(
denominations={"0": {"position": 1, "count": 1}}
positions={"0": {"denomination": 20, "count": 1}}
)
assert "denomination must be > 0" in str(exc.value)
assert "position must be > 0" in str(exc.value)
def test_rejects_negative_denomination(self):
def test_rejects_negative_position(self):
with pytest.raises(ValueError) as exc:
PublishCassettesPayload(
denominations={"-20": {"position": 1, "count": 1}}
positions={"-1": {"denomination": 20, "count": 1}}
)
assert "denomination must be > 0" in str(exc.value)
def test_rejects_duplicate_position(self):
"""Two cassettes can't occupy the same physical slot. The schema
PK is (machine_id, denomination), so duplicates land via the
payload; reject at the validator layer before the publish path
builds an event the ATM will misinterpret."""
with pytest.raises(ValueError) as exc:
PublishCassettesPayload(
denominations={
"20": {"position": 1, "count": 49},
"50": {"position": 1, "count": 100},
}
)
assert "duplicate position" in str(exc.value)
assert "position must be > 0" in str(exc.value)
def test_rejects_negative_count(self):
with pytest.raises(ValueError):
PublishCassettesPayload(
denominations={"20": {"position": 1, "count": -1}}
positions={"1": {"denomination": 20, "count": -1}}
)
def test_rejects_zero_position(self):
def test_rejects_zero_denomination(self):
with pytest.raises(ValueError):
PublishCassettesPayload(
denominations={"20": {"position": 0, "count": 1}}
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(
denominations={"20": {"position": 1, "count": 0}}
positions={"1": {"denomination": 20, "count": 0}}
)
assert p.denominations[20].count == 0
assert p.positions[1].count == 0
# =============================================================================
# CassettePayloadRow — per-row int constraints (single-row tests)
# CassettePayloadRow — per-row int constraints
# =============================================================================
class TestCassettePayloadRow:
def test_happy_path(self):
row = CassettePayloadRow(position=1, count=49)
assert row.position == 1
row = CassettePayloadRow(denomination=20, count=49)
assert row.denomination == 20
assert row.count == 49
@pytest.mark.parametrize("bad_position", [0, -1, -100])
def test_rejects_non_positive_position(self, bad_position):
@pytest.mark.parametrize("bad_denom", [0, -1, -100])
def test_rejects_non_positive_denomination(self, bad_denom):
with pytest.raises(ValueError):
CassettePayloadRow(position=bad_position, count=1)
CassettePayloadRow(denomination=bad_denom, count=1)
def test_rejects_negative_count(self):
with pytest.raises(ValueError):
CassettePayloadRow(position=1, count=-1)
CassettePayloadRow(denomination=20, count=-1)
# =============================================================================
@ -153,16 +169,19 @@ class TestCassettePayloadRow:
class TestUpsertCassetteConfigData:
"""Operator-driven row edit. Both fields optional; same int constraints
as the wire-format row but applied independently per-edit."""
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.position is None
assert d.denomination is None
def test_partial_update_position_only(self):
d = UpsertCassetteConfigData(position=3)
assert d.position == 3
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):
@ -170,15 +189,15 @@ class TestUpsertCassetteConfigData:
circuits a no-op on empty payload (no SQL emitted)."""
d = UpsertCassetteConfigData()
assert d.count is None
assert d.position 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_position(self):
def test_rejects_non_positive_denomination(self):
with pytest.raises(ValueError):
UpsertCassetteConfigData(position=0)
UpsertCassetteConfigData(denomination=0)
# =============================================================================
@ -191,7 +210,7 @@ class TestShouldApplyBootstrapState:
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 the ATM publishes the bootstrap event exactly once per machine,
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