distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for peer profiles."""
2
3import os
4import time
5
6import pytest
7
8
9class TestPeerProfile:
10 def test_construct(self):
11 from i2p_peer.profile import PeerProfile
12 h = os.urandom(32)
13 p = PeerProfile(router_hash=h)
14 assert p.router_hash == h
15 assert p.capacity == 0.0
16 assert p.integration == 0.0
17
18 def test_update_capacity(self):
19 from i2p_peer.profile import PeerProfile
20 p = PeerProfile(os.urandom(32))
21 p.record_success()
22 assert p.capacity > 0
23
24 def test_record_failure(self):
25 from i2p_peer.profile import PeerProfile
26 p = PeerProfile(os.urandom(32))
27 p.record_success()
28 old = p.capacity
29 p.record_failure()
30 assert p.capacity < old
31
32 def test_tunnel_build_success(self):
33 from i2p_peer.profile import PeerProfile
34 p = PeerProfile(os.urandom(32))
35 p.tunnel_build_success()
36 assert p.tunnel_builds_succeeded == 1
37 assert p.tunnel_builds_failed == 0
38
39 def test_tunnel_build_failure(self):
40 from i2p_peer.profile import PeerProfile
41 p = PeerProfile(os.urandom(32))
42 p.tunnel_build_failure()
43 assert p.tunnel_builds_failed == 1
44
45 def test_tunnel_success_rate(self):
46 from i2p_peer.profile import PeerProfile
47 p = PeerProfile(os.urandom(32))
48 p.tunnel_build_success()
49 p.tunnel_build_success()
50 p.tunnel_build_failure()
51 assert abs(p.tunnel_success_rate - 2/3) < 0.01
52
53 def test_tunnel_success_rate_no_builds(self):
54 from i2p_peer.profile import PeerProfile
55 p = PeerProfile(os.urandom(32))
56 assert p.tunnel_success_rate == 0.0
57
58 def test_last_heard_from(self):
59 from i2p_peer.profile import PeerProfile
60 p = PeerProfile(os.urandom(32))
61 now = int(time.time() * 1000)
62 p.update_last_heard(now)
63 assert p.last_heard_from == now
64
65 def test_decay(self):
66 from i2p_peer.profile import PeerProfile
67 p = PeerProfile(os.urandom(32))
68 for _ in range(10):
69 p.record_success()
70 old_cap = p.capacity
71 p.decay(factor=0.5)
72 assert p.capacity < old_cap
73
74 def test_score(self):
75 from i2p_peer.profile import PeerProfile
76 p = PeerProfile(os.urandom(32))
77 p.record_success()
78 p.tunnel_build_success()
79 score = p.score()
80 assert score > 0.0
81
82
83class TestPeerSelector:
84 def test_select_best(self):
85 from i2p_peer.profile import PeerProfile, PeerSelector
86 profiles = []
87 for _ in range(5):
88 p = PeerProfile(os.urandom(32))
89 for _ in range(3):
90 p.record_success()
91 profiles.append(p)
92 # Make one clearly better
93 for _ in range(20):
94 profiles[2].record_success()
95 profiles[2].tunnel_build_success()
96
97 selector = PeerSelector(profiles)
98 best = selector.select_best(n=2)
99 assert len(best) == 2
100 assert profiles[2].router_hash in [p.router_hash for p in best]
101
102 def test_select_with_exclusions(self):
103 from i2p_peer.profile import PeerProfile, PeerSelector
104 profiles = [PeerProfile(os.urandom(32)) for _ in range(5)]
105 for p in profiles:
106 p.record_success()
107 selector = PeerSelector(profiles)
108 excluded = {profiles[0].router_hash, profiles[1].router_hash}
109 result = selector.select_best(n=5, exclude=excluded)
110 for p in result:
111 assert p.router_hash not in excluded
112
113 def test_select_empty(self):
114 from i2p_peer.profile import PeerSelector
115 selector = PeerSelector([])
116 assert selector.select_best(n=3) == []
117
118
119class TestBanManager:
120 def test_ban_peer(self):
121 from i2p_peer.profile import BanManager
122 bm = BanManager()
123 h = os.urandom(32)
124 bm.ban(h, reason="protocol violation", duration_ms=60000)
125 assert bm.is_banned(h)
126
127 def test_not_banned(self):
128 from i2p_peer.profile import BanManager
129 bm = BanManager()
130 assert not bm.is_banned(os.urandom(32))
131
132 def test_ban_expires(self):
133 from i2p_peer.profile import BanManager
134 bm = BanManager()
135 h = os.urandom(32)
136 bm.ban(h, reason="test", duration_ms=1000, now_ms=0)
137 assert bm.is_banned(h, now_ms=500)
138 assert not bm.is_banned(h, now_ms=2000)
139
140 def test_unban(self):
141 from i2p_peer.profile import BanManager
142 bm = BanManager()
143 h = os.urandom(32)
144 bm.ban(h, reason="test", duration_ms=60000)
145 bm.unban(h)
146 assert not bm.is_banned(h)
147
148 def test_ban_count(self):
149 from i2p_peer.profile import BanManager
150 bm = BanManager()
151 for _ in range(3):
152 bm.ban(os.urandom(32), reason="test", duration_ms=60000)
153 assert bm.ban_count() == 3
154
155 def test_cleanup_expired(self):
156 from i2p_peer.profile import BanManager
157 bm = BanManager()
158 bm.ban(os.urandom(32), reason="test", duration_ms=100, now_ms=0)
159 bm.ban(os.urandom(32), reason="test", duration_ms=99999, now_ms=0)
160 bm.cleanup(now_ms=500)
161 assert bm.ban_count() == 1