satmachineadmin/tests/test_cassette_configs.py

"""
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 satmachineadmin 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