aboutsummaryrefslogtreecommitdiff
path: root/tests
diff options
context:
space:
mode:
Diffstat (limited to 'tests')
-rw-r--r--tests/import-wireguard-configs/fake-nmcli45
-rw-r--r--tests/import-wireguard-configs/test_import_wireguard_configs.py167
-rw-r--r--tests/installer-steps/test_orchestrators.py1
-rw-r--r--tests/nvidia-preflight/test_nvidia_preflight.py162
-rwxr-xr-xtests/zfs-pre-snapshot/fake-zfs14
-rw-r--r--tests/zfs-pre-snapshot/test_zfs_pre_snapshot.py116
6 files changed, 505 insertions, 0 deletions
diff --git a/tests/import-wireguard-configs/fake-nmcli b/tests/import-wireguard-configs/fake-nmcli
new file mode 100644
index 0000000..45b88cd
--- /dev/null
+++ b/tests/import-wireguard-configs/fake-nmcli
@@ -0,0 +1,45 @@
+#!/bin/bash
+# Fake nmcli for the import-wireguard-configs tests.
+#
+# Behavior is driven by env vars set by the test harness:
+# FAKE_NMCLI_LOG file every invocation's args are appended to (one line
+# per call; for imports the staged file's basename and
+# content hash context are visible in the args)
+# FAKE_NMCLI_NAMES newline-separated connection names returned by
+# `nmcli -t -f NAME connection show`
+# FAKE_NMCLI_IMPORT_OUT override for the import command's stdout
+# (default: the real NM success line with a per-call
+# deterministic UUID)
+# FAKE_NMCLI_MODIFY_RC exit code for `nmcli connection modify` (default 0)
+#
+# Import calls also copy the staged file into $FAKE_NMCLI_LOG.d/ so tests can
+# assert the temp copy was named wgpvpn.conf and carried the right content.
+set -euo pipefail
+
+echo "$*" >>"$FAKE_NMCLI_LOG"
+
+case "$1 $2" in
+"-t -f")
+ # nmcli -t -f NAME connection show
+ printf '%s\n' "${FAKE_NMCLI_NAMES:-}"
+ ;;
+"connection import")
+ # nmcli connection import type wireguard file <path>
+ file="${6:?}"
+ mkdir -p "$FAKE_NMCLI_LOG.d"
+ n=$(find "$FAKE_NMCLI_LOG.d" -type f | wc -l)
+ cp "$file" "$FAKE_NMCLI_LOG.d/import-$n-$(basename "$file")"
+ if [ -n "${FAKE_NMCLI_IMPORT_OUT:-}" ]; then
+ echo "$FAKE_NMCLI_IMPORT_OUT"
+ else
+ printf "Connection 'wgpvpn' (%08d-aaaa-bbbb-cccc-dddddddddddd) successfully added.\n" "$n"
+ fi
+ ;;
+"connection modify")
+ exit "${FAKE_NMCLI_MODIFY_RC:-0}"
+ ;;
+*)
+ echo "fake-nmcli: unexpected args: $*" >&2
+ exit 99
+ ;;
+esac
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()
diff --git a/tests/installer-steps/test_orchestrators.py b/tests/installer-steps/test_orchestrators.py
index e62c198..48b7508 100644
--- a/tests/installer-steps/test_orchestrators.py
+++ b/tests/installer-steps/test_orchestrators.py
@@ -46,6 +46,7 @@ ORCHESTRATORS = {
"tighten_efi_permissions", "add_nvme_early_module",
"configure_initramfs_hook", "configure_encrypted_autologin",
"configure_tlp_power", "trim_firmware", "configure_grub",
+ "configure_pre_pacman_snapshots",
],
"user_customizations": [
"clone_user_repos", "stow_dotfiles", "prune_waybar_battery",
diff --git a/tests/nvidia-preflight/test_nvidia_preflight.py b/tests/nvidia-preflight/test_nvidia_preflight.py
new file mode 100644
index 0000000..bdacfd5
--- /dev/null
+++ b/tests/nvidia-preflight/test_nvidia_preflight.py
@@ -0,0 +1,162 @@
+"""Tests for the nvidia_preflight_report helper in the archsetup installer.
+
+nvidia_preflight_report is the pure core of the NVIDIA/Wayland preflight
+check: it scans DRM (then PCI display-class) modalias files for the NVIDIA
+vendor id, and when one matches it prints the Wayland warning + required
+environment variables and checks the repo's candidate nvidia-utils major
+version. Return codes: 0 = no NVIDIA GPU, 10 = NVIDIA and the driver
+requirement (535+) is met, 11 = NVIDIA and the requirement is not met
+(driver too old or unknown). The interactive continue/abort prompt lives in
+preflight_checks, not here, so this core is unit testable.
+
+These tests exercise the REAL function body, extracted from the `archsetup`
+script at run time (not a copy), against temp modalias trees and a fake
+pacman on PATH.
+
+Run from repo root:
+ python3 -m unittest tests.nvidia-preflight.test_nvidia_preflight
+"""
+
+import os
+import shutil
+import subprocess
+import tempfile
+import unittest
+
+
+REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))
+ARCHSETUP = os.path.join(REPO_ROOT, "archsetup")
+
+NVIDIA_MODALIAS = "pci:v000010DEd00002684sv00001043sd000088E2bc03sc00i00"
+NVIDIA_MODALIAS_LOWER = "pci:v000010ded00002684sv00001043sd000088e2bc03sc00i00"
+AMD_MODALIAS = "pci:v00001002d0000164Esv00001462sd00007D78bc03sc80i00"
+NON_DISPLAY_NVIDIA = "pci:v000010DEd00002684sv00001043sd000088E2bc0Csc03i30"
+
+
+class NvidiaPreflightHarness(unittest.TestCase):
+ """Source nvidia_preflight_report out of the real archsetup script."""
+
+ def setUp(self):
+ self.tmp = tempfile.mkdtemp(prefix="nvidia-preflight-test-")
+ self.drm = os.path.join(self.tmp, "drm")
+ self.pci = os.path.join(self.tmp, "pci")
+ os.makedirs(self.drm)
+ os.makedirs(self.pci)
+ self.fakebin = os.path.join(self.tmp, "bin")
+ os.makedirs(self.fakebin)
+ self.wrapper = os.path.join(self.tmp, "run.sh")
+ with open(self.wrapper, "w") as f:
+ f.write(
+ "#!/bin/bash\n"
+ 'ARCHSETUP="$1"; shift\n'
+ "source <(sed -n "
+ "'/^nvidia_preflight_report() {/,/^}/p' \"$ARCHSETUP\")\n"
+ "nvidia_preflight_report\n"
+ )
+ os.chmod(self.wrapper, 0o755)
+
+ def tearDown(self):
+ shutil.rmtree(self.tmp, ignore_errors=True)
+
+ def fake_pacman(self, version=None, fail=False):
+ """A pacman stub answering `pacman -Si nvidia-utils`."""
+ path = os.path.join(self.fakebin, "pacman")
+ with open(path, "w") as f:
+ if fail:
+ f.write("#!/bin/sh\nexit 1\n")
+ else:
+ f.write(
+ "#!/bin/sh\n"
+ "printf 'Repository : extra\\n'\n"
+ "printf 'Name : nvidia-utils\\n'\n"
+ "printf 'Version : %s\\n'\n" % version
+ )
+ os.chmod(path, 0o755)
+
+ def add_modalias(self, root, subdir, content):
+ d = os.path.join(root, subdir)
+ os.makedirs(d, exist_ok=True)
+ with open(os.path.join(d, "modalias"), "w") as f:
+ f.write(content + "\n")
+
+ def run_check(self):
+ env = dict(os.environ)
+ env["PATH"] = self.fakebin + os.pathsep + env["PATH"]
+ env["NVIDIA_DRM_GLOB"] = os.path.join(self.drm, "card*", "modalias")
+ env["NVIDIA_PCI_GLOB"] = os.path.join(self.pci, "*", "modalias")
+ return subprocess.run(
+ ["bash", self.wrapper, ARCHSETUP],
+ capture_output=True, text=True, env=env,
+ )
+
+ # ---------------------------------------------------------- normal ----
+ def test_no_gpu_files_returns_zero_and_silent(self):
+ self.fake_pacman(version="575.51.02-1")
+ r = self.run_check()
+ self.assertEqual(r.returncode, 0)
+ self.assertNotIn("NVIDIA", r.stdout)
+
+ def test_amd_only_returns_zero(self):
+ self.fake_pacman(version="575.51.02-1")
+ self.add_modalias(self.drm, "card0", AMD_MODALIAS)
+ r = self.run_check()
+ self.assertEqual(r.returncode, 0)
+ self.assertNotIn("NVIDIA", r.stdout)
+
+ def test_nvidia_with_modern_driver_returns_ten_with_guidance(self):
+ self.fake_pacman(version="575.51.02-1")
+ self.add_modalias(self.drm, "card0", NVIDIA_MODALIAS)
+ r = self.run_check()
+ self.assertEqual(r.returncode, 10)
+ self.assertIn("NVIDIA GPU detected", r.stdout)
+ self.assertIn("LIBVA_DRIVER_NAME=nvidia", r.stdout)
+ self.assertIn("GBM_BACKEND=nvidia-drm", r.stdout)
+ self.assertIn("__GLX_VENDOR_LIBRARY_NAME=nvidia", r.stdout)
+ self.assertIn("575.51.02-1", r.stdout)
+
+ # -------------------------------------------------------- boundary ----
+ def test_lowercase_vendor_id_detected(self):
+ self.fake_pacman(version="575.51.02-1")
+ self.add_modalias(self.drm, "card0", NVIDIA_MODALIAS_LOWER)
+ r = self.run_check()
+ self.assertEqual(r.returncode, 10)
+
+ def test_exactly_535_meets_requirement(self):
+ self.fake_pacman(version="535.216.01-1")
+ self.add_modalias(self.drm, "card0", NVIDIA_MODALIAS)
+ r = self.run_check()
+ self.assertEqual(r.returncode, 10)
+
+ def test_pci_fallback_display_class_only(self):
+ # No DRM entries; PCI holds a display-class NVIDIA device -> detected.
+ self.fake_pacman(version="575.51.02-1")
+ self.add_modalias(self.pci, "0000:01:00.0", NVIDIA_MODALIAS)
+ r = self.run_check()
+ self.assertEqual(r.returncode, 10)
+
+ def test_pci_non_display_nvidia_ignored(self):
+ # An NVIDIA audio/usb function (bc0C) must not trigger the check.
+ self.fake_pacman(version="575.51.02-1")
+ self.add_modalias(self.pci, "0000:01:00.1", NON_DISPLAY_NVIDIA)
+ r = self.run_check()
+ self.assertEqual(r.returncode, 0)
+
+ # ----------------------------------------------------------- error ----
+ def test_old_driver_returns_eleven_with_error(self):
+ self.fake_pacman(version="470.256.02-1")
+ self.add_modalias(self.drm, "card0", NVIDIA_MODALIAS)
+ r = self.run_check()
+ self.assertEqual(r.returncode, 11)
+ self.assertIn("535", r.stdout)
+ self.assertIn("470.256.02-1", r.stdout)
+
+ def test_pacman_failure_returns_eleven_unknown(self):
+ self.fake_pacman(fail=True)
+ self.add_modalias(self.drm, "card0", NVIDIA_MODALIAS)
+ r = self.run_check()
+ self.assertEqual(r.returncode, 11)
+ self.assertIn("unknown", r.stdout)
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/tests/zfs-pre-snapshot/fake-zfs b/tests/zfs-pre-snapshot/fake-zfs
new file mode 100755
index 0000000..508c0f3
--- /dev/null
+++ b/tests/zfs-pre-snapshot/fake-zfs
@@ -0,0 +1,14 @@
+#!/bin/sh
+# Fake zfs for the zfs-pre-snapshot unit test. `snapshot` and `destroy` are
+# logged (FAKE_ZFS_LOG); `list` prints a fixture snapshot set (FAKE_ZFS_SNAPSHOTS).
+# Set FAKE_ZFS_SNAPSHOT_FAIL to make snapshot creation fail.
+case "$1" in
+ snapshot)
+ [ -n "$FAKE_ZFS_SNAPSHOT_FAIL" ] && exit 1
+ echo "snapshot $2" >> "$FAKE_ZFS_LOG"; exit 0 ;;
+ destroy)
+ echo "destroy $2" >> "$FAKE_ZFS_LOG"; exit 0 ;;
+ list)
+ cat "$FAKE_ZFS_SNAPSHOTS" 2>/dev/null; exit 0 ;;
+esac
+exit 0
diff --git a/tests/zfs-pre-snapshot/test_zfs_pre_snapshot.py b/tests/zfs-pre-snapshot/test_zfs_pre_snapshot.py
new file mode 100644
index 0000000..ed7731b
--- /dev/null
+++ b/tests/zfs-pre-snapshot/test_zfs_pre_snapshot.py
@@ -0,0 +1,116 @@
+"""Unit tests for scripts/zfs-pre-snapshot.
+
+The script snapshots the root dataset before a pacman transaction and prunes to
+the most recent KEEP pre-pacman snapshots. These tests drive the real script
+with a fake zfs on PATH (snapshot/destroy logged, list returns a fixture set)
+and env-rooted state, so nothing touches a real pool.
+"""
+
+import os
+import shutil
+import subprocess
+import tempfile
+import time
+import unittest
+
+REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))
+SCRIPT = os.path.join(REPO_ROOT, "scripts/zfs-pre-snapshot")
+FAKE_ZFS = os.path.join(os.path.dirname(__file__), "fake-zfs")
+
+DATASET = "tank/test"
+# Five pre-pacman snapshots oldest->newest (zfs list -s creation is ascending),
+# plus one autosnap that the grep filter must ignore.
+SNAPSHOTS = "\n".join([
+ f"{DATASET}@autosnap_2026-01-01",
+ f"{DATASET}@pre-pacman_2026-06-01",
+ f"{DATASET}@pre-pacman_2026-06-02",
+ f"{DATASET}@pre-pacman_2026-06-03",
+ f"{DATASET}@pre-pacman_2026-06-04",
+ f"{DATASET}@pre-pacman_2026-06-05",
+]) + "\n"
+
+
+class Harness(unittest.TestCase):
+ def setUp(self):
+ self.tmp = tempfile.mkdtemp(prefix="zfs-pre-snap-")
+ self.bin = os.path.join(self.tmp, "bin")
+ os.makedirs(self.bin)
+ shutil.copy(FAKE_ZFS, os.path.join(self.bin, "zfs"))
+ self.log = os.path.join(self.tmp, "zfs.log")
+ self.snaps = os.path.join(self.tmp, "snaps")
+ with open(self.snaps, "w") as f:
+ f.write(SNAPSHOTS)
+ self.lock = os.path.join(self.tmp, "lock")
+
+ def tearDown(self):
+ shutil.rmtree(self.tmp, ignore_errors=True)
+
+ def run_script(self, keep="3", fail=False, snaps=None):
+ env = os.environ.copy()
+ env["PATH"] = self.bin + os.pathsep + env["PATH"]
+ env["ZFS_PRE_DATASET"] = DATASET
+ env["ZFS_PRE_LOCKFILE"] = self.lock
+ env["ZFS_PRE_KEEP"] = keep
+ env["FAKE_ZFS_LOG"] = self.log
+ env["FAKE_ZFS_SNAPSHOTS"] = snaps if snaps is not None else self.snaps
+ if fail:
+ env["FAKE_ZFS_SNAPSHOT_FAIL"] = "1"
+ return subprocess.run([SCRIPT], env=env, capture_output=True, text=True,
+ timeout=15)
+
+ def log_lines(self):
+ try:
+ with open(self.log) as f:
+ return [ln for ln in f.read().splitlines() if ln.strip()]
+ except FileNotFoundError:
+ return []
+
+
+class TestSnapshot(Harness):
+ def test_creates_a_pre_pacman_snapshot(self):
+ self.run_script()
+ snaps = [ln for ln in self.log_lines() if ln.startswith("snapshot ")]
+ self.assertEqual(len(snaps), 1)
+ self.assertIn(f"snapshot {DATASET}@pre-pacman_", snaps[0])
+
+ def test_skips_when_lockfile_is_fresh(self):
+ # A lockfile newer than MIN_INTERVAL → no snapshot this run.
+ open(self.lock, "w").close()
+ os.utime(self.lock, (time.time(), time.time()))
+ self.run_script()
+ self.assertEqual([ln for ln in self.log_lines()
+ if ln.startswith("snapshot ")], [])
+
+
+class TestPrune(Harness):
+ def test_prunes_oldest_beyond_keep(self):
+ # 5 pre-pacman snapshots, KEEP=3 → the two oldest are destroyed.
+ self.run_script(keep="3")
+ destroyed = [ln.split(" ", 1)[1] for ln in self.log_lines()
+ if ln.startswith("destroy ")]
+ self.assertEqual(destroyed,
+ [f"{DATASET}@pre-pacman_2026-06-01",
+ f"{DATASET}@pre-pacman_2026-06-02"])
+
+ def test_never_destroys_non_pre_pacman_snapshots(self):
+ self.run_script(keep="1")
+ destroyed = [ln for ln in self.log_lines() if ln.startswith("destroy ")]
+ self.assertFalse(any("autosnap" in ln for ln in destroyed))
+
+ def test_no_prune_when_at_or_under_keep(self):
+ # KEEP=5 with exactly 5 pre-pacman snapshots → nothing destroyed.
+ self.run_script(keep="5")
+ self.assertEqual([ln for ln in self.log_lines()
+ if ln.startswith("destroy ")], [])
+
+
+class TestError(Harness):
+ def test_snapshot_failure_skips_prune_and_warns(self):
+ r = self.run_script(fail=True)
+ self.assertIn("Failed to create snapshot", r.stderr)
+ self.assertEqual([ln for ln in self.log_lines()
+ if ln.startswith("destroy ")], [])
+
+
+if __name__ == "__main__":
+ unittest.main()