distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for TunnelControllerGroup lifecycle."""
2
3import asyncio
4from pathlib import Path
5from unittest.mock import AsyncMock, MagicMock, patch
6
7import pytest
8
9from i2p_apps.i2ptunnel.group import TunnelControllerGroup
10
11
12class TestGroupLifecycle:
13 @pytest.mark.asyncio
14 async def test_load_empty_config(self, tmp_path):
15 config = tmp_path / "empty.config"
16 config.write_text("")
17 group = TunnelControllerGroup(config)
18 await group.load_and_start()
19 assert len(group.controllers) == 0
20
21 @pytest.mark.asyncio
22 async def test_load_tunnels_from_config(self, tmp_path):
23 config = tmp_path / "test.config"
24 config.write_text(
25 "tunnel.0.name=proxy\n"
26 "tunnel.0.type=httpclient\n"
27 "tunnel.0.listenPort=4444\n"
28 "tunnel.0.interface=127.0.0.1\n"
29 "tunnel.0.startOnLoad=false\n"
30 )
31 group = TunnelControllerGroup(config)
32 await group.load_and_start()
33 assert len(group.controllers) == 1
34 assert group.controllers[0].config.name == "proxy"
35
36 def test_get_controller_by_name(self, tmp_path):
37 config = tmp_path / "test.config"
38 config.write_text(
39 "tunnel.0.name=myproxy\n"
40 "tunnel.0.type=httpclient\n"
41 "tunnel.0.listenPort=4444\n"
42 "tunnel.0.interface=127.0.0.1\n"
43 )
44 group = TunnelControllerGroup(config)
45 # Manually load without async start
46 from i2p_apps.i2ptunnel.config import TunnelConfigParser
47 from i2p_apps.i2ptunnel.controller import TunnelController
48 from i2p_apps.i2ptunnel.sam_session import SessionFactory
49
50 tunnels = TunnelConfigParser.load(config)
51 for td in tunnels:
52 group._controllers.append(
53 TunnelController(td, SessionFactory("127.0.0.1", 7656))
54 )
55
56 assert group.get_controller("myproxy") is not None
57 assert group.get_controller("nonexistent") is None
58
59 def test_list_controllers(self, tmp_path):
60 config = tmp_path / "test.config"
61 config.write_text("")
62 group = TunnelControllerGroup(config)
63 assert group.list_controllers() == []
64
65 @pytest.mark.asyncio
66 async def test_stop_all_empty(self, tmp_path):
67 config = tmp_path / "test.config"
68 config.write_text("")
69 group = TunnelControllerGroup(config)
70 await group.load_and_start()
71 await group.stop_all() # no crash
72
73 @pytest.mark.asyncio
74 async def test_start_all_skips_running(self, tmp_path):
75 config = tmp_path / "test.config"
76 config.write_text("")
77 group = TunnelControllerGroup(config)
78 await group.load_and_start()
79 await group.start_all() # no crash on empty
80
81 @pytest.mark.asyncio
82 async def test_config_d_loading(self, tmp_path):
83 """Loads tunnel configs from config.d/ directory."""
84 config = tmp_path / "i2ptunnel.config"
85 config.write_text(
86 "tunnel.0.name=main\n"
87 "tunnel.0.type=httpclient\n"
88 "tunnel.0.listenPort=4444\n"
89 "tunnel.0.interface=127.0.0.1\n"
90 )
91 config_d = tmp_path / "i2ptunnel.config.d"
92 config_d.mkdir()
93 (config_d / "extra.config").write_text(
94 "tunnel.0.name=extra\n"
95 "tunnel.0.type=server\n"
96 "tunnel.0.targetHost=127.0.0.1\n"
97 "tunnel.0.targetPort=8080\n"
98 "tunnel.0.interface=127.0.0.1\n"
99 )
100 group = TunnelControllerGroup(config)
101 await group.load_and_start()
102 names = [c.config.name for c in group.controllers]
103 assert "main" in names
104 assert "extra" in names