aiolabs/spirekeeper
8
0
Fork 0
Code Issues 23 Pull requests Projects Releases Packages Wiki Activity Actions
spirekeeper/tests/test_fee_publish_triggers.py
Padreug 794d7e5395 feat(v2): wire fee-config publish into machine + super-config triggers (#39 3/3) 
Three trigger points wire fee_transport.publish_fee_config into the
satmachineadmin API endpoints per the #39 spec. All three soft-fail on
transport errors — the underlying CRUD operation (machine create /
update / super-config save) succeeds even when the publish couldn't
reach the relay or the signer, and the operator can re-trigger by
editing again.

views_api.py:
- api_create_machine — publishes always after create, even when
  operator fees default to 0/0 (the resulting super-only payload is
  what unblocks the ATM past its `awaiting-fees` maintenance gate).
  Reads super_config singleton; if absent (m001 should have inserted
  it, so this is an impossible state), skips the publish to avoid
  crashing create.
- api_update_machine — publishes only when either
  operator_cash_*_fee_fraction is in the patch payload. Skip on
  name/location/wallet_id/is_active/fiat_code edits since those don't
  affect the fee model the ATM enforces (avoids unnecessary relay
  churn).
- api_update_super_config — publishes to every active machine when
  either super fraction changes. Per-machine: that machine's
  operator_user_id is the signer (machines owned by different
  operators sign with different keys); each soft-fail is independent.
  Skip if only super_fee_wallet_id changed (no fee-model impact).

Tests (9 cases, all green):
- 3 create-machine triggers: default 0/0 operator fees still publishes
  super-only payload, nonzero operator fees publish full payload,
  None super_config short-circuits without crashing
- 4 update-machine triggers: publishes on cash_in change, publishes on
  cash_out change, skips on name-only, skips on is_active-only
- 2 super-config triggers: publishes per-active-machine signed by
  each machine's operator on fraction change, skips entirely on
  wallet-id-only change (with an assertion that list_all_active_machines
  is never called, proving the short-circuit path)

191/191 tests green. Layer 2 (#39) complete; ready for joint smoke
once bitspire fixes the three deploy gaps from coord-log §2026-06-01T18:30Z
(`relay.aiolabs.dev` default, `VITE_LNBITS_HTTP_URL` dead echo,
operator-fees subscriber not running in maintenance state).

Refs: aiolabs/satmachineadmin#37 (parent), #39 (closes Layer 2),
aiolabs/lamassu-next#57 (Layer 3 consumer — blocked on bitspire-side
gaps).

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-06-01 20:07:56 +02:00

391 lines
14 KiB
Python
Raw Permalink Blame History

"""
Tests for the three views_api trigger points that publish fee config
to ATMs via fee_transport (aiolabs/satmachineadmin#39 Layer 2):
1. api_create_machine — publish always after create (so ATM unblocks
past `awaiting-fees` maintenance, even with default 0/0 operator
fees that produce a super-only payload)
2. api_update_machine — publish only when either operator fee fraction
changes (skip on name/location/wallet_id/is_active-only edits)
3. api_update_super_config — publish to every active machine when
either super fraction changes, signed by each machine's operator
Tests monkeypatch `views_api.publish_fee_config` with a recording stub
to verify the trigger fired (or not) and what arguments it received.
The publisher itself is exercised by test_fee_transport.py — these
tests are about the wiring.
"""
import asyncio
from datetime import datetime
from .. import views_api
from ..models import CreateMachineData, Machine, SuperConfig, UpdateMachineData
_NOW = datetime(2026, 6, 1, 12, 0, 0)
_ATM_PUBKEY_HEX = (
"522a4538f1df96508d9ee8b14072344dd4a566acfe03c25a92a39179c6fca891"
)
_OP_USER_ID = "ac35c9fc842f40f0a0e9809347cd24d1"
def _machine(
machine_id: str = "m1",
npub: str = _ATM_PUBKEY_HEX,
op_in: float = 0.0,
op_out: float = 0.0,
operator_user_id: str = _OP_USER_ID,
) -> Machine:
return Machine(
id=machine_id,
operator_user_id=operator_user_id,
machine_npub=npub,
wallet_id="w1",
name=f"machine-{machine_id}",
location=None,
fiat_code="EUR",
is_active=True,
operator_cash_in_fee_fraction=op_in,
operator_cash_out_fee_fraction=op_out,
created_at=_NOW,
updated_at=_NOW,
)
def _super(in_frac: float = 0.03, out_frac: float = 0.03) -> SuperConfig:
return SuperConfig(
id="default",
super_cash_in_fee_fraction=in_frac,
super_cash_out_fee_fraction=out_frac,
super_fee_wallet_id="super-wallet",
updated_at=_NOW,
)
class _PublishRecorder:
"""Records every (machine.id, super_in, super_out, operator) tuple
publish_fee_config was called with. Drop-in stub for monkeypatching
`views_api.publish_fee_config`."""
def __init__(self):
self.calls: list[tuple[str, float, float, float, float, str]] = []
async def __call__(self, machine, super_config, operator_user_id):
self.calls.append(
(
machine.id,
float(super_config.super_cash_in_fee_fraction),
float(super_config.super_cash_out_fee_fraction),
float(machine.operator_cash_in_fee_fraction),
float(machine.operator_cash_out_fee_fraction),
operator_user_id,
)
)
return {"id": f"evt_{machine.id}", "kind": 30078}
# ---------------------------------------------------------------------------
# Trigger 1: api_create_machine
# ---------------------------------------------------------------------------
class TestCreateMachineTrigger:
def test_publishes_on_create_with_default_operator_fees(self, monkeypatch):
"""Default 0/0 operator fees — payload carries super-only totals.
Publish fires anyway so the ATM gets initial config and can
boot past maintenance."""
recorder = _PublishRecorder()
machine = _machine(op_in=0.0, op_out=0.0)
async def fake_assert_wallet(*args, **kwargs):
return None
async def fake_assert_collision(*args, **kwargs):
return None
async def fake_assert_fee_cap(*args, **kwargs):
return None
async def fake_create_machine(user_id, data):
return machine
async def fake_get_super():
return _super()
monkeypatch.setattr(views_api, "_assert_wallet_owned_by", fake_assert_wallet)
monkeypatch.setattr(views_api, "_assert_no_pubkey_collision", fake_assert_collision)
monkeypatch.setattr(views_api, "_assert_machine_fee_cap_safe", fake_assert_fee_cap)
monkeypatch.setattr(views_api, "create_machine", fake_create_machine)
monkeypatch.setattr(views_api, "get_super_config", fake_get_super)
monkeypatch.setattr(views_api, "publish_fee_config", recorder)
# Build a CreateMachineData + fake User and invoke the endpoint.
from types import SimpleNamespace
data = CreateMachineData(
machine_npub=_ATM_PUBKEY_HEX,
wallet_id="w1",
name="sintra",
)
user = SimpleNamespace(id=_OP_USER_ID)
result = asyncio.run(views_api.api_create_machine(data=data, user=user))
assert result is machine
assert len(recorder.calls) == 1
assert recorder.calls[0] == ("m1", 0.03, 0.03, 0.0, 0.0, _OP_USER_ID)
def test_publishes_on_create_with_nonzero_operator_fees(self, monkeypatch):
recorder = _PublishRecorder()
machine = _machine(op_in=0.05, op_out=0.05)
async def passthrough(*args, **kwargs):
return None
async def fake_create_machine(user_id, data):
return machine
async def fake_get_super():
return _super(0.03, 0.03)
monkeypatch.setattr(views_api, "_assert_wallet_owned_by", passthrough)
monkeypatch.setattr(views_api, "_assert_no_pubkey_collision", passthrough)
monkeypatch.setattr(views_api, "_assert_machine_fee_cap_safe", passthrough)
monkeypatch.setattr(views_api, "create_machine", fake_create_machine)
monkeypatch.setattr(views_api, "get_super_config", fake_get_super)
monkeypatch.setattr(views_api, "publish_fee_config", recorder)
from types import SimpleNamespace
data = CreateMachineData(
machine_npub=_ATM_PUBKEY_HEX,
wallet_id="w1",
operator_cash_in_fee_fraction=0.05,
operator_cash_out_fee_fraction=0.05,
)
user = SimpleNamespace(id=_OP_USER_ID)
asyncio.run(views_api.api_create_machine(data=data, user=user))
assert recorder.calls == [("m1", 0.03, 0.03, 0.05, 0.05, _OP_USER_ID)]
def test_no_super_config_skips_publish(self, monkeypatch):
"""If the super-config singleton is missing (impossible in
practice since m001 inserts it), skip the publish rather than
crash the create. Machine still created."""
recorder = _PublishRecorder()
machine = _machine()
async def passthrough(*args, **kwargs):
return None
async def fake_create_machine(user_id, data):
return machine
async def fake_get_super():
return None
monkeypatch.setattr(views_api, "_assert_wallet_owned_by", passthrough)
monkeypatch.setattr(views_api, "_assert_no_pubkey_collision", passthrough)
monkeypatch.setattr(views_api, "_assert_machine_fee_cap_safe", passthrough)
monkeypatch.setattr(views_api, "create_machine", fake_create_machine)
monkeypatch.setattr(views_api, "get_super_config", fake_get_super)
monkeypatch.setattr(views_api, "publish_fee_config", recorder)
from types import SimpleNamespace
data = CreateMachineData(machine_npub=_ATM_PUBKEY_HEX, wallet_id="w1")
user = SimpleNamespace(id=_OP_USER_ID)
result = asyncio.run(views_api.api_create_machine(data=data, user=user))
assert result is machine
assert recorder.calls == []
# ---------------------------------------------------------------------------
# Trigger 2: api_update_machine
# ---------------------------------------------------------------------------
def _wire_update_machine_patches(
monkeypatch, existing_machine, updated_machine, recorder
):
"""Common setup for api_update_machine tests."""
async def passthrough(*args, **kwargs):
return None
async def fake_get_machine(machine_id):
return existing_machine
async def fake_update_machine(machine_id, data):
return updated_machine
async def fake_get_super():
return _super(0.03, 0.03)
monkeypatch.setattr(views_api, "_assert_wallet_owned_by", passthrough)
monkeypatch.setattr(views_api, "_assert_machine_fee_cap_safe", passthrough)
monkeypatch.setattr(views_api, "get_machine", fake_get_machine)
monkeypatch.setattr(views_api, "update_machine", fake_update_machine)
monkeypatch.setattr(views_api, "get_super_config", fake_get_super)
monkeypatch.setattr(views_api, "publish_fee_config", recorder)
class TestUpdateMachineTrigger:
def test_publishes_when_operator_cash_in_changes(self, monkeypatch):
recorder = _PublishRecorder()
existing = _machine(op_in=0.05, op_out=0.05)
updated = _machine(op_in=0.07, op_out=0.05)
_wire_update_machine_patches(monkeypatch, existing, updated, recorder)
from types import SimpleNamespace
data = UpdateMachineData(operator_cash_in_fee_fraction=0.07)
user = SimpleNamespace(id=_OP_USER_ID)
asyncio.run(
views_api.api_update_machine(machine_id="m1", data=data, user=user)
)
assert len(recorder.calls) == 1
assert recorder.calls[0] == ("m1", 0.03, 0.03, 0.07, 0.05, _OP_USER_ID)
def test_publishes_when_operator_cash_out_changes(self, monkeypatch):
recorder = _PublishRecorder()
existing = _machine(op_in=0.05, op_out=0.05)
updated = _machine(op_in=0.05, op_out=0.08)
_wire_update_machine_patches(monkeypatch, existing, updated, recorder)
from types import SimpleNamespace
data = UpdateMachineData(operator_cash_out_fee_fraction=0.08)
user = SimpleNamespace(id=_OP_USER_ID)
asyncio.run(
views_api.api_update_machine(machine_id="m1", data=data, user=user)
)
assert len(recorder.calls) == 1
assert recorder.calls[0] == ("m1", 0.03, 0.03, 0.05, 0.08, _OP_USER_ID)
def test_no_publish_when_only_name_changes(self, monkeypatch):
"""Name / location / fiat_code / is_active / wallet_id changes
don't affect the fee model the ATM enforces — skip the
republish to avoid relay churn."""
recorder = _PublishRecorder()
existing = _machine()
updated = _machine() # same fees
_wire_update_machine_patches(monkeypatch, existing, updated, recorder)
from types import SimpleNamespace
data = UpdateMachineData(name="new name")
user = SimpleNamespace(id=_OP_USER_ID)
asyncio.run(
views_api.api_update_machine(machine_id="m1", data=data, user=user)
)
assert recorder.calls == []
def test_no_publish_when_only_is_active_changes(self, monkeypatch):
recorder = _PublishRecorder()
_wire_update_machine_patches(monkeypatch, _machine(), _machine(), recorder)
from types import SimpleNamespace
data = UpdateMachineData(is_active=False)
user = SimpleNamespace(id=_OP_USER_ID)
asyncio.run(
views_api.api_update_machine(machine_id="m1", data=data, user=user)
)
assert recorder.calls == []
# ---------------------------------------------------------------------------
# Trigger 3: api_update_super_config
# ---------------------------------------------------------------------------
class TestSuperConfigUpdateTrigger:
def test_publishes_to_every_active_machine_on_super_fraction_change(
self, monkeypatch
):
"""A super-fee change ripples to every active machine since each
machine's total = super + machine.operator. Republish per-machine
with that machine's operator as the signer (machines owned by
different operators sign with different keys)."""
recorder = _PublishRecorder()
new_super = _super(in_frac=0.04, out_frac=0.04)
machines = [
_machine(machine_id="m1", operator_user_id="op_A"),
_machine(machine_id="m2", operator_user_id="op_B", op_in=0.05, op_out=0.07),
_machine(machine_id="m3", operator_user_id="op_A", op_in=0.02, op_out=0.02),
]
async def fake_assert_cap(*args, **kwargs):
return None
async def fake_update_super(data):
return new_super
async def fake_list_active():
return machines
monkeypatch.setattr(
views_api, "_assert_super_config_cap_safe", fake_assert_cap
)
monkeypatch.setattr(views_api, "update_super_config", fake_update_super)
monkeypatch.setattr(
views_api, "list_all_active_machines", fake_list_active
)
monkeypatch.setattr(views_api, "publish_fee_config", recorder)
from types import SimpleNamespace
from ..models import UpdateSuperConfigData
data = UpdateSuperConfigData(super_cash_in_fee_fraction=0.04)
user = SimpleNamespace(id="super_admin")
asyncio.run(views_api.api_update_super_config(data=data, _user=user))
assert len(recorder.calls) == 3
# Verify each call carries the NEW super fractions + that
# machine's operator + own fees
assert recorder.calls[0] == ("m1", 0.04, 0.04, 0.0, 0.0, "op_A")
assert recorder.calls[1] == ("m2", 0.04, 0.04, 0.05, 0.07, "op_B")
assert recorder.calls[2] == ("m3", 0.04, 0.04, 0.02, 0.02, "op_A")
def test_no_publish_when_only_wallet_id_changes(self, monkeypatch):
"""Changing super_fee_wallet_id without touching either fraction
doesn't affect any ATM's fee model — skip the fleet-wide
republish."""
recorder = _PublishRecorder()
new_super = _super(in_frac=0.03, out_frac=0.03)
async def fake_assert_cap(*args, **kwargs):
return None
async def fake_update_super(data):
return new_super
async def fake_list_active():
raise AssertionError(
"list_all_active_machines should not be called when "
"no fraction changed"
)
monkeypatch.setattr(
views_api, "_assert_super_config_cap_safe", fake_assert_cap
)
monkeypatch.setattr(views_api, "update_super_config", fake_update_super)
monkeypatch.setattr(
views_api, "list_all_active_machines", fake_list_active
)
monkeypatch.setattr(views_api, "publish_fee_config", recorder)
from types import SimpleNamespace
from ..models import UpdateSuperConfigData
data = UpdateSuperConfigData(super_fee_wallet_id="new-wallet")
user = SimpleNamespace(id="super_admin")
asyncio.run(views_api.api_update_super_config(data=data, _user=user))
assert recorder.calls == []
Reference in a new issue View git blame Copy permalink