distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for I2PTunnel config parser — TDD: tests before implementation."""
2
3from pathlib import Path
4
5import pytest
6
7from i2p_apps.i2ptunnel.config import TunnelConfigParser, TunnelDefinition, TunnelType
8
9
10class TestTunnelType:
11 """TunnelType enum covers all 12 Java I2P tunnel types."""
12
13 def test_all_types_present(self):
14 expected = {
15 "httpclient", "connectclient", "client", "server",
16 "httpserver", "ircclient", "ircserver", "sockstunnel",
17 "socksirctunnel", "httpbidirserver", "streamrclient",
18 "streamrserver",
19 }
20 actual = {t.value for t in TunnelType}
21 assert actual == expected
22
23 def test_from_string(self):
24 assert TunnelType("httpclient") == TunnelType.HTTPCLIENT
25 assert TunnelType("server") == TunnelType.SERVER
26
27
28class TestTunnelDefinition:
29 """TunnelDefinition dataclass."""
30
31 def test_defaults(self):
32 td = TunnelDefinition(
33 name="test",
34 type=TunnelType.CLIENT,
35 listen_port=1234,
36 )
37 assert td.name == "test"
38 assert td.interface == "127.0.0.1"
39 assert td.start_on_load is False
40 assert td.shared_client is False
41 assert td.options == {}
42 assert td.proxy_list == []
43
44 def test_is_client(self):
45 td = TunnelDefinition(name="t", type=TunnelType.HTTPCLIENT, listen_port=4444)
46 assert td.is_client is True
47
48 def test_is_server(self):
49 td = TunnelDefinition(name="t", type=TunnelType.SERVER, listen_port=0)
50 assert td.is_server is True
51
52
53class TestConfigParserLoad:
54 """Parse i2ptunnel.config Java Properties format."""
55
56 def test_load_sample_config(self, sample_config_path: Path):
57 tunnels = TunnelConfigParser.load(sample_config_path)
58 assert len(tunnels) == 7
59
60 def test_http_proxy_parsed(self, sample_config_path: Path):
61 tunnels = TunnelConfigParser.load(sample_config_path)
62 http = tunnels[0]
63 assert http.name == "I2P HTTP Proxy"
64 assert http.type == TunnelType.HTTPCLIENT
65 assert http.interface == "127.0.0.1"
66 assert http.listen_port == 4444
67 assert http.shared_client is True
68 assert http.start_on_load is True
69 assert http.proxy_list == ["false.i2p"]
70
71 def test_options_extracted(self, sample_config_path: Path):
72 tunnels = TunnelConfigParser.load(sample_config_path)
73 http = tunnels[0]
74 assert http.options["inbound.length"] == "3"
75 assert http.options["outbound.length"] == "3"
76
77 def test_server_tunnel_parsed(self, sample_config_path: Path):
78 tunnels = TunnelConfigParser.load(sample_config_path)
79 server = tunnels[3]
80 assert server.name == "I2P Webserver"
81 assert server.type == TunnelType.HTTPSERVER
82 assert server.target_host == "127.0.0.1"
83 assert server.target_port == 7658
84 assert server.spoofed_host == "mysite.i2p"
85 assert server.priv_key_file == "eepsite/eepPriv.dat"
86
87 def test_client_tunnel_destination(self, sample_config_path: Path):
88 tunnels = TunnelConfigParser.load(sample_config_path)
89 smtp = tunnels[4]
90 assert smtp.type == TunnelType.CLIENT
91 assert smtp.target_destination == "smtp.postman.i2p"
92 assert smtp.listen_port == 7659
93
94 def test_start_on_load_false(self, sample_config_path: Path):
95 tunnels = TunnelConfigParser.load(sample_config_path)
96 irc = tunnels[2]
97 assert irc.start_on_load is False
98
99 def test_missing_properties_use_defaults(self, tmp_path: Path):
100 p = tmp_path / "minimal.config"
101 p.write_text("tunnel.0.name=Bare\ntunnel.0.type=client\n")
102 tunnels = TunnelConfigParser.load(p)
103 assert len(tunnels) == 1
104 assert tunnels[0].listen_port == 0
105 assert tunnels[0].interface == "127.0.0.1"
106
107 def test_comments_and_blank_lines_skipped(self, tmp_path: Path):
108 p = tmp_path / "comments.config"
109 p.write_text("# comment line\n\ntunnel.0.name=T\ntunnel.0.type=client\n")
110 tunnels = TunnelConfigParser.load(p)
111 assert len(tunnels) == 1
112
113 def test_nonexistent_file_returns_empty(self, tmp_path: Path):
114 tunnels = TunnelConfigParser.load(tmp_path / "nofile.config")
115 assert tunnels == []
116
117
118class TestConfigParserLoadDir:
119 """Parse i2ptunnel.config.d/ directory."""
120
121 def test_load_dir(self, sample_config_dir: Path):
122 tunnels = TunnelConfigParser.load_dir(sample_config_dir)
123 assert len(tunnels) == 2
124 names = {t.name for t in tunnels}
125 assert "HTTP Proxy" in names
126 assert "My Server" in names
127
128 def test_load_dir_nonexistent(self, tmp_path: Path):
129 tunnels = TunnelConfigParser.load_dir(tmp_path / "nodir")
130 assert tunnels == []
131
132
133class TestConfigParserSave:
134 """Round-trip: load -> save -> load produces identical result."""
135
136 def test_roundtrip(self, sample_config_path: Path, tmp_path: Path):
137 original = TunnelConfigParser.load(sample_config_path)
138 out_path = tmp_path / "output.config"
139 TunnelConfigParser.save(out_path, original)
140 reloaded = TunnelConfigParser.load(out_path)
141 assert len(reloaded) == len(original)
142 for orig, rel in zip(original, reloaded):
143 assert orig.name == rel.name
144 assert orig.type == rel.type
145 assert orig.listen_port == rel.listen_port
146 assert orig.interface == rel.interface
147 assert orig.start_on_load == rel.start_on_load
148 assert orig.options == rel.options
149
150 def test_save_empty(self, tmp_path: Path):
151 out_path = tmp_path / "empty.config"
152 TunnelConfigParser.save(out_path, [])
153 content = out_path.read_text()
154 assert content.startswith("#") # header comment