"""Research simulator tests, not blockchain consensus tests.""" from __future__ import annotations import importlib.util import json import math import tempfile import unittest from pathlib import Path SOURCE = Path(__file__).resolve().parents[1] / "scripts" / "simulate_pop_dynamic.py" spec = importlib.util.spec_from_file_location("dynamic", SOURCE) mod = importlib.util.module_from_spec(spec) import sys sys.modules[spec.name] = mod spec.loader.exec_module(mod) class FeeTests(unittest.TestCase): def test_worker_fee_staffel(self): assert mod.worker_fee(0, 5) == 0 self.assertAlmostEqual(mod.worker_fee(3, 1), .003) self.assertAlmostEqual(mod.worker_fee(7, 1), .015) self.assertAlmostEqual(mod.worker_fee(8, 1), .020) self.assertAlmostEqual(mod.worker_fee(10, 5), .150) def test_invalid_values(self): for n in [-1, 11]: with self.assertRaises(ValueError): mod.worker_fee(n, 1) for scale in [-2, math.nan, math.inf]: with self.assertRaises(ValueError): mod.worker_fee(2, scale) with self.assertRaises(ValueError): mod.simulate(operators=1) with self.assertRaises(ValueError): mod.simulate(capacities=(12,)) with self.assertRaises(ValueError): mod.simulate(fund_mode="burn") def test_window_fee_progressivity(self): self.assertAlmostEqual(mod.progressive_fee(10), .0005) self.assertAlmostEqual(mod.progressive_fee(100), .0185) self.assertGreater(mod.window_fee(3, 100), mod.window_fee(2,100)) with self.assertRaises(ValueError): mod.window_fee(1, 0) def test_worker_decision_and_budget(self): f = mod.choose_worker_count kw = dict(capacity=10, scale=5, rivals_estimated_tickets=12, f1_chance=.55, reward=1.) self.assertEqual(f(balance=0., mode="fixed", **kw), 0) self.assertEqual(f(balance=2., mode="fixed", **kw), 10) choice = f(balance=2., mode="dynamic", **kw) self.assertLessEqual(choice, 10) self.assertGreaterEqual(choice, 0) # A much stronger levy must make additional workers less worthwhile. high = f(balance=20., mode="dynamic", scale=100, capacity=10, rivals_estimated_tickets=12, f1_chance=.55, reward=1.) self.assertEqual(high, 0) def test_ticket_marginal(self): self.assertAlmostEqual(mod.marginal_ticket_gain(0,4,1), .2) self.assertAlmostEqual(mod.marginal_ticket_gain(1,4,1), 4/30) self.assertAlmostEqual(mod.marginal_ticket_gain(4,0,1), 0) class SimulationTests(unittest.TestCase): def short(self, **kwargs): defaults = dict(operators=6, newcomers=2, join_day=3, days=8, rounds_per_day=30, seed=123, capacities=(0,1,3,5,8,10)) defaults.update(kwargs) return mod.simulate(**defaults) def test_fund_holds_all_fee_income(self): r = self.short(fund_mode="hold", worker_mode="fixed", worker_scale=3) s = r["summary"] self.assertEqual(s["burned"], 0) self.assertAlmostEqual(s["fund_balance"], s["fund_received_fees"], 4) self.assertEqual(s["fund_distributed"], 0) self.assertAlmostEqual(s["fund_paid_window"]+s["fund_paid_worker"],s["fund_received_fees"],4) self.assertAlmostEqual(s["total_wallets"] + s["fund_balance"], s["total_supply"],4) def test_redistribution_conserves_ptm_and_creates_no_new_coins(self): r = self.short(fund_mode="redistribute", payout_interval=2, payout_share=.5, worker_mode="fixed", worker_scale=5) s = r["summary"] self.assertGreater(s["fund_distributed"], 0) self.assertGreater(s["fund_balance"], 0) self.assertAlmostEqual(s["fund_balance"]+s["fund_distributed"], s["fund_received_fees"], 4) self.assertAlmostEqual(s["total_wallets"]+s["fund_balance"], s["total_supply"], 4) self.assertAlmostEqual(sum(o["fund_received"] for o in r["operators"]),s["fund_distributed"], 4) def test_free_hubs_no_fees_no_fund(self): r = self.short(capacities=(0,), window_policy="free",worker_mode="dynamic") self.assertEqual(r["summary"]["fund_received_fees"], 0) self.assertEqual(r["summary"]["fund_balance"], 0) self.assertAlmostEqual(r["summary"]["minted"],r["summary"]["total_wallets"],4) def test_one_winner_per_round_and_ticket_limit(self): r = self.short(worker_mode="dynamic",fund_mode="redistribute",payout_interval=1) s = r["summary"] self.assertEqual(sum(o["wins"] for o in r["operators"])+s["no_ticket_rounds"],s["rounds"]) for op in r["operators"]: self.assertLessEqual(op["tickets"],op["active_rounds"]*(1+op["capacity"])) self.assertLessEqual(sum(o["tickets"] for o in r["operators"]), sum(op["active_rounds"]*(1+op["capacity"]) for op in r["operators"])) def test_newcomers_only_start_on_join_day(self): r = self.short(join_day=4, days=8, rounds_per_day=7) newcomers = [o for o in r["operators"] if o["join_day"] > 0] self.assertEqual(len(newcomers),2) self.assertTrue(all(o["active_rounds"]==4*7 for o in newcomers)) self.assertEqual(r["daily"][2]["active_operators"],6) self.assertEqual(r["daily"][4]["active_operators"],8) def test_fixed_vs_dynamic_different_for_high_cost(self): args = dict(worker_scale=30, fund_mode="hold",initial_balance=20.) a = self.short(worker_mode="fixed",**args) b = self.short(worker_mode="dynamic",**args) self.assertGreater(a["summary"]["total_avg_workers"],b["summary"]["total_avg_workers"]) def test_determinism(self): settings = dict(worker_mode="dynamic", fund_mode="redistribute", payout_interval=2) self.assertEqual(self.short(**settings), self.short(**settings)) def test_zero_payout_matches_hold_except_mode(self): a = self.short(fund_mode="hold") b = self.short(fund_mode="redistribute",payout_share=0,payout_interval=2) self.assertEqual(a["summary"],b["summary"]) self.assertEqual(a["daily"],b["daily"]) self.assertEqual(a["operators"],b["operators"]) def test_rolling_90day_expiry(self): op = mod.Operator(1,1,0,0.) op.reward(0,2.) self.assertEqual(op.recent,2.) op.expire(90) self.assertEqual(op.recent,0.) self.assertEqual(op.history[0],0.) def test_export(self): a = self.short() b = self.short(worker_mode="fixed") with tempfile.TemporaryDirectory() as temp: paths = mod.write_output(Path(temp),[a,b],mod.aggregate([a,b]),"test") self.assertEqual(len(paths),5) self.assertTrue(all(p.is_file() and p.stat().st_size>0 for p in paths)) payload = json.loads(paths[4].read_text()) self.assertEqual(payload["scenario_count"],2) self.assertEqual(len(payload["comparisons"]),2) if __name__=="__main__": unittest.main()