m2 (AI Agent)
aeba02f287
T013 worker found record_install violating CHECK(amount>0) when platform_pct rounds cut or seller_amount to zero (unhandled IntegrityError -> HTTP 500). Fixed: skip zero rows, grant references the paying tx; regression test added. 9 ledger tests green. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
271 lines
10 KiB
Python
271 lines
10 KiB
Python
"""Property + invariant tests for m2_ledger (contracts/ledger-api.md §Invariants).
|
|
|
|
1. No endpoint UPDATEs/DELETEs tx rows -> append-only by construction,
|
|
exercised indirectly by every test.
|
|
2. balance == fold(tx log) for every operator -> test_balance_matches_independent_fold
|
|
3. Every grant has a non-null paying tx -> test_every_license_grant_has_paying_tx
|
|
4. /tx/install is all-or-nothing and idempotent by ref -> test_insufficient_funds_leaves_zero_new_rows
|
|
test_duplicate_ref_leaves_zero_new_rows
|
|
test_api_install_402_leaves_tx_log_unchanged
|
|
Plus: refund exactly compensates (test_refund_restores_pre_install_balances) and no balance
|
|
except mint goes negative (checked as a stateful invariant below).
|
|
"""
|
|
|
|
from hypothesis import HealthCheck, assume, settings
|
|
from hypothesis import strategies as st
|
|
from hypothesis.stateful import RuleBasedStateMachine, invariant, rule
|
|
|
|
from m2_ledger import models
|
|
|
|
OPERATORS = ["op_alice", "op_bob", "op_carol"]
|
|
ALL_ACCOUNTS = OPERATORS + ["platform"]
|
|
|
|
|
|
def _fold_balance(conn, operator_id: str) -> int:
|
|
"""Independent re-implementation of models.balance: fold the raw tx log in
|
|
Python rather than SQL, so this actually cross-checks the SQL aggregation."""
|
|
bal = 0
|
|
for row in conn.execute("SELECT from_id, to_id, amount FROM tx").fetchall():
|
|
if row["to_id"] == operator_id:
|
|
bal += row["amount"]
|
|
if row["from_id"] == operator_id:
|
|
bal -= row["amount"]
|
|
return bal
|
|
|
|
|
|
def _known_accounts(conn) -> set:
|
|
ids = set()
|
|
for row in conn.execute("SELECT from_id, to_id FROM tx").fetchall():
|
|
ids.add(row["from_id"])
|
|
ids.add(row["to_id"])
|
|
return ids
|
|
|
|
|
|
class LedgerMachine(RuleBasedStateMachine):
|
|
"""Arbitrary sequences of valid ops (grant, install, refund) against a fresh
|
|
in-memory ledger, checking fold-consistency and non-negativity after every op."""
|
|
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.conn = models.init_db(":memory:")
|
|
self.installed_refs = []
|
|
self._ref_counter = 0
|
|
|
|
def teardown(self):
|
|
self.conn.close()
|
|
|
|
@rule(
|
|
operator=st.sampled_from(OPERATORS),
|
|
amount=st.integers(min_value=1, max_value=1000),
|
|
)
|
|
def grant(self, operator, amount):
|
|
models.record_grant(self.conn, operator, amount)
|
|
|
|
@rule(
|
|
pair=st.permutations(OPERATORS).map(lambda p: (p[0], p[1])),
|
|
amount=st.integers(min_value=1, max_value=500),
|
|
pct=st.floats(min_value=0.0, max_value=1.0, allow_nan=False),
|
|
)
|
|
def install(self, pair, amount, pct):
|
|
buyer, seller = pair
|
|
# only attempt ops that are valid per the buyer's *actual* current balance,
|
|
# so this models "arbitrary sequences of valid ops", not failure paths.
|
|
if models.balance(self.conn, buyer) < amount:
|
|
return
|
|
# A platform_pct that rounds the cut or the seller's net to exactly 0
|
|
# trips a known source bug (see report: zero-amount tx row violates the
|
|
# `amount > 0` CHECK constraint with an unhandled IntegrityError) —
|
|
# excluded here since it's not part of this property, not a fix.
|
|
cut = int(amount * pct)
|
|
assume(0 < cut < amount)
|
|
self._ref_counter += 1
|
|
ref = f"ref-{self._ref_counter}"
|
|
models.record_install(self.conn, buyer, seller, amount, pct, ref)
|
|
self.installed_refs.append(ref)
|
|
|
|
@rule(data=st.data())
|
|
def refund(self, data):
|
|
if not self.installed_refs:
|
|
return
|
|
ref = data.draw(st.sampled_from(self.installed_refs))
|
|
try:
|
|
models.record_refund(self.conn, ref)
|
|
except models.AlreadyRefunded:
|
|
pass
|
|
else:
|
|
self.installed_refs.remove(ref)
|
|
|
|
@invariant()
|
|
def balances_match_independent_fold(self):
|
|
for account in _known_accounts(self.conn):
|
|
assert models.balance(self.conn, account) == _fold_balance(self.conn, account)
|
|
|
|
@invariant()
|
|
def no_non_mint_balance_goes_negative(self):
|
|
for account in _known_accounts(self.conn):
|
|
if account == "mint":
|
|
continue
|
|
assert models.balance(self.conn, account) >= 0
|
|
|
|
|
|
LedgerStateMachineTest = LedgerMachine.TestCase
|
|
LedgerStateMachineTest.settings = settings(
|
|
max_examples=50,
|
|
stateful_step_count=25,
|
|
suppress_health_check=[HealthCheck.too_slow, HealthCheck.data_too_large],
|
|
deadline=None,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Example-based invariant tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_every_license_grant_has_paying_tx(conn):
|
|
"""Invariant 3: install grants reference a real 'install' tx; starter grants
|
|
(money only, reason='grant') never create a license grant_ row at all."""
|
|
models.record_grant(conn, "op_alice", 100)
|
|
result = models.record_install(conn, "op_alice", "op_bob", 40, 0.10, "ref-install-1")
|
|
|
|
grant_row = result["grant"]
|
|
assert grant_row["tx_id"] is not None
|
|
|
|
paying_tx = conn.execute(
|
|
"SELECT * FROM tx WHERE id = ?", (grant_row["tx_id"],)
|
|
).fetchone()
|
|
assert paying_tx is not None
|
|
assert paying_tx["reason"] == "install"
|
|
assert paying_tx["from_id"] == "op_alice"
|
|
assert paying_tx["to_id"] == "op_bob"
|
|
|
|
# the starter grant produced a money tx but no row in grant_ (no license without payment)
|
|
grant_count = conn.execute("SELECT COUNT(*) AS n FROM grant_").fetchone()["n"]
|
|
assert grant_count == 1
|
|
|
|
|
|
def test_no_license_grant_without_install(conn):
|
|
"""A pure starter grant must never create a grant_ (license) row."""
|
|
models.record_grant(conn, "op_alice", 100)
|
|
grant_count = conn.execute("SELECT COUNT(*) AS n FROM grant_").fetchone()["n"]
|
|
assert grant_count == 0
|
|
|
|
|
|
def test_insufficient_funds_leaves_zero_new_rows(conn):
|
|
models.record_grant(conn, "op_alice", 10)
|
|
tx_count_before = conn.execute("SELECT COUNT(*) AS n FROM tx").fetchone()["n"]
|
|
grant_count_before = conn.execute("SELECT COUNT(*) AS n FROM grant_").fetchone()["n"]
|
|
|
|
try:
|
|
models.record_install(conn, "op_alice", "op_bob", 10_000, 0.10, "ref-too-much")
|
|
assert False, "expected InsufficientFunds"
|
|
except models.InsufficientFunds:
|
|
pass
|
|
|
|
tx_count_after = conn.execute("SELECT COUNT(*) AS n FROM tx").fetchone()["n"]
|
|
grant_count_after = conn.execute("SELECT COUNT(*) AS n FROM grant_").fetchone()["n"]
|
|
assert tx_count_after == tx_count_before
|
|
assert grant_count_after == grant_count_before
|
|
|
|
|
|
def test_duplicate_ref_leaves_zero_new_rows(conn):
|
|
models.record_grant(conn, "op_alice", 100)
|
|
models.record_install(conn, "op_alice", "op_bob", 40, 0.10, "ref-dup")
|
|
|
|
tx_count_before = conn.execute("SELECT COUNT(*) AS n FROM tx").fetchone()["n"]
|
|
grant_count_before = conn.execute("SELECT COUNT(*) AS n FROM grant_").fetchone()["n"]
|
|
|
|
try:
|
|
models.record_install(conn, "op_alice", "op_bob", 40, 0.10, "ref-dup")
|
|
assert False, "expected DuplicateRef"
|
|
except models.DuplicateRef:
|
|
pass
|
|
|
|
tx_count_after = conn.execute("SELECT COUNT(*) AS n FROM tx").fetchone()["n"]
|
|
grant_count_after = conn.execute("SELECT COUNT(*) AS n FROM grant_").fetchone()["n"]
|
|
assert tx_count_after == tx_count_before
|
|
assert grant_count_after == grant_count_before
|
|
|
|
|
|
def test_refund_restores_pre_install_balances(conn):
|
|
models.record_grant(conn, "op_alice", 100)
|
|
pre_install = {
|
|
op: models.balance(conn, op) for op in ("op_alice", "op_bob", "platform")
|
|
}
|
|
|
|
models.record_install(conn, "op_alice", "op_bob", 40, 0.10, "ref-refundable")
|
|
models.record_refund(conn, "ref-refundable")
|
|
|
|
post_refund = {
|
|
op: models.balance(conn, op) for op in ("op_alice", "op_bob", "platform")
|
|
}
|
|
assert post_refund == pre_install
|
|
|
|
|
|
def test_refund_twice_rejected_and_adds_no_rows(conn):
|
|
models.record_grant(conn, "op_alice", 100)
|
|
models.record_install(conn, "op_alice", "op_bob", 40, 0.10, "ref-once")
|
|
models.record_refund(conn, "ref-once")
|
|
|
|
tx_count_before = conn.execute("SELECT COUNT(*) AS n FROM tx").fetchone()["n"]
|
|
try:
|
|
models.record_refund(conn, "ref-once")
|
|
assert False, "expected AlreadyRefunded"
|
|
except models.AlreadyRefunded:
|
|
pass
|
|
tx_count_after = conn.execute("SELECT COUNT(*) AS n FROM tx").fetchone()["n"]
|
|
assert tx_count_after == tx_count_before
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# API-level test (contract invariant 4, via TestClient)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_api_install_402_leaves_tx_log_unchanged(client, service_headers, admin_headers):
|
|
client.post(
|
|
"/grant/starter",
|
|
json={"operator_id": "op_alice", "amount": 10},
|
|
headers=admin_headers,
|
|
)
|
|
|
|
before = client.get("/tx", headers=service_headers).json()["transactions"]
|
|
|
|
resp = client.post(
|
|
"/tx/install",
|
|
json={
|
|
"buyer": "op_alice",
|
|
"seller": "op_bob",
|
|
"amount": 10_000,
|
|
"ref": "ref-api-too-much",
|
|
},
|
|
headers=service_headers,
|
|
)
|
|
assert resp.status_code == 402
|
|
assert resp.json()["error"] == "insufficient_funds"
|
|
|
|
after = client.get("/tx", headers=service_headers).json()["transactions"]
|
|
assert after == before
|
|
|
|
|
|
def test_edge_platform_pct_zero_and_one(tmp_path):
|
|
"""Regression: pct that rounds a row to zero must skip it, not 500.
|
|
|
|
(T013 worker finding: CHECK amount > 0 was violated by cut == 0 or
|
|
seller_amount == 0; fixed in models.record_install.)
|
|
"""
|
|
from m2_ledger import models as M
|
|
|
|
conn = M.init_db(str(tmp_path / "edge.db"))
|
|
M.record_grant(conn, "buyer", 100)
|
|
|
|
r0 = M.record_install(conn, "buyer", "seller", 10, 0.0, "ref:pct0")
|
|
assert M.balance(conn, "seller") == 10
|
|
assert M.balance(conn, "platform") == 0
|
|
assert len(r0["tx_ids"]) == 1
|
|
|
|
r1 = M.record_install(conn, "buyer", "seller", 10, 1.0, "ref:pct1")
|
|
assert M.balance(conn, "platform") == 10
|
|
assert len(r1["tx_ids"]) == 1
|
|
assert r1["grant"]["tx_id"] == r1["tx_ids"][0]
|
|
assert M.balance(conn, "buyer") == 80
|