distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for RouterConfig — configuration loading, validation, and env overrides."""
2
3import os
4import pytest
5
6from i2p_router.config import RouterConfig
7
8
9class TestDefaultConfig:
10 """Default config values are sensible."""
11
12 def test_default_router_name(self):
13 cfg = RouterConfig()
14 assert cfg.router_name == "i2p-python-router"
15
16 def test_default_listen_host(self):
17 cfg = RouterConfig()
18 assert cfg.listen_host == "0.0.0.0"
19
20 def test_default_listen_port(self):
21 cfg = RouterConfig()
22 assert cfg.listen_port == 9700
23
24 def test_default_tunnel_counts(self):
25 cfg = RouterConfig()
26 assert cfg.inbound_tunnel_count == 3
27 assert cfg.outbound_tunnel_count == 3
28
29 def test_default_tunnel_length(self):
30 cfg = RouterConfig()
31 assert cfg.tunnel_length == 3
32
33 def test_default_tunnel_lifetime(self):
34 cfg = RouterConfig()
35 assert cfg.tunnel_lifetime_seconds == 600
36
37 def test_default_bandwidth_unlimited(self):
38 cfg = RouterConfig()
39 assert cfg.bandwidth_limit_kbps == 0
40
41 def test_default_floodfill_off(self):
42 cfg = RouterConfig()
43 assert cfg.floodfill is False
44
45 def test_default_reseed_urls_empty(self):
46 cfg = RouterConfig()
47 assert cfg.reseed_urls == []
48
49 def test_default_timeouts(self):
50 cfg = RouterConfig()
51 assert cfg.handshake_timeout_seconds == 30
52 assert cfg.idle_timeout_seconds == 300
53
54 def test_default_data_dir(self):
55 cfg = RouterConfig()
56 assert cfg.data_dir == "~/.i2p-python"
57
58
59class TestFromDict:
60 """Config loading from dict."""
61
62 def test_from_dict_overrides_defaults(self):
63 cfg = RouterConfig.from_dict({
64 "router_name": "my-router",
65 "listen_port": 8000,
66 "floodfill": True,
67 })
68 assert cfg.router_name == "my-router"
69 assert cfg.listen_port == 8000
70 assert cfg.floodfill is True
71 # Non-overridden fields keep defaults
72 assert cfg.listen_host == "0.0.0.0"
73 assert cfg.tunnel_length == 3
74
75 def test_from_dict_empty_gives_defaults(self):
76 cfg = RouterConfig.from_dict({})
77 default = RouterConfig()
78 assert cfg.router_name == default.router_name
79 assert cfg.listen_port == default.listen_port
80
81 def test_from_dict_all_fields(self):
82 d = {
83 "router_name": "test",
84 "listen_host": "127.0.0.1",
85 "listen_port": 7777,
86 "inbound_tunnel_count": 5,
87 "outbound_tunnel_count": 5,
88 "tunnel_length": 2,
89 "tunnel_lifetime_seconds": 300,
90 "bandwidth_limit_kbps": 1024,
91 "floodfill": True,
92 "reseed_urls": ["https://example.com/reseed"],
93 "handshake_timeout_seconds": 10,
94 "idle_timeout_seconds": 120,
95 "data_dir": "/tmp/i2p-test",
96 }
97 cfg = RouterConfig.from_dict(d)
98 for key, val in d.items():
99 assert getattr(cfg, key) == val
100
101 def test_from_dict_ignores_unknown_keys(self):
102 cfg = RouterConfig.from_dict({"unknown_key": "value"})
103 assert cfg.router_name == "i2p-python-router"
104
105
106class TestValidation:
107 """Invalid config raises ValueError."""
108
109 def test_negative_port_raises(self):
110 cfg = RouterConfig(listen_port=-1)
111 with pytest.raises(ValueError, match="listen_port"):
112 cfg.validate()
113
114 def test_port_too_high_raises(self):
115 cfg = RouterConfig(listen_port=70000)
116 with pytest.raises(ValueError, match="listen_port"):
117 cfg.validate()
118
119 def test_zero_port_is_valid(self):
120 cfg = RouterConfig(listen_port=0)
121 cfg.validate() # Should not raise
122
123 def test_max_port_is_valid(self):
124 cfg = RouterConfig(listen_port=65535)
125 cfg.validate() # Should not raise
126
127 def test_negative_inbound_tunnel_count_raises(self):
128 cfg = RouterConfig(inbound_tunnel_count=-1)
129 with pytest.raises(ValueError, match="inbound_tunnel_count"):
130 cfg.validate()
131
132 def test_negative_outbound_tunnel_count_raises(self):
133 cfg = RouterConfig(outbound_tunnel_count=-1)
134 with pytest.raises(ValueError, match="outbound_tunnel_count"):
135 cfg.validate()
136
137 def test_negative_tunnel_length_raises(self):
138 cfg = RouterConfig(tunnel_length=-1)
139 with pytest.raises(ValueError, match="tunnel_length"):
140 cfg.validate()
141
142 def test_negative_tunnel_lifetime_raises(self):
143 cfg = RouterConfig(tunnel_lifetime_seconds=-1)
144 with pytest.raises(ValueError, match="tunnel_lifetime_seconds"):
145 cfg.validate()
146
147 def test_negative_bandwidth_raises(self):
148 cfg = RouterConfig(bandwidth_limit_kbps=-1)
149 with pytest.raises(ValueError, match="bandwidth_limit_kbps"):
150 cfg.validate()
151
152 def test_negative_handshake_timeout_raises(self):
153 cfg = RouterConfig(handshake_timeout_seconds=-1)
154 with pytest.raises(ValueError, match="handshake_timeout_seconds"):
155 cfg.validate()
156
157 def test_negative_idle_timeout_raises(self):
158 cfg = RouterConfig(idle_timeout_seconds=-1)
159 with pytest.raises(ValueError, match="idle_timeout_seconds"):
160 cfg.validate()
161
162 def test_valid_config_does_not_raise(self):
163 cfg = RouterConfig()
164 cfg.validate() # Should not raise
165
166
167class TestEnvOverrides:
168 """Environment variable overrides."""
169
170 def test_env_router_name(self, monkeypatch):
171 monkeypatch.setenv("I2P_ROUTER_NAME", "env-router")
172 cfg = RouterConfig.with_env_overrides()
173 assert cfg.router_name == "env-router"
174
175 def test_env_listen_host(self, monkeypatch):
176 monkeypatch.setenv("I2P_LISTEN_HOST", "127.0.0.1")
177 cfg = RouterConfig.with_env_overrides()
178 assert cfg.listen_host == "127.0.0.1"
179
180 def test_env_listen_port(self, monkeypatch):
181 monkeypatch.setenv("I2P_LISTEN_PORT", "8080")
182 cfg = RouterConfig.with_env_overrides()
183 assert cfg.listen_port == 8080
184
185 def test_env_tunnel_length(self, monkeypatch):
186 monkeypatch.setenv("I2P_TUNNEL_LENGTH", "5")
187 cfg = RouterConfig.with_env_overrides()
188 assert cfg.tunnel_length == 5
189
190 def test_env_floodfill(self, monkeypatch):
191 monkeypatch.setenv("I2P_FLOODFILL", "true")
192 cfg = RouterConfig.with_env_overrides()
193 assert cfg.floodfill is True
194
195 def test_env_floodfill_false(self, monkeypatch):
196 monkeypatch.setenv("I2P_FLOODFILL", "false")
197 cfg = RouterConfig.with_env_overrides()
198 assert cfg.floodfill is False
199
200 def test_env_data_dir(self, monkeypatch):
201 monkeypatch.setenv("I2P_DATA_DIR", "/opt/i2p")
202 cfg = RouterConfig.with_env_overrides()
203 assert cfg.data_dir == "/opt/i2p"
204
205 def test_env_overrides_with_base_config(self, monkeypatch):
206 monkeypatch.setenv("I2P_LISTEN_PORT", "7777")
207 base = RouterConfig(router_name="base-router", listen_port=5000)
208 cfg = RouterConfig.with_env_overrides(base)
209 # Env override wins over base
210 assert cfg.listen_port == 7777
211 # Non-overridden base value preserved
212 assert cfg.router_name == "base-router"
213
214 def test_env_bandwidth_limit(self, monkeypatch):
215 monkeypatch.setenv("I2P_BANDWIDTH_LIMIT_KBPS", "512")
216 cfg = RouterConfig.with_env_overrides()
217 assert cfg.bandwidth_limit_kbps == 512
218
219 def test_no_env_vars_gives_defaults(self):
220 # Ensure no I2P_ env vars are set for this test
221 cfg = RouterConfig.with_env_overrides()
222 default = RouterConfig()
223 assert cfg.router_name == default.router_name
224 assert cfg.listen_port == default.listen_port