distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for peer history tracking."""
2
3import time
4
5import pytest
6
7
8class TestDBHistory:
9 def test_record_and_success_rate(self):
10 from i2p_peer.history import DBHistory
11
12 h = DBHistory()
13 for _ in range(7):
14 h.record_store(success=True, latency_ms=100.0)
15 for _ in range(3):
16 h.record_store(success=False, latency_ms=200.0)
17 assert abs(h.success_rate - 0.7) < 0.001
18
19 def test_max_entries_eviction(self):
20 from i2p_peer.history import DBHistory
21
22 h = DBHistory(max_entries=10)
23 for i in range(20):
24 h.record_store(success=True, latency_ms=float(i))
25 # Should have only 10 entries
26 assert len(h.recent_entries) == 10
27 # Oldest should be evicted, so min latency should be 10.0
28 latencies = [e.latency_ms for e in h.recent_entries]
29 assert min(latencies) == 10.0
30
31 def test_average_latency(self):
32 from i2p_peer.history import DBHistory
33
34 h = DBHistory()
35 h.record_store(success=True, latency_ms=100.0)
36 h.record_store(success=True, latency_ms=200.0)
37 h.record_lookup(success=True, latency_ms=300.0)
38 assert abs(h.average_latency - 200.0) < 0.001
39
40 def test_average_latency_empty(self):
41 from i2p_peer.history import DBHistory
42
43 h = DBHistory()
44 assert h.average_latency == 0.0
45
46 def test_success_rate_empty(self):
47 from i2p_peer.history import DBHistory
48
49 h = DBHistory()
50 assert h.success_rate == 0.0
51
52 def test_record_lookup(self):
53 from i2p_peer.history import DBHistory
54
55 h = DBHistory()
56 h.record_lookup(success=True, latency_ms=50.0)
57 h.record_lookup(success=False, latency_ms=100.0)
58 assert abs(h.success_rate - 0.5) < 0.001
59
60 def test_recent_entries_returns_list(self):
61 from i2p_peer.history import DBHistory
62
63 h = DBHistory()
64 h.record_store(success=True)
65 entries = h.recent_entries
66 assert len(entries) == 1
67 assert entries[0].success is True
68
69
70class TestTunnelHistory:
71 def test_build_success_rate(self):
72 from i2p_peer.history import TunnelHistory
73
74 h = TunnelHistory()
75 for _ in range(8):
76 h.record_build(success=True)
77 for _ in range(2):
78 h.record_build(success=False)
79 assert abs(h.build_success_rate - 0.8) < 0.001
80
81 def test_participation_rate(self):
82 from i2p_peer.history import TunnelHistory
83
84 h = TunnelHistory()
85 for _ in range(6):
86 h.record_participation(success=True, data_transferred=1024)
87 for _ in range(4):
88 h.record_participation(success=False)
89 assert abs(h.participation_rate - 0.6) < 0.001
90
91 def test_tunnel_history_empty(self):
92 from i2p_peer.history import TunnelHistory
93
94 h = TunnelHistory()
95 assert h.build_success_rate == 0.0
96 assert h.participation_rate == 0.0
97
98 def test_tunnel_history_max_entries(self):
99 from i2p_peer.history import TunnelHistory
100
101 h = TunnelHistory(max_entries=5)
102 for _ in range(10):
103 h.record_build(success=True)
104 # Internal entries should be capped
105 assert len(h._entries) == 5