| """Unit tests for CreditLedger.""" |
| import sys |
| from pathlib import Path |
| sys.path.insert(0, str(Path(__file__).parent.parent)) |
|
|
| from ledger.ledger import CreditLedger |
|
|
| def test_earn_and_balance(): |
| ledger = CreditLedger(decay_lambda=0.0) |
| ledger.earn("agent1", "task1", "action1", 10.0, 1.0, 100.0, "test", "model_call") |
| bal = ledger.balance("agent1", "model_call") |
| assert abs(bal - 10.0) < 0.01, f"Expected 10.0, got {bal}" |
| print("PASS: test_earn_and_balance") |
|
|
| def test_spend(): |
| ledger = CreditLedger(decay_lambda=0.0) |
| ledger.earn("agent1", "task1", "action1", 10.0, 1.0, 100.0, "test") |
| ok = ledger.spend("agent1", "task2", "action2", 3.0) |
| assert ok |
| bal = ledger.balance("agent1") |
| assert abs(bal - 7.0) < 0.01, f"Expected 7.0, got {bal}" |
| print("PASS: test_spend") |
|
|
| def test_spend_insufficient(): |
| ledger = CreditLedger(decay_lambda=0.0) |
| ledger.earn("agent1", "task1", "action1", 2.0, 1.0, 100.0, "test") |
| ok = ledger.spend("agent1", "task2", "action2", 5.0) |
| assert not ok |
| print("PASS: test_spend_insufficient") |
|
|
| def test_transfer_blocked(): |
| ledger = CreditLedger(decay_lambda=0.0) |
| ledger.earn("alice", "task1", "action1", 10.0, 1.0, 100.0, "test") |
| ok = ledger.transfer("alice", "bob", 5.0) |
| assert not ok |
| assert ledger.balance("alice") > 9.0, "Credits should not be transferred" |
| assert ledger.balance("bob") < 0.1, "Bob should not receive credits" |
| print("PASS: test_transfer_blocked") |
|
|
| def test_decay(): |
| ledger = CreditLedger(decay_lambda=0.1) |
| ledger.earn("agent1", "task1", "action1", 10.0, 1.0, 100.0, "test") |
| import time; time.sleep(0.05) |
| bal = ledger.balance("agent1") |
| assert bal < 10.0, f"Expected <10.0 after decay, got {bal}" |
| print("PASS: test_decay") |
|
|
| def test_capability_scope(): |
| ledger = CreditLedger(decay_lambda=0.0) |
| ledger.earn("agent1", "t1", "a1", 10.0, 1.0, 100.0, "test", "retrieval") |
| ledger.earn("agent1", "t1", "a1", 5.0, 1.0, 100.0, "test", "model_call") |
| assert abs(ledger.balance("agent1", "retrieval") - 10.0) < 0.01 |
| assert abs(ledger.balance("agent1", "model_call") - 5.0) < 0.01 |
| print("PASS: test_capability_scope") |
|
|
| def run_all(): |
| test_earn_and_balance() |
| test_spend() |
| test_spend_insufficient() |
| test_transfer_blocked() |
| test_decay() |
| test_capability_scope() |
| print("\nAll ledger tests passed!") |
|
|
| if __name__ == "__main__": |
| run_all() |
|
|
