distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for RouterBootstrap — bootstrap orchestrator and main event loop.
2
3TDD: tests written before implementation.
4"""
5
6import asyncio
7import hashlib
8import os
9import tempfile
10import unittest
11from unittest.mock import AsyncMock, MagicMock, patch, PropertyMock
12
13from i2p_router.config import RouterConfig
14
15
16class TestRouterBootstrapInit(unittest.TestCase):
17 """Test RouterBootstrap initialization and configuration."""
18
19 def test_create_with_defaults(self):
20 from i2p_router.bootstrap import RouterBootstrap
21
22 config = RouterConfig(
23 listen_host="127.0.0.1",
24 listen_port=0,
25 data_dir=tempfile.mkdtemp(),
26 )
27 bootstrap = RouterBootstrap(config)
28 assert bootstrap.config is config
29 assert bootstrap.state == "stopped"
30 assert bootstrap.peer_count == 0
31
32 def test_create_with_custom_config(self):
33 from i2p_router.bootstrap import RouterBootstrap
34
35 config = RouterConfig(
36 listen_host="0.0.0.0",
37 listen_port=12345,
38 data_dir=tempfile.mkdtemp(),
39 bandwidth_limit_kbps=50,
40 )
41 bootstrap = RouterBootstrap(config)
42 assert bootstrap.config.listen_port == 12345
43 assert bootstrap.config.bandwidth_limit_kbps == 50
44
45
46class TestIdentityManagement(unittest.TestCase):
47 """Test identity load/generate during bootstrap."""
48
49 def test_generates_identity_if_none_exists(self):
50 from i2p_router.bootstrap import RouterBootstrap
51
52 data_dir = tempfile.mkdtemp()
53 config = RouterConfig(
54 listen_host="127.0.0.1",
55 listen_port=0,
56 data_dir=data_dir,
57 )
58 bootstrap = RouterBootstrap(config)
59
60 def run():
61 return asyncio.run(bootstrap._load_or_generate_identity())
62
63 run()
64
65 assert bootstrap._key_bundle is not None
66 assert len(bootstrap._key_bundle.signing_public) == 32
67 assert len(bootstrap._key_bundle.ntcp2_public) == 32
68 # Keys should be persisted
69 key_path = os.path.join(data_dir, "router.keys.json")
70 assert os.path.exists(key_path)
71
72 def test_loads_existing_identity(self):
73 from i2p_router.bootstrap import RouterBootstrap
74 from i2p_router.identity import RouterKeyBundle
75
76 data_dir = tempfile.mkdtemp()
77 # Pre-generate and save keys
78 bundle = RouterKeyBundle.generate()
79 key_path = os.path.join(data_dir, "router.keys.json")
80 bundle.save(key_path)
81
82 config = RouterConfig(
83 listen_host="127.0.0.1",
84 listen_port=0,
85 data_dir=data_dir,
86 )
87 bootstrap = RouterBootstrap(config)
88 asyncio.run(bootstrap._load_or_generate_identity())
89
90 # Should load the same keys
91 assert bootstrap._key_bundle.signing_public == bundle.signing_public
92 assert bootstrap._key_bundle.ntcp2_public == bundle.ntcp2_public
93
94
95class TestReseedIntegration(unittest.TestCase):
96 """Test reseed triggering during bootstrap."""
97
98 def test_reseed_triggered_when_netdb_empty(self):
99 from i2p_router.bootstrap import RouterBootstrap
100
101 config = RouterConfig(
102 listen_host="127.0.0.1",
103 listen_port=0,
104 data_dir=tempfile.mkdtemp(),
105 )
106 bootstrap = RouterBootstrap(config)
107
108 mock_reseed = AsyncMock(return_value=[os.urandom(256) for _ in range(5)])
109
110 async def run():
111 await bootstrap._load_or_generate_identity()
112 with patch.object(bootstrap._reseed_client, "reseed", mock_reseed):
113 await bootstrap._reseed_if_needed()
114
115 asyncio.run(run())
116 # The reseed client was called (even though random bytes aren't valid RI)
117 mock_reseed.assert_called_once()
118
119 def test_reseed_skipped_when_enough_peers(self):
120 from i2p_router.bootstrap import RouterBootstrap
121
122 config = RouterConfig(
123 listen_host="127.0.0.1",
124 listen_port=0,
125 data_dir=tempfile.mkdtemp(),
126 )
127 bootstrap = RouterBootstrap(config)
128
129 async def run():
130 await bootstrap._load_or_generate_identity()
131 # Simulate having enough entries in the datastore
132 for i in range(60):
133 key = hashlib.sha256(f"peer{i}".encode()).digest()
134 bootstrap._context.datastore.put(
135 MagicMock(key=key, data=os.urandom(100))
136 )
137 return await bootstrap._reseed_if_needed()
138
139 result = asyncio.run(run())
140 assert result == 0 # No reseed needed
141
142
143class TestPeerConnectionFlow(unittest.TestCase):
144 """Test peer connection logic."""
145
146 def test_connection_pool_tracks_peers(self):
147 from i2p_router.peer_connector import ConnectionPool
148
149 pool = ConnectionPool(max_connections=3)
150 h1 = os.urandom(32)
151 h2 = os.urandom(32)
152
153 assert pool.add(h1, MagicMock()) is True
154 assert pool.add(h2, MagicMock()) is True
155 assert pool.active_count == 2
156 assert pool.is_connected(h1)
157 assert pool.is_connected(h2)
158
159 def test_connection_pool_rejects_over_max(self):
160 from i2p_router.peer_connector import ConnectionPool
161
162 pool = ConnectionPool(max_connections=1)
163 h1 = os.urandom(32)
164 h2 = os.urandom(32)
165
166 assert pool.add(h1, MagicMock()) is True
167 assert pool.add(h2, MagicMock()) is False
168 assert pool.active_count == 1
169
170
171class TestBootstrapGetStatus(unittest.TestCase):
172 """Test status reporting during bootstrap."""
173
174 def test_status_before_start(self):
175 from i2p_router.bootstrap import RouterBootstrap
176
177 config = RouterConfig(
178 listen_host="127.0.0.1",
179 listen_port=0,
180 data_dir=tempfile.mkdtemp(),
181 )
182 bootstrap = RouterBootstrap(config)
183 status = bootstrap.get_status()
184
185 assert status["state"] == "stopped"
186 assert status["peer_count"] == 0
187 assert status["netdb_routerinfos"] == 0
188
189 def test_status_after_identity_load(self):
190 from i2p_router.bootstrap import RouterBootstrap
191
192 config = RouterConfig(
193 listen_host="127.0.0.1",
194 listen_port=0,
195 data_dir=tempfile.mkdtemp(),
196 )
197 bootstrap = RouterBootstrap(config)
198 asyncio.run(bootstrap._load_or_generate_identity())
199
200 status = bootstrap.get_status()
201 assert "router_hash" in status
202 assert len(status["router_hash"]) > 0
203
204
205class TestBootstrapShutdown(unittest.TestCase):
206 """Test graceful shutdown."""
207
208 def test_shutdown_from_stopped(self):
209 from i2p_router.bootstrap import RouterBootstrap
210
211 config = RouterConfig(
212 listen_host="127.0.0.1",
213 listen_port=0,
214 data_dir=tempfile.mkdtemp(),
215 )
216 bootstrap = RouterBootstrap(config)
217 # Should not raise
218 asyncio.run(bootstrap.shutdown())
219 assert bootstrap.state == "stopped"
220
221 def test_shutdown_closes_connections(self):
222 from i2p_router.bootstrap import RouterBootstrap
223
224 config = RouterConfig(
225 listen_host="127.0.0.1",
226 listen_port=0,
227 data_dir=tempfile.mkdtemp(),
228 )
229 bootstrap = RouterBootstrap(config)
230 # Simulate that bootstrap is running so shutdown() doesn't bail
231 bootstrap._state = "running"
232
233 # Add mock connections to pool
234 mock_conn = AsyncMock()
235 mock_conn.close = AsyncMock()
236 mock_conn.is_alive = MagicMock(return_value=True)
237 h1 = os.urandom(32)
238 bootstrap._conn_pool.add(h1, mock_conn)
239
240 asyncio.run(bootstrap.shutdown())
241 assert bootstrap._conn_pool.active_count == 0
242 assert bootstrap.state == "stopped"
243 mock_conn.close.assert_called_once()
244
245
246class TestMessageReader(unittest.TestCase):
247 """Test reading I2NP messages from NTCP2 connections."""
248
249 def test_message_reader_dispatches_to_context(self):
250 from i2p_router.bootstrap import MessageReader
251
252 ctx = MagicMock()
253 reader = MessageReader(ctx)
254
255 # Simulate an I2NP block with type=1 (DatabaseStore)
256 # Short header: type(1) + msg_id(4) + expiration(4) + size(2) + payload
257 import struct
258 payload = os.urandom(32)
259 i2np_data = struct.pack("!BIiH", 1, 12345, 0, len(payload)) + payload
260
261 reader.handle_i2np_block(i2np_data)
262 ctx.process_inbound.assert_called_once_with(1, payload)
263
264
265class TestDatabaseStoreExchange(unittest.TestCase):
266 """Test RouterInfo exchange via DatabaseStore messages."""
267
268 def test_build_database_store_message(self):
269 from i2p_router.bootstrap import build_database_store_i2np
270
271 router_hash = os.urandom(32)
272 ri_bytes = os.urandom(256)
273
274 msg = build_database_store_i2np(router_hash, ri_bytes)
275
276 # Should be I2NP short header + DatabaseStore payload
277 import struct
278 msg_type = struct.unpack("!B", msg[:1])[0]
279 assert msg_type == 1 # DATABASE_STORE
280
281
282if __name__ == "__main__":
283 unittest.main()