distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for i2p_stat.rate — Rate tracking."""
2
3import time
4from i2p_stat.rate import Rate, RateSummaryListener
5
6
7class TestRate:
8 def test_create_rate(self):
9 r = Rate(60000)
10 assert r.period == 60000
11 assert r.get_current_event_count() == 0
12 assert r.get_lifetime_event_count() == 0
13
14 def test_add_data(self):
15 r = Rate(60000)
16 r.add_data(100)
17 assert r.get_current_event_count() == 1
18 assert r.get_current_total_value() == 100.0
19 assert r.get_lifetime_event_count() == 1
20 assert r.get_lifetime_total_value() == 100.0
21
22 def test_add_multiple(self):
23 r = Rate(60000)
24 r.add_data(10)
25 r.add_data(20)
26 r.add_data(30)
27 assert r.get_current_event_count() == 3
28 assert r.get_current_total_value() == 60.0
29 assert r.get_lifetime_event_count() == 3
30
31 def test_coalesce_shifts_current_to_last(self):
32 # Use very short period for test
33 r = Rate(100) # 100ms period
34 r.add_data(50)
35 r.add_data(50)
36 time.sleep(0.15) # Wait > period
37 r.coalesce()
38 assert r.get_current_event_count() == 0
39 assert r.get_last_event_count() > 0
40 assert r.get_last_total_value() > 0
41
42 def test_coalesce_too_early_is_noop(self):
43 r = Rate(60000) # 60s period
44 r.add_data(100)
45 r.coalesce() # Should be a no-op (< period - SLACK)
46 assert r.get_current_event_count() == 1 # Still in current
47 assert r.get_last_event_count() == 0
48
49 def test_average_value(self):
50 r = Rate(100)
51 r.add_data(10)
52 r.add_data(30)
53 time.sleep(0.15)
54 r.coalesce()
55 avg = r.get_average_value()
56 # Average should be approximately (10+30)/2 = 20
57 assert avg > 0
58
59 def test_lifetime_average(self):
60 r = Rate(60000)
61 r.add_data(10)
62 r.add_data(20)
63 r.add_data(30)
64 assert r.get_lifetime_average_value() == 20.0
65
66 def test_extreme_tracks_highest(self):
67 r = Rate(100)
68 # Period 1: value 100
69 r.add_data(100)
70 time.sleep(0.15)
71 r.coalesce()
72 # Period 2: value 50
73 r.add_data(50)
74 time.sleep(0.15)
75 r.coalesce()
76 # Extreme should still be from period 1
77 assert r.get_extreme_total_value() >= r.get_last_total_value()
78
79 def test_invalid_period_raises(self):
80 try:
81 Rate(0)
82 assert False, "Should have raised"
83 except ValueError:
84 pass
85 try:
86 Rate(-1)
87 assert False, "Should have raised"
88 except ValueError:
89 pass
90
91 def test_avg_or_lifetime_avg_fallback(self):
92 r = Rate(60000)
93 r.add_data(100)
94 # No coalesce, so last is empty — should fall back to lifetime
95 avg = r.get_avg_or_lifetime_avg()
96 assert avg == 100.0
97
98 def test_lifetime_periods(self):
99 r = Rate(100)
100 time.sleep(0.25)
101 assert r.get_lifetime_periods() >= 2
102
103 def test_saturation(self):
104 r = Rate(100)
105 r.add_data(10, 50) # 50ms event in 100ms period
106 time.sleep(0.15)
107 r.coalesce()
108 sat = r.get_last_event_saturation()
109 assert 0 < sat <= 1.0
110
111
112class TestRateSummaryListener:
113 def test_listener_called_on_coalesce(self):
114 r = Rate(100)
115 calls = []
116
117 class TestListener(RateSummaryListener):
118 def add(self, total_value, event_count, total_event_time, period):
119 calls.append((total_value, event_count, period))
120
121 r.set_summary_listener(TestListener())
122 r.add_data(42)
123 time.sleep(0.15)
124 r.coalesce()
125 assert len(calls) == 1
126 assert calls[0][2] == 100 # period