distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for extended PeerProfile with full profiling metrics."""
2
3import os
4import time
5
6import pytest
7
8
9class TestPeerProfileExtended:
10 """Tests for the extended PeerProfile fields and methods."""
11
12 def test_initial_state(self):
13 from i2p_peer.profile import PeerProfile
14
15 h = os.urandom(32)
16 p = PeerProfile(router_hash=h)
17 assert p.router_hash == h
18 assert p.tunnel_builds_succeeded == 0
19 assert p.tunnel_builds_failed == 0
20 assert p.tunnel_builds_rejected == 0
21 assert p.send_success_count == 0
22 assert p.send_failure_count == 0
23 assert p.db_store_success_count == 0
24 assert p.db_store_failure_count == 0
25 assert p.db_lookup_success_count == 0
26 assert p.db_lookup_failure_count == 0
27 assert p.last_heard_from == 0.0
28 assert p.first_heard_about == 0.0
29 assert p.last_send_success == 0.0
30 assert p.last_send_failure == 0.0
31 assert p.capacity == 0.0
32 assert p.speed == 0.0
33 assert p.integration == 0.0
34 assert p.is_expanding is False
35 assert p.is_active is True
36 assert p.is_banned is False
37 assert p.ban_expiry == 0.0
38
39 def test_tunnel_build_success_rate(self):
40 from i2p_peer.profile import PeerProfile
41
42 p = PeerProfile(os.urandom(32))
43 for _ in range(7):
44 p.record_tunnel_build_success()
45 for _ in range(3):
46 p.record_tunnel_build_failure()
47 assert p.tunnel_builds_succeeded == 7
48 assert p.tunnel_builds_failed == 3
49 assert abs(p.tunnel_success_rate - 0.7) < 0.001
50
51 def test_tunnel_success_rate_no_builds(self):
52 from i2p_peer.profile import PeerProfile
53
54 p = PeerProfile(os.urandom(32))
55 assert p.tunnel_success_rate == 0.0
56
57 def test_tunnel_build_rejection(self):
58 from i2p_peer.profile import PeerProfile
59
60 p = PeerProfile(os.urandom(32))
61 p.record_tunnel_build_rejection()
62 p.record_tunnel_build_rejection()
63 assert p.tunnel_builds_rejected == 2
64
65 def test_send_success_rate(self):
66 from i2p_peer.profile import PeerProfile
67
68 p = PeerProfile(os.urandom(32))
69 for _ in range(8):
70 p.record_send_success()
71 for _ in range(2):
72 p.record_send_failure()
73 assert p.send_success_count == 8
74 assert p.send_failure_count == 2
75 assert abs(p.send_success_rate - 0.8) < 0.001
76
77 def test_send_success_rate_no_sends(self):
78 from i2p_peer.profile import PeerProfile
79
80 p = PeerProfile(os.urandom(32))
81 assert p.send_success_rate == 0.0
82
83 def test_db_success_rate(self):
84 from i2p_peer.profile import PeerProfile
85
86 p = PeerProfile(os.urandom(32))
87 for _ in range(5):
88 p.record_db_store_success()
89 for _ in range(3):
90 p.record_db_lookup_success()
91 for _ in range(2):
92 p.record_db_store_failure()
93 # 8 success out of 10 total
94 assert abs(p.db_success_rate - 0.8) < 0.001
95
96 def test_db_success_rate_no_ops(self):
97 from i2p_peer.profile import PeerProfile
98
99 p = PeerProfile(os.urandom(32))
100 assert p.db_success_rate == 0.0
101
102 def test_latency_recording(self):
103 from i2p_peer.profile import PeerProfile
104
105 p = PeerProfile(os.urandom(32))
106 p.record_latency(100.0)
107 p.record_latency(200.0)
108 p.record_latency(300.0)
109 assert abs(p.average_latency - 200.0) < 0.001
110
111 def test_latency_no_samples(self):
112 from i2p_peer.profile import PeerProfile
113
114 p = PeerProfile(os.urandom(32))
115 assert p.average_latency == 0.0
116
117 def test_latency_max_samples(self):
118 from i2p_peer.profile import PeerProfile
119
120 p = PeerProfile(os.urandom(32))
121 # Fill with 50 samples of 100ms
122 for _ in range(50):
123 p.record_latency(100.0)
124 # Add 10 more of 200ms (should evict oldest)
125 for _ in range(10):
126 p.record_latency(200.0)
127 # Should have 50 samples: 40 * 100 + 10 * 200 = 6000 / 50 = 120
128 assert abs(p.average_latency - 120.0) < 0.001
129
130 def test_ban_and_unban(self):
131 from i2p_peer.profile import PeerProfile
132
133 p = PeerProfile(os.urandom(32))
134 assert not p.is_currently_banned
135 p.ban(60.0) # 60 seconds
136 assert p.is_banned is True
137 assert p.ban_expiry > time.time()
138 assert p.is_currently_banned
139 p.unban()
140 assert p.is_banned is False
141 assert not p.is_currently_banned
142
143 def test_ban_expiry(self):
144 from i2p_peer.profile import PeerProfile
145
146 p = PeerProfile(os.urandom(32))
147 # Ban for 0 seconds (immediately expired)
148 p.ban(0.0)
149 assert not p.is_currently_banned
150
151 def test_heard_from_updates_timestamp(self):
152 from i2p_peer.profile import PeerProfile
153
154 p = PeerProfile(os.urandom(32))
155 assert p.last_heard_from == 0.0
156 before = time.time()
157 p.heard_from()
158 after = time.time()
159 assert before <= p.last_heard_from <= after
160
161 def test_first_heard_about_set_once(self):
162 from i2p_peer.profile import PeerProfile
163
164 p = PeerProfile(os.urandom(32))
165 p.heard_from()
166 first = p.first_heard_about
167 assert first > 0.0
168 time.sleep(0.01)
169 p.heard_from()
170 # first_heard_about should not change
171 assert p.first_heard_about == first
172
173 def test_to_dict_from_dict_roundtrip(self):
174 from i2p_peer.profile import PeerProfile
175
176 h = os.urandom(32)
177 p = PeerProfile(h)
178 for _ in range(5):
179 p.record_tunnel_build_success()
180 for _ in range(2):
181 p.record_tunnel_build_failure()
182 p.record_tunnel_build_rejection()
183 for _ in range(3):
184 p.record_send_success()
185 p.record_send_failure()
186 p.record_db_store_success()
187 p.record_db_lookup_failure()
188 p.record_latency(150.0)
189 p.record_latency(250.0)
190 p.heard_from()
191 p.capacity = 0.75
192 p.speed = 0.6
193 p.integration = 0.5
194 p.is_expanding = True
195
196 d = p.to_dict()
197 p2 = PeerProfile.from_dict(d)
198
199 assert p2.router_hash == h
200 assert p2.tunnel_builds_succeeded == 5
201 assert p2.tunnel_builds_failed == 2
202 assert p2.tunnel_builds_rejected == 1
203 assert p2.send_success_count == 3
204 assert p2.send_failure_count == 1
205 assert p2.db_store_success_count == 1
206 assert p2.db_lookup_failure_count == 1
207 assert abs(p2.capacity - 0.75) < 0.001
208 assert abs(p2.speed - 0.6) < 0.001
209 assert abs(p2.integration - 0.5) < 0.001
210 assert p2.is_expanding is True
211 assert len(p2._latency_samples) == 2
212 assert p2.last_heard_from == p.last_heard_from
213
214 def test_is_established(self):
215 from i2p_peer.profile import PeerProfile
216
217 p = PeerProfile(os.urandom(32))
218 # Not established with no history
219 assert not p.is_established
220 # Need at least some tunnel builds and sends
221 for _ in range(5):
222 p.record_tunnel_build_success()
223 for _ in range(5):
224 p.record_send_success()
225 assert p.is_established
226
227 def test_record_db_lookup_success(self):
228 from i2p_peer.profile import PeerProfile
229
230 p = PeerProfile(os.urandom(32))
231 p.record_db_lookup_success()
232 assert p.db_lookup_success_count == 1
233
234 def test_record_db_lookup_failure(self):
235 from i2p_peer.profile import PeerProfile
236
237 p = PeerProfile(os.urandom(32))
238 p.record_db_lookup_failure()
239 assert p.db_lookup_failure_count == 1
240
241 def test_send_timestamps_updated(self):
242 from i2p_peer.profile import PeerProfile
243
244 p = PeerProfile(os.urandom(32))
245 before = time.time()
246 p.record_send_success()
247 assert p.last_send_success >= before
248 p.record_send_failure()
249 assert p.last_send_failure >= before