1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
|
"""Tests for the import-wireguard-configs.sh one-time migration script.
The script imports every assets/wireguard-config/*.conf into NetworkManager
as a wireguard connection with autoconnect forced off. NM quirks under test:
the import filename must be a valid interface name (<= 15 chars), so every
config stages through a temp copy named wgpvpn.conf and is renamed to the
real config name immediately after import — by the UUID parsed from the
import output, never by the transient wgpvpn name. A leftover connection
literally named wgpvpn (an earlier run died between import and rename, so
it still has autoconnect on) makes the script refuse to run.
nmcli is faked via a stub on PATH (fake-nmcli in this directory) that logs
every invocation and snapshots the staged import file.
Run from repo root:
python3 -m unittest tests.import-wireguard-configs.test_import_wireguard_configs
"""
import os
import shutil
import subprocess
import tempfile
import unittest
REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))
SCRIPT = os.path.join(REPO_ROOT, "scripts", "import-wireguard-configs.sh")
FAKE_NMCLI = os.path.join(os.path.dirname(os.path.abspath(__file__)), "fake-nmcli")
class ImportWireguardConfigs(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.mkdtemp(prefix="import-wg-test-")
self.addCleanup(shutil.rmtree, self.tmp, ignore_errors=True)
self.confdir = os.path.join(self.tmp, "configs")
os.mkdir(self.confdir)
self.bindir = os.path.join(self.tmp, "bin")
os.mkdir(self.bindir)
shutil.copy(FAKE_NMCLI, os.path.join(self.bindir, "nmcli"))
os.chmod(os.path.join(self.bindir, "nmcli"), 0o755)
self.log = os.path.join(self.tmp, "nmcli.log")
def write_conf(self, name, body="[Interface]\nPrivateKey = k\n"):
path = os.path.join(self.confdir, name + ".conf")
with open(path, "w") as f:
f.write(body)
return path
def run_script(self, confdir=None, names="", env_extra=None):
env = dict(os.environ)
env["PATH"] = self.bindir + os.pathsep + env["PATH"]
env["FAKE_NMCLI_LOG"] = self.log
env["FAKE_NMCLI_NAMES"] = names
if env_extra:
env.update(env_extra)
return subprocess.run(
["bash", SCRIPT, confdir or self.confdir],
capture_output=True, text=True, timeout=10, env=env,
)
def log_lines(self):
if not os.path.exists(self.log):
return []
with open(self.log) as f:
return [ln.strip() for ln in f if ln.strip()]
# --- Normal cases ----------------------------------------------------
def test_imports_every_conf_with_autoconnect_off(self):
self.write_conf("USNY")
self.write_conf("USDC")
r = self.run_script()
self.assertEqual(r.returncode, 0, r.stderr)
modifies = [ln for ln in self.log_lines() if ln.startswith("connection modify")]
self.assertEqual(len(modifies), 2)
for ln in modifies:
self.assertIn("connection.autoconnect no", ln)
self.assertIn("imported: USDC", r.stdout)
self.assertIn("imported: USNY", r.stdout)
def test_renames_by_uuid_from_import_output_not_by_name(self):
self.write_conf("USNY")
r = self.run_script()
self.assertEqual(r.returncode, 0, r.stderr)
modify = [ln for ln in self.log_lines() if ln.startswith("connection modify")][0]
# The modify targets the UUID the import printed, and never the
# transient wgpvpn name.
self.assertIn("00000000-aaaa-bbbb-cccc-dddddddddddd", modify)
self.assertIn("connection.id USNY", modify)
self.assertNotIn("modify wgpvpn", modify)
def test_long_name_stages_through_wgpvpn_temp_copy(self):
# switzerlan-zurich1 is 18 chars — over NM's 15-char interface-name
# limit, the reason the staging copy exists at all.
body = "[Interface]\nPrivateKey = long-name-key\n"
self.write_conf("switzerlan-zurich1", body)
r = self.run_script()
self.assertEqual(r.returncode, 0, r.stderr)
staged = os.listdir(self.log + ".d")
self.assertEqual(len(staged), 1)
self.assertTrue(staged[0].endswith("wgpvpn.conf"), staged)
with open(os.path.join(self.log + ".d", staged[0])) as f:
self.assertEqual(f.read(), body)
self.assertIn("imported: switzerlan-zurich1", r.stdout)
# --- Idempotence -----------------------------------------------------
def test_already_imported_names_skip(self):
self.write_conf("USNY")
self.write_conf("USDC")
r = self.run_script(names="USNY\nsome-wifi")
self.assertEqual(r.returncode, 0, r.stderr)
self.assertIn("skip: USNY", r.stdout)
self.assertIn("imported: USDC", r.stdout)
modifies = [ln for ln in self.log_lines() if ln.startswith("connection modify")]
self.assertEqual(len(modifies), 1)
def test_all_imported_is_a_clean_noop(self):
self.write_conf("USNY")
r = self.run_script(names="USNY")
self.assertEqual(r.returncode, 0, r.stderr)
imports = [ln for ln in self.log_lines() if ln.startswith("connection import")]
self.assertEqual(imports, [])
# --- Boundary cases --------------------------------------------------
def test_empty_config_dir_fails_loudly(self):
r = self.run_script()
self.assertEqual(r.returncode, 1)
self.assertIn("no .conf files", r.stderr)
def test_missing_config_dir_fails_loudly(self):
r = self.run_script(confdir=os.path.join(self.tmp, "nope"))
self.assertEqual(r.returncode, 1)
self.assertIn("no such config dir", r.stderr)
# --- Error cases -----------------------------------------------------
def test_stale_wgpvpn_connection_refuses_to_run(self):
self.write_conf("USNY")
r = self.run_script(names="wgpvpn\nUSDC")
self.assertEqual(r.returncode, 1)
self.assertIn("stale", r.stderr)
self.assertIn("nmcli connection delete wgpvpn", r.stderr)
imports = [ln for ln in self.log_lines() if ln.startswith("connection import")]
self.assertEqual(imports, [])
def test_unparseable_import_output_aborts(self):
self.write_conf("USNY")
r = self.run_script(env_extra={"FAKE_NMCLI_IMPORT_OUT": "something unexpected"})
self.assertEqual(r.returncode, 1)
self.assertIn("could not parse a UUID", r.stderr)
modifies = [ln for ln in self.log_lines() if ln.startswith("connection modify")]
self.assertEqual(modifies, [])
def test_modify_failure_aborts_the_run(self):
self.write_conf("USNY")
self.write_conf("USDC")
r = self.run_script(env_extra={"FAKE_NMCLI_MODIFY_RC": "4"})
self.assertNotEqual(r.returncode, 0)
# set -e stops at the first failed modify — only one import attempted.
imports = [ln for ln in self.log_lines() if ln.startswith("connection import")]
self.assertEqual(len(imports), 1)
if __name__ == "__main__":
unittest.main()
|
