import tempfile import unittest from pathlib import Path from coffer.ledger import InsufficientCredit, TransferNotAllowed from coffer.store import NotAReviewer, OrderNotPending, TooManyPendingOrders from coffer.product import CofferProduct, ProductError from coffer.public_ledger import tx_public from coffer.units import human_account class PeopleTests(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.db = Path(self.tmp.name) / "coffer.db" self.p = CofferProduct(self.db) self.p.bootstrap_opening_books() self.p.record_build_contract() self.p.signup_flow("atlas") # an agent: 40 granted, 20 seat charged def tearDown(self): self.p.close() self.tmp.cleanup() def test_human_signup_grant_once(self): r = self.p.human_signup("sam", 101) self.assertEqual(r["account_id"], "human:sam") self.assertEqual(r["balance"]["promotional"], 40.0) self.assertTrue(self.p.human_signup("sam", 101)["already"]) self.assertEqual(self.p.balance("human:sam")["spendable"], 40.0) with self.assertRaises(ValueError): human_account("Sam Smith ") def test_grant_keyed_on_gitea_id_rename_and_concurrency(self): self.p.human_signup("sam", 101) with self.assertRaises(ProductError): # rename: the Gitea id already had its grant self.p.human_signup("sam-renamed", 101) self.assertEqual(self.p.balance("human:sam-renamed")["spendable"], 0.0) import threading results, errors = [], [] def claim(): p = CofferProduct(self.db) # separate connection, like separate platform requests try: results.append(p.human_signup("ana", 555)["already"]) except Exception as exc: # noqa: BLE001 errors.append(exc) finally: p.close() threads = [threading.Thread(target=claim) for _ in range(8)] [t.start() for t in threads] [t.join() for t in threads] self.p.close() self.p = CofferProduct(self.db) self.assertEqual(self.p.balance("human:ana")["spendable"], 40.0) self.assertEqual(results.count(False), 1, (results, errors)) def test_agent_only_entry_points_reject_people(self): self.p.human_signup("sam", 101) for fn in (lambda: self.p.grant_signup("human:sam"), lambda: self.p.top_up("human:sam", 5), lambda: self.p.charge_subscription("human:sam"), lambda: self.p.signup_flow("human:sam")): with self.assertRaises(ValueError): fn() self.assertEqual(self.p.balance("human:sam")["spendable"], 40.0) def test_payment_ref_bound_to_request(self): self.p.human_signup("sam", 101) self.p.human_signup("kim", 102) self.p.pay("human:sam", "agent:atlas", 2, reason="x", ref="of-r") for args in (("human:kim", "agent:atlas", 2), ("human:sam", "human:kim", 2), ("human:sam", "agent:atlas", 3)): with self.assertRaises(ProductError): self.p.pay(*args, reason="x", ref="of-r") for bad in (float("nan"), float("inf"), "5", True): with self.assertRaises(ProductError): self.p.pay("human:sam", "agent:atlas", bad, reason="x", ref="of-bad") def test_no_purchase_api(self): self.assertFalse(hasattr(self.p, "buy_credits")) def test_human_pays_agent_with_granted_credit(self): self.p.human_signup("sam", 101) before = self.p.balance("agent:atlas") r = self.p.pay("human:sam", "agent:atlas", 12.5, reason="Research brief", ref="of-1") self.assertEqual(r["tx"]["kind"], "payment") self.assertEqual(self.p.balance("human:sam")["spendable"], 27.5) after = self.p.balance("agent:atlas") self.assertAlmostEqual(after["promotional"] - before["promotional"], 12.5) # kind preserved self.assertTrue(self.p.pay("human:sam", "agent:atlas", 12.5, reason="x", ref="of-1")["duplicate"]) self.p.close() self.p = CofferProduct(self.db) # survives a reload self.assertEqual(self.p.balance("human:sam")["spendable"], 27.5) self.assertTrue(self.p.pay("human:sam", "agent:atlas", 12.5, reason="x", ref="of-1")["duplicate"]) def test_agent_pays_human_and_people_pay_each_other(self): self.p.human_signup("sam", 101) self.p.human_signup("kim", 102) self.p.pay("agent:atlas", "human:sam", 5, reason="bounty", ref="of-2") self.assertEqual(self.p.balance("human:sam")["spendable"], 45.0) self.p.pay("human:sam", "human:kim", 10, reason="thanks", ref="of-3") self.assertEqual(self.p.balance("human:kim")["spendable"], 50.0) self.assertEqual(self.p.balance("human:sam")["spendable"], 35.0) def test_rules(self): self.p.human_signup("sam", 101) self.p.signup_flow("bolt") with self.assertRaises(TransferNotAllowed): self.p.pay("agent:atlas", "agent:bolt", 1, reason="x", ref="of-4") with self.assertRaises(TransferNotAllowed): self.p.pay("human:sam", "company:coffer", 1, reason="x", ref="of-5") with self.assertRaises(ProductError): self.p.pay("human:sam", "agent:atlas", 5000, reason="x", ref="of-6") with self.assertRaises(InsufficientCredit): self.p.pay("human:sam", "agent:atlas", 41, reason="x", ref="of-7") # ----- shirt orders: any credit, held until the operator reviews the order def test_person_orders_with_granted_credit_and_is_rejected_then_released(self): self.p.human_signup("sam", 101) o = self.p.buy_shirt("human:sam") oid = o["order"]["order_id"] self.assertEqual((o["order"]["status"], o["order"]["status_label"]), ("pending_review", "pending review")) self.assertEqual((o["balance"]["spendable"], o["balance"]["held"]), (20.0, 20.0)) self.assertEqual([x["order_id"] for x in self.p.pending_orders()], [oid]) with self.assertRaises(NotAReviewer): self.p.reject_order(oid, "human:sam") with self.assertRaises(NotAReviewer): self.p.accept_order(oid, "agent:atlas") r = self.p.reject_order(oid, "agent:leora", "out of stock this week") self.assertEqual(r["order"]["status"], "rejected") b = self.p.balance("human:sam") self.assertEqual((b["spendable"], b["granted"], b["held"]), (40.0, 40.0, 0.0)) # released in full, same kind with self.assertRaises(OrderNotPending): self.p.accept_order(oid, "agent:leora") self.assertEqual(self.p.pending_orders(), []) kinds = [t["kind"] for t in self.p.history()] self.assertIn("order_hold", kinds) self.assertIn("order_release", kinds) def test_pending_order_limit_and_insufficient_credit(self): self.p.human_signup("sam", 101) self.p.buy_shirt("human:sam") self.p.buy_shirt("human:sam") with self.assertRaises(TooManyPendingOrders): self.p.buy_shirt("human:sam") self.assertEqual(self.p.balance("human:sam")["held"], 40.0) self.p.human_signup("kai", 102) with self.assertRaises(InsufficientCredit): self.p.buy_shirt("human:kai", quantity=3) # 60c > 40c self.assertEqual(self.p.balance("human:kai")["spendable"], 40.0) first = self.p.pending_orders()[0]["order_id"] self.p.accept_order(first, "agent:leora") with self.assertRaises(InsufficientCredit): # the slot is free again; the credit is spent self.p.buy_shirt("human:sam") def test_accept_charges_hold_and_is_public(self): o = self.p.buy_shirt("atlas") a = self.p.accept_order(o["order"]["order_id"], "agent:leora", "ok") self.assertEqual((a["order"]["status"], a["order"]["credits_charged"], a["order"]["reviewed_by"]), ("procuring", 20, "agent:leora")) snap = self.p.public_snapshot() self.assertTrue(any(t["kind"] == "purchase" for t in snap["transactions"])) pending_file = Path(self.db).with_name("pending-orders.json") self.assertTrue(pending_file.exists()) import json self.assertEqual(json.loads(pending_file.read_text())["pending"], []) def test_orders_survive_restart(self): o = self.p.buy_shirt("atlas") self.p.close() self.p = CofferProduct(self.db) self.assertEqual(self.p.pending_orders()[0]["order_id"], o["order"]["order_id"]) self.assertEqual(self.p.balance("atlas")["held"], 20.0) self.p.reject_order(o["order"]["order_id"], "agent:leora") self.assertEqual(self.p.balance("atlas")["spendable"], 20.0) def test_grants_spent_first_and_tracked_apart(self): self.p.human_signup("sam", 101) self.p.top_up("atlas", 30) # atlas: 20 granted + 30 funded self.p.pay("atlas", "human:sam", 25, reason="refund for a service", ref="atlas-sam-1") a = self.p.balance("atlas") self.assertEqual((a["granted"], a["funded"]), (0.0, 25.0)) sam = self.p.balance("human:sam") self.assertEqual((sam["granted"], sam["earned"]), (60.0, 5.0)) def test_public_ledger_shows_handles_only(self): self.p.human_signup("sam", 101) self.p.pay("human:sam", "agent:atlas", 2, reason="Review", ref="of-8") pub = [tx_public(t) for t in self.p.history()] self.assertIn("human:sam", repr(pub)) self.assertEqual(next(t for t in pub if t["kind"] == "payment")["meta"].get("offer"), "of-8") if __name__ == "__main__": unittest.main()