distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for FloodfillManager — tracking floodfill routers for NetDB lookups."""
2
3import os
4
5import pytest
6
7from i2p_netdb.floodfill import FloodfillManager
8
9
10class TestFloodfillAdd:
11 def test_add_floodfill_and_retrieve(self):
12 mgr = FloodfillManager()
13 peer_hash = os.urandom(32)
14 info = {"caps": "f", "host": "127.0.0.1"}
15 mgr.add_floodfill(peer_hash, info)
16 assert mgr.is_floodfill(peer_hash)
17 assert peer_hash in mgr.get_all()
18
19 def test_add_multiple_floodfills(self):
20 mgr = FloodfillManager()
21 hashes = [os.urandom(32) for _ in range(5)]
22 for h in hashes:
23 mgr.add_floodfill(h, {"caps": "f"})
24 assert mgr.count == 5
25 for h in hashes:
26 assert mgr.is_floodfill(h)
27
28 def test_add_duplicate_overwrites(self):
29 mgr = FloodfillManager()
30 peer_hash = os.urandom(32)
31 mgr.add_floodfill(peer_hash, {"version": "1"})
32 mgr.add_floodfill(peer_hash, {"version": "2"})
33 assert mgr.count == 1
34
35
36class TestFloodfillRemove:
37 def test_remove_floodfill(self):
38 mgr = FloodfillManager()
39 peer_hash = os.urandom(32)
40 mgr.add_floodfill(peer_hash, {"caps": "f"})
41 mgr.remove_floodfill(peer_hash)
42 assert not mgr.is_floodfill(peer_hash)
43 assert mgr.count == 0
44
45 def test_remove_nonexistent_is_noop(self):
46 mgr = FloodfillManager()
47 mgr.remove_floodfill(os.urandom(32)) # should not raise
48 assert mgr.count == 0
49
50
51class TestFloodfillSelect:
52 def test_select_returns_one_from_available(self):
53 mgr = FloodfillManager()
54 hashes = [os.urandom(32) for _ in range(3)]
55 for h in hashes:
56 mgr.add_floodfill(h, {"caps": "f"})
57 selected = mgr.select_floodfill()
58 assert selected in hashes
59
60 def test_select_with_exclude_set(self):
61 mgr = FloodfillManager()
62 h1 = os.urandom(32)
63 h2 = os.urandom(32)
64 h3 = os.urandom(32)
65 mgr.add_floodfill(h1, {})
66 mgr.add_floodfill(h2, {})
67 mgr.add_floodfill(h3, {})
68 selected = mgr.select_floodfill(exclude={h1, h2})
69 assert selected == h3
70
71 def test_select_all_excluded_returns_none(self):
72 mgr = FloodfillManager()
73 h1 = os.urandom(32)
74 mgr.add_floodfill(h1, {})
75 selected = mgr.select_floodfill(exclude={h1})
76 assert selected is None
77
78 def test_select_from_empty_returns_none(self):
79 mgr = FloodfillManager()
80 assert mgr.select_floodfill() is None
81
82
83class TestFloodfillNonFloodfill:
84 def test_non_floodfill_not_returned(self):
85 mgr = FloodfillManager()
86 ff_hash = os.urandom(32)
87 non_ff_hash = os.urandom(32)
88 mgr.add_floodfill(ff_hash, {"caps": "f"})
89 # non_ff_hash was never added
90 assert not mgr.is_floodfill(non_ff_hash)
91 assert non_ff_hash not in mgr.get_all()
92
93
94class TestFloodfillCount:
95 def test_count_empty(self):
96 mgr = FloodfillManager()
97 assert mgr.count == 0
98
99 def test_count_after_adds_and_removes(self):
100 mgr = FloodfillManager()
101 hashes = [os.urandom(32) for _ in range(4)]
102 for h in hashes:
103 mgr.add_floodfill(h, {})
104 assert mgr.count == 4
105 mgr.remove_floodfill(hashes[0])
106 mgr.remove_floodfill(hashes[1])
107 assert mgr.count == 2
108
109 def test_get_all_returns_list(self):
110 mgr = FloodfillManager()
111 hashes = [os.urandom(32) for _ in range(3)]
112 for h in hashes:
113 mgr.add_floodfill(h, {})
114 all_ffs = mgr.get_all()
115 assert isinstance(all_ffs, list)
116 assert len(all_ffs) == 3
117 for h in hashes:
118 assert h in all_ffs