distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for I2PTunnel CLI."""
2
3import sys
4import tempfile
5from pathlib import Path
6from unittest.mock import patch, AsyncMock, MagicMock
7
8import pytest
9
10from i2p_apps.i2ptunnel.cli import _get_parser, _run_list, _run_add, main
11
12
13class TestParser:
14 def test_start_command(self):
15 parser = _get_parser()
16 args = parser.parse_args(["start", "--sam-port", "7777"])
17 assert args.command == "start"
18 assert args.sam_port == 7777
19
20 def test_list_command(self):
21 parser = _get_parser()
22 args = parser.parse_args(["list"])
23 assert args.command == "list"
24
25 def test_add_command(self):
26 parser = _get_parser()
27 args = parser.parse_args([
28 "add", "--name", "myproxy", "--type", "httpclient",
29 "--port", "4444", "--interface", "0.0.0.0", "--start-on-load"
30 ])
31 assert args.command == "add"
32 assert args.name == "myproxy"
33 assert args.tunnel_type == "httpclient"
34 assert args.port == 4444
35 assert args.interface == "0.0.0.0"
36 assert args.start_on_load is True
37
38 def test_add_defaults(self):
39 parser = _get_parser()
40 args = parser.parse_args(["add", "--name", "t", "--type", "server", "--port", "80"])
41 assert args.interface == "127.0.0.1"
42 assert args.target == ""
43 assert args.start_on_load is False
44
45 def test_no_command(self):
46 parser = _get_parser()
47 args = parser.parse_args([])
48 assert args.command is None
49
50 def test_start_with_config(self):
51 parser = _get_parser()
52 args = parser.parse_args(["start", "--config", "/tmp/my.config"])
53 assert args.config == Path("/tmp/my.config")
54
55
56class TestRunList:
57 def test_no_config_file(self, capsys):
58 parser = _get_parser()
59 args = parser.parse_args(["list", "--config", "/nonexistent/path"])
60 with pytest.raises(SystemExit):
61 _run_list(args)
62 captured = capsys.readouterr()
63 assert "not found" in captured.err
64
65 def test_lists_tunnels(self, tmp_path, capsys):
66 config = tmp_path / "test.config"
67 config.write_text(
68 "tunnel.0.name=myhttp\n"
69 "tunnel.0.type=httpclient\n"
70 "tunnel.0.listenPort=4444\n"
71 "tunnel.0.interface=127.0.0.1\n"
72 "tunnel.0.startOnLoad=true\n"
73 )
74 parser = _get_parser()
75 args = parser.parse_args(["list", "--config", str(config)])
76 _run_list(args)
77 captured = capsys.readouterr()
78 assert "myhttp" in captured.out
79
80 def test_empty_config(self, tmp_path, capsys):
81 config = tmp_path / "empty.config"
82 config.write_text("")
83 parser = _get_parser()
84 args = parser.parse_args(["list", "--config", str(config)])
85 _run_list(args)
86 captured = capsys.readouterr()
87 assert "No tunnels" in captured.out
88
89
90class TestRunAdd:
91 def test_add_creates_config(self, tmp_path, capsys):
92 config = tmp_path / "new.config"
93 parser = _get_parser()
94 args = parser.parse_args([
95 "add", "--config", str(config),
96 "--name", "testserver", "--type", "server",
97 "--port", "8080", "--target", "127.0.0.1"
98 ])
99 _run_add(args)
100 captured = capsys.readouterr()
101 assert "Added tunnel" in captured.out
102 assert config.exists()
103
104 def test_add_unknown_type(self, tmp_path):
105 config = tmp_path / "test.config"
106 parser = _get_parser()
107 args = parser.parse_args([
108 "add", "--config", str(config),
109 "--name", "bad", "--type", "nonexistent", "--port", "80"
110 ])
111 with pytest.raises(SystemExit):
112 _run_add(args)
113
114 def test_add_appends_to_existing(self, tmp_path, capsys):
115 config = tmp_path / "existing.config"
116 config.write_text(
117 "tunnel.0.name=first\n"
118 "tunnel.0.type=httpclient\n"
119 "tunnel.0.listenPort=4444\n"
120 "tunnel.0.interface=127.0.0.1\n"
121 )
122 parser = _get_parser()
123 args = parser.parse_args([
124 "add", "--config", str(config),
125 "--name", "second", "--type", "httpclient",
126 "--port", "4445"
127 ])
128 _run_add(args)
129 # Verify both tunnels exist
130 text = config.read_text()
131 assert "first" in text
132 assert "second" in text
133
134
135class TestMain:
136 def test_no_args_prints_help(self):
137 with patch("sys.argv", ["i2p-tunnel"]):
138 with pytest.raises(SystemExit) as exc:
139 main()
140 assert exc.value.code == 1
141
142 def test_list_command(self, tmp_path, capsys):
143 config = tmp_path / "t.config"
144 config.write_text("")
145 with patch("sys.argv", ["i2p-tunnel", "list", "--config", str(config)]):
146 main()
147 captured = capsys.readouterr()
148 assert "No tunnels" in captured.out