Padreug
1e235e3e42
Some checks failed
ci.yml / test: slice-2 coverage — NIP-17 round trip, alert policy, settle path, op shape (pull_request) Failing after 0s
424 lines
14 KiB
Python
424 lines
14 KiB
Python
"""
|
|
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")
|