distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for peer score calculators."""
2
3import os
4
5import pytest
6
7
8class TestCapacityCalculator:
9 def test_capacity_all_success(self):
10 from i2p_peer.calculator import CapacityCalculator
11 from i2p_peer.profile import PeerProfile
12
13 p = PeerProfile(os.urandom(32))
14 for _ in range(20):
15 p.record_tunnel_build_success()
16 p.record_send_success()
17 p.record_db_store_success()
18 p.heard_from()
19 score = CapacityCalculator.calc(p)
20 # All success should yield high capacity
21 assert score >= 0.8
22 assert score <= 1.0
23
24 def test_capacity_all_failure(self):
25 from i2p_peer.calculator import CapacityCalculator
26 from i2p_peer.profile import PeerProfile
27
28 p = PeerProfile(os.urandom(32))
29 for _ in range(20):
30 p.record_tunnel_build_failure()
31 p.record_send_failure()
32 p.record_db_store_failure()
33 score = CapacityCalculator.calc(p)
34 # All failure should yield low capacity
35 assert score <= 0.15
36
37 def test_capacity_mixed(self):
38 from i2p_peer.calculator import CapacityCalculator
39 from i2p_peer.profile import PeerProfile
40
41 p = PeerProfile(os.urandom(32))
42 for _ in range(7):
43 p.record_tunnel_build_success()
44 for _ in range(3):
45 p.record_tunnel_build_failure()
46 for _ in range(8):
47 p.record_send_success()
48 for _ in range(2):
49 p.record_send_failure()
50 for _ in range(6):
51 p.record_db_store_success()
52 for _ in range(4):
53 p.record_db_store_failure()
54 p.heard_from()
55 score = CapacityCalculator.calc(p)
56 # Mixed should be moderate
57 assert 0.3 < score < 0.9
58
59 def test_capacity_no_history(self):
60 from i2p_peer.calculator import CapacityCalculator
61 from i2p_peer.profile import PeerProfile
62
63 p = PeerProfile(os.urandom(32))
64 score = CapacityCalculator.calc(p)
65 assert score == 0.0
66
67 def test_capacity_range(self):
68 from i2p_peer.calculator import CapacityCalculator
69 from i2p_peer.profile import PeerProfile
70
71 p = PeerProfile(os.urandom(32))
72 for _ in range(5):
73 p.record_tunnel_build_success()
74 score = CapacityCalculator.calc(p)
75 assert 0.0 <= score <= 1.0
76
77
78class TestSpeedCalculator:
79 def test_speed_low_latency(self):
80 from i2p_peer.calculator import SpeedCalculator
81 from i2p_peer.profile import PeerProfile
82
83 p = PeerProfile(os.urandom(32))
84 for _ in range(10):
85 p.record_latency(50.0) # 50ms is fast
86 score = SpeedCalculator.calc(p)
87 assert score >= 0.7
88 assert score <= 1.0
89
90 def test_speed_high_latency(self):
91 from i2p_peer.calculator import SpeedCalculator
92 from i2p_peer.profile import PeerProfile
93
94 p = PeerProfile(os.urandom(32))
95 for _ in range(10):
96 p.record_latency(5000.0) # 5 seconds is slow
97 score = SpeedCalculator.calc(p)
98 assert score <= 0.3
99
100 def test_speed_no_latency(self):
101 from i2p_peer.calculator import SpeedCalculator
102 from i2p_peer.profile import PeerProfile
103
104 p = PeerProfile(os.urandom(32))
105 score = SpeedCalculator.calc(p)
106 assert score == 0.0
107
108 def test_speed_range(self):
109 from i2p_peer.calculator import SpeedCalculator
110 from i2p_peer.profile import PeerProfile
111
112 p = PeerProfile(os.urandom(32))
113 p.record_latency(500.0)
114 score = SpeedCalculator.calc(p)
115 assert 0.0 <= score <= 1.0
116
117
118class TestIntegrationCalculator:
119 def test_integration_from_db_ops(self):
120 from i2p_peer.calculator import IntegrationCalculator
121 from i2p_peer.profile import PeerProfile
122
123 p = PeerProfile(os.urandom(32))
124 for _ in range(10):
125 p.record_db_store_success()
126 p.record_db_lookup_success()
127 score = IntegrationCalculator.calc(p)
128 assert score >= 0.8
129 assert score <= 1.0
130
131 def test_integration_no_ops(self):
132 from i2p_peer.calculator import IntegrationCalculator
133 from i2p_peer.profile import PeerProfile
134
135 p = PeerProfile(os.urandom(32))
136 score = IntegrationCalculator.calc(p)
137 assert score == 0.0
138
139 def test_integration_mixed(self):
140 from i2p_peer.calculator import IntegrationCalculator
141 from i2p_peer.profile import PeerProfile
142
143 p = PeerProfile(os.urandom(32))
144 for _ in range(5):
145 p.record_db_store_success()
146 for _ in range(5):
147 p.record_db_store_failure()
148 score = IntegrationCalculator.calc(p)
149 assert 0.3 < score < 0.7