git / moonbite-network / tests/test_merchants.py · raw
"""Tests for the non-custodial merchant layer — payment observation.Focus on the automation that lets a shop auto-reconcile: list_invoices andpoll_pending_invoices. A payment is a NEW inbound total over the baselinesnapshotted when the invoice was raised; the server only observes it. The chainis a tiny in-memory ``received_lookup`` we can advance by hand.Also covers the 2026-10-09 bug-bounty fixes: * per-invoice payment attribution (one payment cannot settle two invoices sharing an address), and * merchant-scoped, token-authenticated ledger reads (no anonymous dump)."""import importlibimport pytestimport merchants@pytest.fixture()def fresh_db(tmp_path): merchants._conn = None merchants._DB_PATH = tmp_path / "merchants.db" yield if merchants._conn is not None: merchants._conn.close() merchants._conn = None importlib.reload(merchants)class FakeChain: """A mutable per-address running total of received base units.""" def __init__(self): self._totals = {} def pay(self, address, coins): # coins -> base units (100 units == 1 coin, per merchants.UNITS_PER_COIN) self._totals[address] = self._totals.get(address, 0) + int( coins * merchants.UNITS_PER_COIN ) def lookup(self, address): return self._totals.get(address, 0)ADDR_A = "moon1cafe0000000000000000000000000000"ADDR_B = "moon1shop0000000000000000000000000000"def _merchant(addr=ADDR_A, name="Cafe", category="food"): """Register a merchant and return (merchant_id, manage_token).""" m = merchants.add_merchant(name, category, addr, address_validator=None) return m["id"], m["manage_token"]def _invoice(addr, amount, lookup, mid, token, **kw): """Raise an invoice as an authenticated merchant (audit H1: no anonymous rows).""" return merchants.create_invoice( addr, amount, lookup, merchant_id=mid, manage_token=token, **kw )def test_invoice_starts_pending_and_flips_paid_on_payment(fresh_db): chain = FakeChain() mid, token = _merchant(ADDR_A) inv = _invoice(ADDR_A, "2.5", chain.lookup, mid, token, memo="latte") assert inv["status"] == "pending" assert inv["paid_units"] == 0 chain.pay(ADDR_A, 3) # overpay is fine (>= amount) updated = merchants.invoice_status(inv["id"], chain.lookup) assert updated["status"] == "paid" assert updated["paid_at"] is not Nonedef test_baseline_ignores_preexisting_funds(fresh_db): chain = FakeChain() chain.pay(ADDR_A, 100) # merchant already had prior receipts mid, token = _merchant(ADDR_A) inv = _invoice(ADDR_A, "5", chain.lookup, mid, token) # The pre-existing balance must NOT satisfy the new invoice. assert merchants.invoice_status(inv["id"], chain.lookup)["status"] == "pending" chain.pay(ADDR_A, 5) # a genuinely new payment assert merchants.invoice_status(inv["id"], chain.lookup)["status"] == "paid"def test_one_payment_cannot_settle_two_invoices_on_same_address(fresh_db): """Regression for the double-settlement bug: a single payment backs ONE invoice, not every pending invoice sharing the merchant's address.""" chain = FakeChain() mid, token = _merchant(ADDR_A) a = _invoice(ADDR_A, "5", chain.lookup, mid, token) b = _invoice(ADDR_A, "5", chain.lookup, mid, token) chain.pay(ADDR_A, 5) # exactly one invoice's worth # Settling the whole address group is order-independent: querying either # invoice yields the same allocation. Exactly one is paid, one still pending. sa = merchants.invoice_status(a["id"], chain.lookup) sb = merchants.invoice_status(b["id"], chain.lookup) statuses = sorted([sa["status"], sb["status"]]) assert statuses == ["paid", "pending"], statuses # A second payment covers the remaining invoice; now both are paid. chain.pay(ADDR_A, 5) sa2 = merchants.invoice_status(a["id"], chain.lookup) sb2 = merchants.invoice_status(b["id"], chain.lookup) assert sa2["status"] == "paid" and sb2["status"] == "paid"def test_polling_the_second_invoice_does_not_steal_the_first_payment(fresh_db): """Even if the newer invoice is polled first, the payment is allocated to the older invoice (oldest-first), so the pool is never double-counted.""" chain = FakeChain() mid, token = _merchant(ADDR_A) older = _invoice(ADDR_A, "5", chain.lookup, mid, token) newer = _invoice(ADDR_A, "5", chain.lookup, mid, token) chain.pay(ADDR_A, 5) # Poll the NEWER one first. merchants.invoice_status(newer["id"], chain.lookup) # Still only one invoice is paid in total. paid = [ i for i in (older["id"], newer["id"]) if merchants.invoice_status(i, chain.lookup)["status"] == "paid" ] assert len(paid) == 1def test_poll_pending_invoices_sweeps_and_marks_paid(fresh_db): chain = FakeChain() mid, token = _merchant(ADDR_A) mid_b, token_b = _merchant(ADDR_B, name="Shop B") a = _invoice(ADDR_A, "1", chain.lookup, mid, token) b = _invoice(ADDR_B, "2", chain.lookup, mid_b, token_b) c = _invoice(ADDR_A, "10", chain.lookup, mid, token) # unpaid chain.pay(ADDR_A, 1) # satisfies a (amount 1) but not c (amount 10) chain.pay(ADDR_B, 2) # satisfies b summary = merchants.poll_pending_invoices(chain.lookup, merchant_id=mid, manage_token=token) assert summary["checked"] == 2 assert set(summary["paid"]) == {a["id"]} assert summary["expired"] == [] # Distinct addresses hit once each, not once per invoice. assert summary["addresses_scanned"] == 1 summary_b = merchants.poll_pending_invoices( chain.lookup, merchant_id=mid_b, manage_token=token_b ) assert summary_b["paid"] == [b["id"]] assert merchants.invoice_status(c["id"], chain.lookup)["status"] == "pending"def test_expiry_when_ttl_lapses(fresh_db): chain = FakeChain() mid, token = _merchant(ADDR_A) # A lapsed TTL flips the invoice to 'expired' on the next observation. inv = _invoice(ADDR_A, "1", chain.lookup, mid, token, ttl=-10) assert merchants.invoice_status(inv["id"], chain.lookup)["status"] == "expired" # And an unpaid invoice that lapses between sweeps is caught by the sweep. inv2 = _invoice(ADDR_A, "1", chain.lookup, mid, token, ttl=3600) with merchants._lock: merchants._connect().execute( "UPDATE invoices SET expires_at = ? WHERE id = ?", (0, inv2["id"]), ) merchants._connect().commit() summary = merchants.poll_pending_invoices(chain.lookup, merchant_id=mid, manage_token=token) assert inv2["id"] in summary["expired"]def test_poll_is_idempotent(fresh_db): chain = FakeChain() mid, token = _merchant(ADDR_A) inv = _invoice(ADDR_A, "1", chain.lookup, mid, token) chain.pay(ADDR_A, 1) first = merchants.poll_pending_invoices(chain.lookup, merchant_id=mid, manage_token=token) assert inv["id"] in first["paid"] # Second sweep: nothing pending remains, so it neither re-pays nor errors. second = merchants.poll_pending_invoices(chain.lookup, merchant_id=mid, manage_token=token) assert second["checked"] == 0 and second["paid"] == []def test_list_invoices_scopes_by_merchant_and_checks_chain(fresh_db): chain = FakeChain() mid, token = _merchant(ADDR_A) inv = _invoice(ADDR_A, "1", chain.lookup, mid, token) mid_b, token_b = _merchant(ADDR_B, name="Shop B") _invoice(ADDR_B, "1", chain.lookup, mid_b, token_b) # a different merchant scoped = merchants.list_invoices(chain.lookup, merchant_id=mid, manage_token=token) assert [r["id"] for r in scoped] == [inv["id"]] chain.pay(ADDR_A, 1) scoped2 = merchants.list_invoices(chain.lookup, merchant_id=mid, manage_token=token) assert scoped2[0]["status"] == "paid"def test_list_and_poll_require_a_valid_token(fresh_db): chain = FakeChain() mid, token = _merchant(ADDR_A) _invoice(ADDR_A, "1", chain.lookup, mid, token) # No token -> refused. with pytest.raises(merchants.MerchantAuthError): merchants.list_invoices(chain.lookup, merchant_id=mid) # Wrong token -> refused. with pytest.raises(merchants.MerchantAuthError): merchants.list_invoices(chain.lookup, merchant_id=mid, manage_token="deadbeef") # Unscoped (the old full-ledger dump) -> refused. with pytest.raises(merchants.MerchantAuthError): merchants.list_invoices(chain.lookup, merchant_id=None, manage_token=token) with pytest.raises(merchants.MerchantAuthError): merchants.poll_pending_invoices(chain.lookup, merchant_id=mid, manage_token="nope") # Correct token -> allowed. assert merchants.list_invoices(chain.lookup, merchant_id=mid, manage_token=token) is not Nonedef test_manage_token_is_returned_once_and_never_leaked(fresh_db): created = merchants.add_merchant("Shop", "goods", ADDR_A, address_validator=None) assert created.get("manage_token") # surfaced exactly once, on creation # Public directory reads must never echo the token. assert "manage_token" not in merchants.get_merchant(created["id"]) for row in merchants.list_merchants(): assert "manage_token" not in row# --------------------------------------------------------------------------- ## Audit H1 (2026-10-09): invoice creation is token-gated and address-bound, and# the per-address allocation group is scoped to one merchant.# --------------------------------------------------------------------------- #def test_anonymous_invoice_is_refused(fresh_db): chain = FakeChain() _merchant(ADDR_A) with pytest.raises(merchants.MerchantAuthError): merchants.create_invoice(ADDR_A, "1", chain.lookup) with pytest.raises(merchants.MerchantAuthError): merchants.create_invoice(ADDR_A, "1", chain.lookup, merchant_id=None, manage_token="x")def test_invoice_requires_matching_token(fresh_db): chain = FakeChain() mid, token = _merchant(ADDR_A) with pytest.raises(merchants.MerchantAuthError): merchants.create_invoice(ADDR_A, "1", chain.lookup, merchant_id=mid) with pytest.raises(merchants.MerchantAuthError): merchants.create_invoice(ADDR_A, "1", chain.lookup, merchant_id=mid, manage_token="bad") # Another merchant's valid token does not authorise this merchant id. mid_b, token_b = _merchant(ADDR_B, name="Shop B") with pytest.raises(merchants.MerchantAuthError): merchants.create_invoice(ADDR_A, "1", chain.lookup, merchant_id=mid, manage_token=token_b)def test_invoice_must_target_the_merchants_own_address(fresh_db): chain = FakeChain() mid, token = _merchant(ADDR_A) # A valid token cannot raise an invoice against someone else's address. with pytest.raises(ValueError): _invoice(ADDR_B, "1", chain.lookup, mid, token) # Omitting the address defaults to the registered one. inv = _invoice(None, "1", chain.lookup, mid, token) assert inv["address"] == ADDR_A inv2 = _invoice("", "1", chain.lookup, mid, token) assert inv2["address"] == ADDR_Adef test_dust_row_from_another_merchant_cannot_starve_a_real_invoice(fresh_db): """Regression for H1: a row on the same address but a different merchant id (or a legacy anonymous row) is NOT part of this merchant's allocation group, so it can never sit ahead of the real invoice in the oldest-first walk.""" chain = FakeChain() mid, token = _merchant(ADDR_A, name="Real Shop") # Simulate legacy/foreign rows inserted directly (the API no longer allows it). now = int(__import__("time").time()) with merchants._lock: conn = merchants._connect() for i, foreign_mid in enumerate(("someone-else", None)): conn.execute( """INSERT INTO invoices (id, merchant_id, address, amount, amount_units, memo, status, baseline_units, created_at, expires_at, paid_at) VALUES (?, ?, ?, '0.01', 1, '', 'pending', 0, ?, ?, NULL)""", (f"dust{i}", foreign_mid, ADDR_A, now - 1000, now + 100000), ) conn.commit() real = _invoice(ADDR_A, "5", chain.lookup, mid, token) chain.pay(ADDR_A, 5) # exactly the real invoice's amount settled = merchants.invoice_status(real["id"], chain.lookup) assert settled["status"] == "paid" # Foreign rows were untouched by this merchant's settlement. with merchants._lock: rows = merchants._connect().execute( "SELECT id, status FROM invoices WHERE id LIKE 'dust%'" ).fetchall() assert all(r["status"] == "pending" for r in rows)def test_same_merchant_double_credit_guard_still_holds(fresh_db): """The c376b36 cursor logic is unchanged inside a merchant's own group.""" chain = FakeChain() mid, token = _merchant(ADDR_A) a = _invoice(ADDR_A, "3", chain.lookup, mid, token) b = _invoice(ADDR_A, "3", chain.lookup, mid, token) chain.pay(ADDR_A, 3) sa = merchants.invoice_status(a["id"], chain.lookup) sb = merchants.invoice_status(b["id"], chain.lookup) assert sorted([sa["status"], sb["status"]]) == ["paid", "pending"]