diff --git a/coffer/ledger.py b/coffer/ledger.py index 17bd0e3..a135eb4 100644 --- a/coffer/ledger.py +++ b/coffer/ledger.py @@ -231,8 +231,12 @@ class Ledger: Investments and contract payments draw funded credit only (granted credit is for services); they raise InsufficientCredit if funded is short. """ - if funded_only is None: - funded_only = tx_kind in FUNDED_ONLY_TX_KINDS + if tx_kind in FUNDED_ONLY_TX_KINDS: + if funded_only is False: + raise TransferNotAllowed(f"{tx_kind} must draw funded credit only") + funded_only = True + elif funded_only is None: + funded_only = False _positive_int(amount_subunits, "transfer amount") if not reason: raise ValueError("a transfer needs a reason") @@ -265,11 +269,17 @@ class Ledger: Money moves back from the original credit account (its funded lots) to the original debit account, restoring each draw as its original kind. + + Operator correction tool only. It has no authorization of its own: an + admin/operator gate is REQUIRED before exposing it through any API, + CLI command or agent-facing route. """ if not reason: raise ValueError("a reversal needs a reason") if not tx.debit_account or not tx.credit_account: raise ValueError("only transfers between two accounts can be reversed") + if tx.kind.endswith("_reversal") or "reverses_seq" in tx.meta: + raise ValueError("a reversal cannot itself be reversed") if any(t.meta.get("reverses_seq") == tx.seq for t in self.transactions): raise ValueError(f"seq {tx.seq} is already reversed") draws = tx.meta.get("draws") or [] diff --git a/coffer/product.py b/coffer/product.py index de96830..3f9270d 100644 --- a/coffer/product.py +++ b/coffer/product.py @@ -350,6 +350,35 @@ class CofferProduct: "payee": self.balance(payee), } + def reverse_transaction(self, seq: int, reason: str) -> dict[str, Any]: + """Reverse ledger tx `seq` by its original draw kinds, saved atomically. + + The in-memory reversal and the SQLite writes (lot rows rewritten, + reversal txs inserted) commit together or not at all; on any failure + the DB is rolled back and the ledger reloaded from it. + Operator-only: requires an admin gate before any API/CLI exposure. + """ + tx = next((t for t in self.ledger.history() if t.seq == seq), None) + if tx is None: + raise ProductError(f"no transaction with seq {seq}") + try: + revs = self.ledger.reverse_tx(tx, reason) + for aid in {tx.debit_account, tx.credit_account}: + self._rewrite_lots(aid) + for r in revs: + self._persist_tx(r) + self.db.commit() + except Exception: + self.db._conn.rollback() + self._load() + raise + self._after_mutation() + return { + "reversals": [self.ledger.describe_tx(r) for r in revs], + tx.debit_account: self.balance(tx.debit_account), + tx.credit_account: self.balance(tx.credit_account), + } + def grant_signup( self, agent_id: str, diff --git a/tests/test_funded_only.py b/tests/test_funded_only.py index 2c74b5f..faf56bd 100644 --- a/tests/test_funded_only.py +++ b/tests/test_funded_only.py @@ -46,7 +46,11 @@ class FundedOnlyTest(unittest.TestCase): def test_reversal_restores_original_draw_kinds(self): lg = ledger(40, 60) - bad = lg.transfer(L, O, to_subunits(100), "inv", tx_kind="investment", funded_only=False) + from unittest import mock + # Simulate the pre-fix code path that produced ledger seq 39 (promo drawn). + with mock.patch("coffer.ledger.FUNDED_ONLY_TX_KINDS", frozenset()): + bad = lg.transfer(L, O, to_subunits(100), "inv", tx_kind="investment") + self.assertEqual(bad.meta["draws"][0]["kind"], "promotional") revs = lg.reverse_tx(bad, "correction:inv") self.assertEqual([(t.kind, t.amount, t.meta["credit_kind"]) for t in revs], [("investment_reversal", 4000, "promotional"), @@ -62,3 +66,60 @@ class FundedOnlyTest(unittest.TestCase): if __name__ == "__main__": unittest.main() + + +class FundedOnlyGuardsTest(unittest.TestCase): + def test_cannot_opt_out_of_funded_only(self): + from coffer.ledger import TransferNotAllowed + for kind in ("investment", "contract_payment"): + lg = ledger(40, 200) + with self.assertRaises(TransferNotAllowed): + lg.transfer(L, O, to_subunits(100), "x", tx_kind=kind, funded_only=False) + self.assertEqual(lg.account(L).balance_subunits()["promotional"], 4000) + + def test_reversal_of_reversal_refused(self): + lg = ledger(0, 100) + tx = lg.transfer(L, O, to_subunits(100), "inv", tx_kind="investment") + rev = lg.reverse_tx(tx, "fix")[0] + with self.assertRaises(ValueError): + lg.reverse_tx(rev, "fix again") + + +class ProductReverseAtomicTest(unittest.TestCase): + def setUp(self): + import tempfile, os + from coffer.product import CofferProduct + self.d = tempfile.TemporaryDirectory() + os.environ["COFFER_AUTO_PUBLISH_NEWS"] = "0" + self.path = f"{self.d.name}/c.db" + self.p = CofferProduct(self.path) + self.p.ledger.credit_subunits(L, 10000, CreditKind.FUNDED, "top-up", log=False) + tx = self.p.ledger.transfer(L, O, 10000, "inv", tx_kind="investment") + self.p._commit_accounts(L, O, txs=[tx]) + self.seq = tx.seq + + def tearDown(self): + self.p.close(); self.d.cleanup() + + def test_reverse_persists(self): + from coffer.product import CofferProduct + self.p.reverse_transaction(self.seq, "fix") + q = CofferProduct(self.path) + self.assertEqual(q.balance(L)["funded_centicredits"], 10000) + self.assertEqual(q.balance(O)["spendable_centicredits"], 0) + self.assertEqual(q.ledger.history()[-1].meta["reverses_seq"], self.seq) + q.close() + + def test_failed_save_rolls_back(self): + from unittest import mock + from coffer.product import CofferProduct + n = len(self.p.ledger.history()) + with mock.patch.object(self.p, "_persist_tx", side_effect=RuntimeError("disk")): + with self.assertRaises(RuntimeError): + self.p.reverse_transaction(self.seq, "fix") + self.assertEqual(len(self.p.ledger.history()), n) + self.assertEqual(self.p.balance(O)["spendable_centicredits"], 10000) + q = CofferProduct(self.path) + self.assertEqual(len(q.ledger.history()), n) + self.assertEqual(q.balance(O)["spendable_centicredits"], 10000) + q.close()