"""Off-chain PoP cluster simulator tests; no chain implementation is tested.""" from __future__ import annotations import importlib.util import math import sys import unittest from pathlib import Path SCRIPT = Path(__file__).resolve().parents[1] / "scripts" / "simulate_pop_cluster.py" SPEC = importlib.util.spec_from_file_location("simulate_pop_cluster", SCRIPT) mod = importlib.util.module_from_spec(SPEC) sys.modules[SPEC.name] = mod SPEC.loader.exec_module(mod) class PopClusterTests(unittest.TestCase): def test_worker_fees_reproduce_v04_examples(self): expected = {0: 0, 1: .001, 3: .003, 4: .006, 7: .015, 8: .020, 10: .030} for count, fee in expected.items(): self.assertAlmostEqual(mod.worker_fee(count, 'marginal'), fee) self.assertAlmostEqual(mod.worker_fee(7, 'tier'), .003) self.assertAlmostEqual(mod.worker_fee(10, 'tier'), .005) self.assertAlmostEqual(mod.worker_fee(10, 'marginal', 2), .060) def test_window_fees_mirror_offline_v02(self): self.assertAlmostEqual(mod.window_fee(1, 100, 'low'), 0) self.assertAlmostEqual(mod.progressive_fee(100), .0185) self.assertAlmostEqual(mod.window_fee(2, 100, 'low'), .001185) self.assertAlmostEqual(mod.window_fee(3, 100, 'low'), .00337) self.assertAlmostEqual(mod.window_fee(2, 0, 'mild'), .001) def test_marginal_ticket_value_declines_for_owner_with_existing_tickets(self): self.assertAlmostEqual(mod.marginal_share_value(1, 0, 4), 0.2) self.assertAlmostEqual(mod.marginal_share_value(1, 1, 4), 4 / 30) self.assertAlmostEqual(mod.marginal_share_value(1, 3, 0), 0) def test_validations(self): for count in (-1, 11): with self.assertRaises(ValueError): mod.worker_fee(count) with self.assertRaises(ValueError): mod.worker_fee(2, scale=float('nan')) with self.assertRaises(ValueError): mod.simulate(operators=0) with self.assertRaises(ValueError): mod.simulate(operators=3, target_workers=11) with self.assertRaises(ValueError): mod.simulate(operators=3, target_workers=0, worker_scale=-1) def test_worker_activation_is_upfront_and_budget_limited(self): op = mod.Operator(0, True, 10, 'always', balance=.006) activated = mod.activate_workers(op, 'marginal', 1) self.assertEqual(activated, 4) self.assertAlmostEqual(op.balance, 0) self.assertAlmostEqual(op.burned_worker, .006) self.assertEqual(op.skipped_worker_budget, 6) op2 = mod.Operator(1, True, 10, 'always', balance=0) self.assertEqual(mod.activate_workers(op2, 'marginal', 1), 0) self.assertAlmostEqual(op2.burned_worker, 0) def test_one_payout_per_round_and_supply_invariant(self): result = mod.simulate(operators=4, target_workers=3, days=3, rounds_per_day=25, seed=123, target_policy='always', others_policy='free', window_model='low', initial_balance=2) summ = result['summary'] self.assertEqual(summ['rounds'], 75) self.assertEqual(sum(op['wins'] for op in result['operators']), 75 - summ['no_ticket_rounds']) self.assertAlmostEqual(summ['minted'] - summ['burned_total'] + 8, summ['circulating'], places=5) self.assertGreater(summ['burned_workers'], 0) self.assertGreater(summ['burned_windows'], 0) self.assertGreater(result['operators'][0]['window2'], 0) self.assertGreater(result['operators'][0]['window3'], 0) self.assertEqual(sum(op['tickets'] for op in result['operators']), summ['total_tickets']) self.assertEqual(result['operators'][0]['wallet'], summ['target_wallet']) def test_seed_is_deterministic(self): args = dict(operators=5, days=2, rounds_per_day=30, target_workers=5, others_workers=1, seed=567, initial_balance=5) self.assertEqual(mod.simulate(**args), mod.simulate(**args)) def test_zero_fees_with_free_only_equals_minted(self): data = mod.simulate(operators=5, days=2, rounds_per_day=120, target_workers=0, target_policy='free', others_policy='free') self.assertEqual(data['summary']['burned_total'], 0) self.assertAlmostEqual(sum(x['wallet'] for x in data['operators']), data['summary']['minted'], places=5) def test_individual_operator_90_day_history_expires(self): op = mod.Operator(0, True, 0, 'free') op.add_win(0, 1) self.assertEqual(op.recent, 1) for day in range(1, 90): op.expire_day(day) self.assertEqual(op.recent, 1) op.expire_day(90) self.assertEqual(op.recent, 0) def test_newcomers_join_at_configured_day(self): data = mod.simulate(operators=3, newcomers=2, join_day=2, days=4, rounds_per_day=8, target_policy='free', others_policy='free') self.assertEqual([d['active_operators'] for d in data['daily']], [3, 3, 5, 5]) self.assertEqual(len(data['operators']), 5) def test_comparison_delta_normalization(self): results = [mod.simulate(operators=3, target_workers=w, days=1, rounds_per_day=50, seed=1, target_policy='free') for w in (0, 2)] rows = mod.compare(results) self.assertIsNone(rows[0]['delta_net_per_added_worker']) self.assertAlmostEqual(rows[1]['delta_net_per_added_worker'], (rows[1]['net_ptm']-rows[0]['net_ptm'])/2) if __name__ == '__main__': unittest.main()