distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for Frequency — rolling average event frequency tracker."""
2
3import time
4from unittest.mock import patch
5
6import pytest
7
8from i2p_stat.frequency import Frequency
9
10
11class TestFrequencyInit:
12 def test_valid_period(self):
13 f = Frequency(1000)
14 assert f.period == 1000
15
16 def test_zero_period_raises(self):
17 with pytest.raises(ValueError, match="positive"):
18 Frequency(0)
19
20 def test_negative_period_raises(self):
21 with pytest.raises(ValueError, match="positive"):
22 Frequency(-1)
23
24
25class TestFrequencyBasics:
26 def test_initial_avg_interval(self):
27 f = Frequency(1000)
28 # Initially period + 1 (no events)
29 assert f.get_average_interval() == 1001.0
30
31 def test_initial_event_count(self):
32 f = Frequency(1000)
33 assert f.get_event_count() == 0
34
35 def test_initial_events_per_period(self):
36 f = Frequency(1000)
37 # period / avg_interval = 1000 / 1001 < 1
38 assert f.get_average_events_per_period() < 1.0
39
40 def test_initial_strict_avg(self):
41 f = Frequency(1000)
42 # No events → period + 1
43 assert f.get_strict_average_interval() == 1001.0
44
45
46class TestFrequencyEvents:
47 def test_event_increments_count(self):
48 f = Frequency(10000)
49 f.event_occurred()
50 assert f.get_event_count() == 1
51 f.event_occurred()
52 assert f.get_event_count() == 2
53
54 def test_rapid_events_decrease_interval(self):
55 f = Frequency(10000)
56 initial = f.get_average_interval()
57 # Fire events with small delay so interval < period
58 for _ in range(5):
59 time.sleep(0.002)
60 f.event_occurred()
61 after = f.get_average_interval()
62 # Rapid events should decrease the average interval
63 assert after < initial
64
65 def test_min_avg_interval_tracks_minimum(self):
66 f = Frequency(10000)
67 # Fire events with small delay
68 for _ in range(10):
69 time.sleep(0.002)
70 f.event_occurred()
71 min_after_rapid = f.get_min_average_interval()
72 # min should be less than initial (period + 1)
73 assert min_after_rapid < 10001.0
74
75 def test_max_events_per_period(self):
76 f = Frequency(10000)
77 for _ in range(5):
78 time.sleep(0.002)
79 f.event_occurred()
80 max_epp = f.get_max_average_events_per_period()
81 assert max_epp > 0
82
83 def test_strict_average(self):
84 f = Frequency(10000)
85 time.sleep(0.002)
86 f.event_occurred()
87 f.event_occurred()
88 f.event_occurred()
89 strict = f.get_strict_average_interval()
90 # Should be (now - start) / 3, a small positive value
91 assert strict > 0
92
93 def test_recalculate_no_events(self):
94 f = Frequency(1) # 1ms period
95 # Sleep >1ms so interval exceeds period
96 time.sleep(0.01)
97 f.recalculate()
98 # Should reset to period + 1
99 assert f.get_average_interval() == 2.0
100
101
102class TestFrequencyEdgeCases:
103 def test_events_per_period_zero_avg(self):
104 """If avg_interval is somehow 0, return 0."""
105 f = Frequency(1000)
106 f._avg_interval = 0.0
107 assert f.get_average_events_per_period() == 0.0
108
109 def test_max_events_zero_min(self):
110 f = Frequency(1000)
111 f._min_avg_interval = 0.0
112 assert f.get_max_average_events_per_period() == 0.0