import tempfile import unittest from pathlib import Path from coffer.ledger import GrantedCreditNotForGoods, InsufficientCredit, TransferNotAllowed 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") # ----- granted vs earned: free grants buy services, never physical goods def test_grants_cannot_become_goods(self): # Sybil farming: many people each pay their 40c grant to one account, which tries to buy shirts for i in range(3): self.p.human_signup(f"sock{i}", 900 + i) self.p.pay(f"human:sock{i}", "human:collector" if i else "human:collector", 40, reason="gift", ref=f"farm-{i}") bal = self.p.balance("human:collector") self.assertEqual((bal["granted"], bal["earned"], bal["goods_spendable"]), (120.0, 0.0, 0.0)) self.p.pay("human:collector", "atlas", 100, reason="service", ref="farm-to-agent") a = self.p.balance("atlas") self.assertEqual(a["goods_spendable"], 0.0) # still granted credit at the agent with self.assertRaises(GrantedCreditNotForGoods): self.p.buy_shirt("atlas") def test_earned_credit_can_buy_goods_and_grants_spend_first(self): self.p.human_signup("sam", 101) self.p.top_up("atlas", 30) # atlas: 20 granted + 30 funded r = 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)) # granted spent first sam = self.p.balance("human:sam") self.assertEqual((sam["granted"], sam["earned"]), (60.0, 5.0)) # 20 granted + 5 funded arrived as earned self.p.pay("human:sam", "atlas", 45, reason="service", ref="sam-atlas-1") # 40+20 granted first self.assertEqual(self.p.balance("human:sam")["earned"], 5.0) out = self.p.buy_shirt("atlas") # 20c shirt from funded credit self.assertEqual(out["order"]["credits_charged"], 20) self.assertEqual(self.p.balance("atlas")["granted"], 45.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()