distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for NetDB throttle components."""
2
3from i2p_netdb.throttle import NegativeLookupCache, FloodThrottler, LookupThrottler
4
5
6def test_negative_cache_after_max_fails():
7 cache = NegativeLookupCache()
8 key = b"\x01" * 32
9 assert not cache.should_skip(key)
10 for _ in range(3):
11 cache.lookup_failed(key)
12 assert cache.should_skip(key)
13
14
15def test_negative_cache_clear_resets():
16 cache = NegativeLookupCache()
17 key = b"\x02" * 32
18 for _ in range(3):
19 cache.lookup_failed(key)
20 cache.clear()
21 assert not cache.should_skip(key)
22
23
24def test_bad_destination_permanent():
25 cache = NegativeLookupCache()
26 key = b"\x03" * 32
27 cache.add_bad_destination(key)
28 assert cache.is_bad_destination(key)
29 cache.clear()
30 assert cache.is_bad_destination(key) # survives clear
31
32
33def test_flood_throttle_limit():
34 ft = FloodThrottler()
35 peer = b"\x04" * 32
36 for _ in range(3):
37 assert not ft.should_throttle(peer)
38 assert ft.should_throttle(peer)
39
40
41def test_flood_throttle_clear():
42 ft = FloodThrottler()
43 peer = b"\x05" * 32
44 for _ in range(4):
45 ft.should_throttle(peer)
46 ft.clear()
47 assert not ft.should_throttle(peer)
48
49
50def test_lookup_throttle_duplicate():
51 lt = LookupThrottler()
52 key = b"\x06" * 32
53 gw = b"\x07" * 32
54 assert not lt.should_throttle(key, 100, gw)
55 assert lt.should_throttle(key, 100, gw) # duplicate
56
57
58def test_lookup_throttle_different_tunnel():
59 lt = LookupThrottler()
60 key = b"\x08" * 32
61 gw = b"\x09" * 32
62 assert not lt.should_throttle(key, 100, gw)
63 assert not lt.should_throttle(key, 200, gw) # different tunnel