aboutsummaryrefslogtreecommitdiff
path: root/tests/import-wireguard-configs/test_import_wireguard_configs.py
diff options
context:
space:
mode:
Diffstat (limited to 'tests/import-wireguard-configs/test_import_wireguard_configs.py')
-rw-r--r--tests/import-wireguard-configs/test_import_wireguard_configs.py167
1 files changed, 167 insertions, 0 deletions
diff --git a/tests/import-wireguard-configs/test_import_wireguard_configs.py b/tests/import-wireguard-configs/test_import_wireguard_configs.py
new file mode 100644
index 0000000..0307041
--- /dev/null
+++ b/tests/import-wireguard-configs/test_import_wireguard_configs.py
@@ -0,0 +1,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()