103 lines
3.6 KiB
Python
103 lines
3.6 KiB
Python
"""Tests for trust-ratio-driven P-AMM exchange rates."""
|
|
|
|
from src.crdt.trust_pamm import (
|
|
PeerTrustMetrics,
|
|
PeerTrustState,
|
|
TrustPAMMParams,
|
|
exchange_rate,
|
|
merge_trust,
|
|
peer_exchange_rate,
|
|
trust_ratio,
|
|
update_trust,
|
|
)
|
|
|
|
|
|
class TestTrustRatio:
|
|
def test_full_delivery_full_trust(self):
|
|
m = PeerTrustMetrics(delivered=100, promised=100, uptime_fraction=1.0, flow_penalty=1.0)
|
|
assert trust_ratio(m) == 1.0
|
|
|
|
def test_partial_delivery(self):
|
|
m = PeerTrustMetrics(delivered=50, promised=100, uptime_fraction=1.0, flow_penalty=1.0)
|
|
assert abs(trust_ratio(m) - 0.5) < 1e-10
|
|
|
|
def test_uptime_scales(self):
|
|
m = PeerTrustMetrics(delivered=100, promised=100, uptime_fraction=0.5, flow_penalty=1.0)
|
|
assert abs(trust_ratio(m) - 0.5) < 1e-10
|
|
|
|
def test_flow_penalty_scales(self):
|
|
m = PeerTrustMetrics(delivered=100, promised=100, uptime_fraction=1.0, flow_penalty=0.8)
|
|
assert abs(trust_ratio(m) - 0.8) < 1e-10
|
|
|
|
def test_no_promises_full_trust(self):
|
|
m = PeerTrustMetrics(delivered=0, promised=0)
|
|
assert trust_ratio(m) == 1.0
|
|
|
|
def test_clamped_to_unit(self):
|
|
m = PeerTrustMetrics(delivered=200, promised=100, uptime_fraction=1.0, flow_penalty=1.0)
|
|
assert trust_ratio(m) == 1.0
|
|
|
|
|
|
class TestExchangeRate:
|
|
def test_full_trust_par_rate(self):
|
|
params = TrustPAMMParams()
|
|
rate = exchange_rate(1.0, params)
|
|
assert rate == 1.0
|
|
|
|
def test_low_trust_discounted(self):
|
|
params = TrustPAMMParams()
|
|
rate = exchange_rate(0.5, params)
|
|
assert rate < 1.0
|
|
assert rate >= params.floor_rate
|
|
|
|
def test_zero_trust_floor(self):
|
|
params = TrustPAMMParams()
|
|
rate = exchange_rate(0.0, params)
|
|
assert rate >= params.floor_rate
|
|
|
|
def test_monotone_with_trust(self):
|
|
params = TrustPAMMParams()
|
|
r_low = exchange_rate(0.3, params)
|
|
r_mid = exchange_rate(0.6, params)
|
|
r_high = exchange_rate(0.9, params)
|
|
assert r_low <= r_mid <= r_high
|
|
|
|
|
|
class TestPeerExchangeRate:
|
|
def test_unknown_peer_par_rate(self):
|
|
state = PeerTrustState()
|
|
params = TrustPAMMParams()
|
|
assert peer_exchange_rate(state, "unknown", params) == 1.0
|
|
|
|
def test_tracked_peer(self):
|
|
state = PeerTrustState()
|
|
state = update_trust(state, "alice", 50, 100, 1.0, 1.0)
|
|
params = TrustPAMMParams()
|
|
rate = peer_exchange_rate(state, "alice", params)
|
|
assert rate < 1.0
|
|
|
|
|
|
class TestTrustMerge:
|
|
def test_merge_commutativity(self):
|
|
a = PeerTrustState(peers={"p1": PeerTrustMetrics(50, 100, 1.0, 1.0)})
|
|
b = PeerTrustState(peers={"p1": PeerTrustMetrics(80, 100, 1.0, 1.0)})
|
|
assert merge_trust(a, b).peers["p1"].delivered == merge_trust(b, a).peers["p1"].delivered
|
|
|
|
def test_merge_idempotency(self):
|
|
a = PeerTrustState(peers={"p1": PeerTrustMetrics(50, 100, 1.0, 1.0)})
|
|
aa = merge_trust(a, a)
|
|
assert trust_ratio(aa.peers["p1"]) == trust_ratio(a.peers["p1"])
|
|
|
|
def test_merge_pessimistic(self):
|
|
a = PeerTrustState(peers={"p1": PeerTrustMetrics(50, 100, 1.0, 1.0)}) # ratio=0.5
|
|
b = PeerTrustState(peers={"p1": PeerTrustMetrics(80, 100, 1.0, 1.0)}) # ratio=0.8
|
|
merged = merge_trust(a, b)
|
|
assert trust_ratio(merged.peers["p1"]) == 0.5 # Takes the lower
|
|
|
|
def test_merge_disjoint(self):
|
|
a = PeerTrustState(peers={"p1": PeerTrustMetrics(50, 100, 1.0, 1.0)})
|
|
b = PeerTrustState(peers={"p2": PeerTrustMetrics(80, 100, 1.0, 1.0)})
|
|
merged = merge_trust(a, b)
|
|
assert "p1" in merged.peers
|
|
assert "p2" in merged.peers
|