test: slice-2 coverage — NIP-17 round trip, alert policy, settle path, op shape

tests/test_slice2.py

@@ -0,0 +1,424 @@
+"""
+ADR-005 slice 2: operator alerts, off-machine settlement, the
+settle_transaction op, alerts_pubkey.
+
+  - NIP-17 builder: a gift wrap (kind 1059) signed by a throwaway key, p-tagged
+    to the recipient, backdated; its content opens with the recipient's key
+    to a seal (kind 13) signed by the sender, which opens to a kind-14 rumor
+    carrying the text. Built with a fake LocalSigner-style signer so no bunker
+    is needed.
+  - notify_cash_outcome is best-effort: publishes once, marks notified, never
+    raises, and does not re-alert a settlement already notified.
+  - apply_report_to_settlement alerts on cash_owed / partial_pending and not on
+    a confirmed dispense.
+  - settle_cash_owed_settlement: note appended, status → pending, distribution
+    run; refuses other statuses.
+  - models: settle_transaction op (machine-wide, carries txid + note, wire
+    shape), alerts_pubkey validator (hex / npub / clear).
+"""
+
+import asyncio
+import json
+from datetime import datetime
+from types import SimpleNamespace
+
+import coincurve
+import pytest
+from lnbits.utils.nostr import verify_event
+from pydantic import ValidationError
+
+from .. import dispense_transport, distribution, notify
+from ..models import (
+    CassetteOp,
+    CreateCassetteOpData,
+    DcaSettlement,
+    DispenseReportIn,
+    Machine,
+    UpdateSuperConfigData,
+)
+from ..nip44 import decrypt_from, encrypt_for
+
+_NOW = datetime(2026, 10, 9, 7, 2, 33)
+_OP_SEC = "00" * 31 + "01"
+_ALERT_SEC = "00" * 31 + "03"
+
+
+def _pub(sec_hex: str) -> str:
+    return (
+        coincurve.PrivateKey(bytes.fromhex(sec_hex)).public_key.format(True)[1:].hex()
+    )
+
+
+_OP_PUB = _pub(_OP_SEC)
+_ALERT_PUB = _pub(_ALERT_SEC)
+
+
+def _machine() -> Machine:
+    return Machine(
+        id="m1",
+        operator_user_id="op1",
+        machine_npub="df20" * 16,
+        wallet_id="w1",
+        name="sintra",
+        location=None,
+        fiat_code="EUR",
+        is_active=True,
+        created_at=_NOW,
+        updated_at=_NOW,
+    )
+
+
+def _settlement(status="cash_owed", notified=None) -> DcaSettlement:
+    return DcaSettlement(
+        id="s1",
+        machine_id="m1",
+        payment_hash="ab" * 32,
+        bitspire_event_id=None,
+        bitspire_txid="tx_mv0madw6_wdhtea1v",
+        wire_sats=54440,
+        fiat_amount=40.0,
+        fiat_code="EUR",
+        exchange_rate=1361.0,
+        principal_sats=54440,
+        fee_sats=0,
+        platform_fee_sats=0,
+        operator_fee_sats=0,
+        tx_type="cash_out",
+        bills_json=None,
+        cassettes_json=None,
+        status=status,
+        error_message=None,
+        processed_at=None,
+        created_at=_NOW,
+        operator_notified_at=notified,
+    )
+
+
+def _report(**over) -> DispenseReportIn:
+    body: dict = {
+        "txid": "tx_mv0madw6_wdhtea1v",
+        "payment_hash": "ab" * 32,
+        "dispense_confirmed": False,
+        "error": "Note stopped at the cassette exit",
+        "error_code": "F56DispenseError",
+        "raw_code": "78 42",
+        "error_class": "terminal",
+        "fiat_cents": 4000,
+        "currency": "EUR",
+        "bills": [{"denomination": 20, "requested": 2, "dispensed": 0, "rejected": 0}],
+        "at": 1791529353,
+    }
+    body.update(over)
+    return DispenseReportIn(**body)
+
+
+# ---------------------------------------------------------------------------
+# A signer that behaves like a LocalSigner holding _OP_SEC, without lnbits' DB
+# ---------------------------------------------------------------------------
+
+
+class _FakeSigner:
+    def can_sign(self):
+        return True
+
+    async def nip44_encrypt(self, plaintext, peer):
+        return encrypt_for(plaintext, _OP_SEC, peer)
+
+    async def sign_event(self, event):
+        from lnbits.utils.nostr import sign_event
+
+        return sign_event(event, _OP_PUB, coincurve.PrivateKey(bytes.fromhex(_OP_SEC)))
+
+
+def _wire_signer(monkeypatch):
+    account = SimpleNamespace(pubkey=_OP_PUB, signer_type="LocalSigner", prvkey=_OP_SEC)
+
+    async def resolve(_uid):
+        return account, _FakeSigner()
+
+    async def sign_as_operator(_uid, event):
+        import time
+
+        event["created_at"] = int(time.time())
+        return await _FakeSigner().sign_event(event)
+
+    monkeypatch.setattr(notify, "resolve_operator_signer", resolve)
+    monkeypatch.setattr(notify, "sign_as_operator", sign_as_operator)
+    return account
+
+
+# ---------------------------------------------------------------------------
+# NIP-17 builder
+# ---------------------------------------------------------------------------
+
+
+class TestGiftWrap:
+    def test_wrap_seal_rumor_round_trip_to_the_recipient(self, monkeypatch):
+        _wire_signer(monkeypatch)
+        wrap = asyncio.run(
+            notify.build_gift_wrapped_dm("op1", _ALERT_PUB, "hello operator")
+        )
+
+        assert wrap["kind"] == 1059
+        assert wrap["tags"] == [["p", _ALERT_PUB]]
+        assert wrap["pubkey"] != _OP_PUB  # throwaway key, not the sender
+        assert verify_event(wrap)
+        import time
+
+        assert wrap["created_at"] <= int(time.time())
+        assert wrap["created_at"] >= int(time.time()) - 2 * 86400 - 5
+
+        seal = json.loads(decrypt_from(wrap["content"], _ALERT_SEC, wrap["pubkey"]))
+        assert seal["kind"] == 13 and seal["pubkey"] == _OP_PUB and seal["tags"] == []
+        assert verify_event(seal)
+
+        rumor = json.loads(decrypt_from(seal["content"], _ALERT_SEC, _OP_PUB))
+        assert rumor["kind"] == 14
+        assert rumor["pubkey"] == _OP_PUB
+        assert rumor["content"] == "hello operator"
+        assert rumor["tags"] == [["p", _ALERT_PUB]]
+        assert "sig" not in rumor and len(rumor["id"]) == 64
+
+    def test_note_to_self_when_recipient_is_the_sender(self, monkeypatch):
+        _wire_signer(monkeypatch)
+        wrap = asyncio.run(notify.build_gift_wrapped_dm("op1", _OP_PUB, "to me"))
+        seal = json.loads(decrypt_from(wrap["content"], _OP_SEC, wrap["pubkey"]))
+        rumor = json.loads(decrypt_from(seal["content"], _OP_SEC, _OP_PUB))
+        assert rumor["content"] == "to me"
+
+
+class TestNotifyCashOutcome:
+    def _wire(self, monkeypatch, *, alerts_pubkey=None, publish_fails=False):
+        _wire_signer(monkeypatch)
+        published, marked = [], []
+
+        async def get_super():
+            return SimpleNamespace(alerts_pubkey=alerts_pubkey)
+
+        async def publish(ev):
+            if publish_fails:
+                from ..nostr_publish import RelayUnavailable
+
+                raise RelayUnavailable("no relay")
+            published.append(ev)
+
+        async def mark(sid):
+            marked.append(sid)
+
+        monkeypatch.setattr(notify, "get_super_config", get_super)
+        monkeypatch.setattr(notify, "publish_signed_event", publish)
+        monkeypatch.setattr(notify, "mark_settlement_notified", mark)
+        return published, marked
+
+    def test_alerts_once_to_the_operator_by_default(self, monkeypatch):
+        published, marked = self._wire(monkeypatch)
+        ok = asyncio.run(
+            notify.notify_cash_outcome(
+                _settlement(), _machine(), _report(), "cash_owed"
+            )
+        )
+        assert ok is True
+        assert marked == ["s1"]
+        assert published[0]["tags"] == [["p", _OP_PUB]]
+        seal = json.loads(
+            decrypt_from(published[0]["content"], _OP_SEC, published[0]["pubkey"])
+        )
+        rumor = json.loads(decrypt_from(seal["content"], _OP_SEC, _OP_PUB))
+        assert "CASH OWED" in rumor["content"]
+        assert "sintra" in rumor["content"]
+        assert "40.00 EUR" in rumor["content"]
+        assert "tx_mv0madw6_wdhtea1v" in rumor["content"]
+        assert "78 42" not in rumor["content"]  # raw code stays in the dashboard
+        assert "out of service" in rumor["content"]  # terminal → latch mentioned
+
+    def test_alerts_pubkey_override_and_partial_wording(self, monkeypatch):
+        published, _ = self._wire(monkeypatch, alerts_pubkey=_ALERT_PUB)
+        rep = _report(
+            bills=[{"denomination": 20, "requested": 2, "dispensed": 1, "rejected": 1}]
+        )
+        asyncio.run(
+            notify.notify_cash_outcome(
+                _settlement("partial_pending"), _machine(), rep, "partial_pending"
+            )
+        )
+        assert published[0]["tags"] == [["p", _ALERT_PUB]]
+        seal = json.loads(
+            decrypt_from(published[0]["content"], _ALERT_SEC, published[0]["pubkey"])
+        )
+        rumor = json.loads(decrypt_from(seal["content"], _ALERT_SEC, _OP_PUB))
+        assert (
+            "PARTIAL" in rumor["content"] and "received 20.00 EUR" in rumor["content"]
+        )
+
+    def test_does_not_re_alert_and_never_raises(self, monkeypatch):
+        published, marked = self._wire(monkeypatch)
+        ok = asyncio.run(
+            notify.notify_cash_outcome(
+                _settlement(notified=_NOW), _machine(), _report(), "cash_owed"
+            )
+        )
+        assert ok is False and published == [] and marked == []
+
+        published, marked = self._wire(monkeypatch, publish_fails=True)
+        ok = asyncio.run(
+            notify.notify_cash_outcome(
+                _settlement(), _machine(), _report(), "cash_owed"
+            )
+        )
+        assert ok is False and marked == []
+
+
+class TestCaptureAlerts:
+    def _wire(self, monkeypatch, status_after):
+        alerts = []
+
+        async def apply(sid, report, new_status, reported_at):
+            return _settlement(new_status)
+
+        async def alert(settlement, machine, report, status):
+            alerts.append(status)
+
+        monkeypatch.setattr(dispense_transport, "apply_dispense_outcome", apply)
+        monkeypatch.setattr(dispense_transport, "_spawn_distribution", lambda sid: None)
+        monkeypatch.setattr(notify, "notify_cash_outcome", alert)
+        return alerts
+
+    def test_alerts_on_owed_and_partial_not_on_confirmed(self, monkeypatch):
+        alerts = self._wire(monkeypatch, "cash_owed")
+        asyncio.run(
+            dispense_transport.apply_report_to_settlement(
+                _settlement("awaiting_dispense"), _report(), _machine()
+            )
+        )
+        rep_partial = _report(
+            bills=[{"denomination": 20, "requested": 2, "dispensed": 1, "rejected": 1}]
+        )
+        asyncio.run(
+            dispense_transport.apply_report_to_settlement(
+                _settlement("awaiting_dispense"), rep_partial, _machine()
+            )
+        )
+        rep_ok = _report(
+            dispense_confirmed=True,
+            error=None,
+            error_code=None,
+            raw_code=None,
+            error_class=None,
+            bills=[{"denomination": 20, "requested": 2, "dispensed": 2, "rejected": 0}],
+        )
+        asyncio.run(
+            dispense_transport.apply_report_to_settlement(
+                _settlement("awaiting_dispense"), rep_ok, _machine()
+            )
+        )
+        assert alerts == ["cash_owed", "partial_pending"]
+
+
+# ---------------------------------------------------------------------------
+# Off-machine settlement
+# ---------------------------------------------------------------------------
+
+
+class TestSettleCashOwed:
+    def _wire(self, monkeypatch):
+        notes, statuses, processed = [], [], []
+
+        async def note(sid, text, author):
+            notes.append((sid, text, author))
+            return _settlement()
+
+        async def status(sid, st, msg):
+            statuses.append((sid, st))
+            return None
+
+        async def process(sid):
+            processed.append(sid)
+
+        async def get(sid):
+            return _settlement("processed")
+
+        import importlib
+
+        crud = importlib.import_module("..crud", __package__)
+        monkeypatch.setattr(crud, "append_settlement_note", note)
+        monkeypatch.setattr(distribution, "mark_settlement_status", status)
+        monkeypatch.setattr(distribution, "process_settlement", process)
+        monkeypatch.setattr(distribution, "get_settlement", get)
+        return notes, statuses, processed
+
+    def test_closes_at_full_amount(self, monkeypatch):
+        notes, statuses, processed = self._wire(monkeypatch)
+        after = asyncio.run(
+            distribution.settle_cash_owed_settlement(
+                _settlement(), "handed 40 EUR by hand", "op1"
+            )
+        )
+        assert after.status == "processed"
+        assert notes[0][1].startswith(
+            "Settled off-machine (full amount): handed 40 EUR"
+        )
+        assert statuses == [("s1", "pending")]
+        assert processed == ["s1"]
+
+    def test_refuses_other_statuses(self, monkeypatch):
+        self._wire(monkeypatch)
+        with pytest.raises(ValueError):
+            asyncio.run(
+                distribution.settle_cash_owed_settlement(
+                    _settlement("processed"), "x", "op1"
+                )
+            )
+
+
+# ---------------------------------------------------------------------------
+# Models
+# ---------------------------------------------------------------------------
+
+
+class TestSettleOpAndAlertsPubkey:
+    def test_settle_transaction_is_machine_wide_and_carries_txid_note(self):
+        op = CassetteOp(
+            id="st1",
+            machine_id="m1",
+            position=0,
+            op_type="settle_transaction",
+            txid="tx_mv0madw6_wdhtea1v",
+            note="paid by hand",
+            created_at=_NOW,
+        )
+        assert op.to_wire_dict() == {
+            "id": "st1",
+            "at": int(_NOW.timestamp()),
+            "type": "settle_transaction",
+            "txid": "tx_mv0madw6_wdhtea1v",
+            "note": "paid by hand",
+        }
+        CreateCassetteOpData(
+            position=0, op_type="settle_transaction", txid="tx_1", note="n"
+        )
+        with pytest.raises(ValidationError):
+            CreateCassetteOpData(
+                position=0, op_type="settle_transaction"
+            )  # txid required
+        with pytest.raises(ValidationError):
+            CreateCassetteOpData(position=2, op_type="settle_transaction", txid="tx_1")
+        with pytest.raises(ValidationError):
+            CreateCassetteOpData(position=1, op_type="refill", bills=3, txid="tx_1")
+        with pytest.raises(ValidationError):
+            CreateCassetteOpData(position=1, op_type="refill", bills=3, note="no")
+
+    def test_alerts_pubkey_accepts_hex_npub_and_clear(self):
+        assert (
+            UpdateSuperConfigData(alerts_pubkey=_ALERT_PUB.upper()).alerts_pubkey
+            == _ALERT_PUB
+        )
+        from lnbits.utils.nostr import hex_to_npub
+
+        assert (
+            UpdateSuperConfigData(alerts_pubkey=hex_to_npub(_ALERT_PUB)).alerts_pubkey
+            == _ALERT_PUB
+        )
+        assert UpdateSuperConfigData(alerts_pubkey="").alerts_pubkey == ""
+        assert UpdateSuperConfigData().alerts_pubkey is None
+        with pytest.raises(ValidationError):
+            UpdateSuperConfigData(alerts_pubkey="not-a-key")