distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for ConnThrottler — TDD: tests before implementation."""
2
3import time
4
5import pytest
6
7from i2p_apps.i2ptunnel.throttler import ConnThrottler
8
9
10class TestConnThrottler:
11 def test_allow_under_limit(self):
12 t = ConnThrottler(max_per_peer=5, max_total=100, window_seconds=60, ban_seconds=300)
13 t.record("peer1")
14 assert t.allow("peer1") is True
15
16 def test_reject_at_per_peer_limit(self):
17 t = ConnThrottler(max_per_peer=3, max_total=100, window_seconds=60, ban_seconds=300)
18 for _ in range(3):
19 t.record("peer1")
20 assert t.allow("peer1") is False
21
22 def test_reject_at_global_limit(self):
23 t = ConnThrottler(max_per_peer=100, max_total=3, window_seconds=60, ban_seconds=300)
24 for i in range(3):
25 t.record(f"peer{i}")
26 assert t.allow("peer99") is False
27
28 def test_different_peers_independent(self):
29 t = ConnThrottler(max_per_peer=2, max_total=100, window_seconds=60, ban_seconds=300)
30 t.record("peer1")
31 t.record("peer1")
32 assert t.allow("peer1") is False
33 assert t.allow("peer2") is True # different peer, still under limit
34
35 def test_ban_expiry(self):
36 t = ConnThrottler(max_per_peer=2, max_total=100, window_seconds=60, ban_seconds=10)
37 t.record("peer1")
38 t.record("peer1")
39 # At limit, peer is banned
40 assert t.allow("peer1") is False
41 # Force ban to be expired
42 t._ban_expiry["peer1"] = time.monotonic() - 1
43 # After ban expires, records are cleared — peer can connect again
44 assert t.allow("peer1") is True
45
46 def test_window_expiry(self):
47 t = ConnThrottler(max_per_peer=1, max_total=100, window_seconds=1, ban_seconds=300)
48 t.record("peer1")
49 # At limit, peer is banned
50 assert t.allow("peer1") is False
51 # Simulate both ban and window passing
52 t._ban_expiry["peer1"] = time.monotonic() - 1
53 t._peer_records["peer1"].clear()
54 t._global_records.clear()
55 assert t.allow("peer1") is True
56
57 def test_allow_no_records(self):
58 t = ConnThrottler(max_per_peer=5, max_total=100, window_seconds=60, ban_seconds=300)
59 assert t.allow("newpeer") is True
60
61 def test_zero_limits_means_unlimited(self):
62 t = ConnThrottler(max_per_peer=0, max_total=0, window_seconds=60, ban_seconds=300)
63 for _ in range(100):
64 t.record("peer1")
65 assert t.allow("peer1") is True