3 files changed, 131 insertions, 0 deletions
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/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()