1 files changed, 109 insertions, 0 deletions
diff --git a/tests/installer-steps/test_mask_tmp_mount_for_zfs.py b/tests/installer-steps/test_mask_tmp_mount_for_zfs.py
new file mode 100644
index 0000000..f03086f
--- /dev/null
+++ b/tests/installer-steps/test_mask_tmp_mount_for_zfs.py
@@ -0,0 +1,109 @@
+"""Test mask_tmp_mount_for_zfs — the tmpfs-over-ZFS /tmp guard.
+
+When the pool carries a /tmp dataset, systemd's tmp.mount (tmpfs) races it at
+boot; when tmpfs wins, the dataset is shadowed and systemd-tmpfiles-clean
+fails repeatedly with "Protocol driver not attached" (velox, 2026-04-10). The
+fix is masking tmp.mount so the dataset owns /tmp — but only when such a
+dataset actually exists, and never on a machine without zfs at all.
+
+Method: sed-extract mask_tmp_mount_for_zfs from the real `archsetup`; fake
+zfs / run_task / systemctl. The zfs-binary-absent case runs with a stripped
+PATH containing only the tools the function itself needs.
+
+Run from repo root:
+    python3 -m unittest tests.installer-steps.test_mask_tmp_mount_for_zfs
+"""
+
+import os
+import re
+import subprocess
+import tempfile
+import textwrap
+import unittest
+
+REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))
+ARCHSETUP = os.path.join(REPO_ROOT, "archsetup")
+
+
+def run(zfs_body=None, strip_zfs_from_path=False):
+    """zfs_body: bash body for a fake zfs function, or None for no fake.
+    strip_zfs_from_path: run with a minimal PATH that has grep/sed but no zfs.
+    """
+    with tempfile.TemporaryDirectory() as d:
+        path_setup = ""
+        if strip_zfs_from_path:
+            fakebin = os.path.join(d, "bin")
+            os.makedirs(fakebin)
+            for tool in ("grep", "sed", "cat"):
+                src = subprocess.run(["bash", "-lc", f"command -v {tool}"],
+                                     capture_output=True, text=True).stdout.strip()
+                if src:
+                    os.symlink(src, os.path.join(fakebin, tool))
+            path_setup = f'PATH="{fakebin}"'
+        zfs_fake = f"zfs() {{ {zfs_body} }}" if zfs_body is not None else ""
+        script = textwrap.dedent(f"""\
+            logfile=/dev/null
+            action=""
+            display() {{ :; }}
+            run_task() {{ echo "TASK: $1"; shift; "$@"; }}
+            systemctl() {{ echo "SYSTEMCTL: $*"; }}
+            error_warn() {{ echo "WARN: $1"; return 1; }}
+            {zfs_fake}
+            source <(sed -n '/^mask_tmp_mount_for_zfs() {{/,/^}}/p' "{ARCHSETUP}")
+            {path_setup}
+            mask_tmp_mount_for_zfs
+            echo "RC=$?"
+            exit 0
+        """)
+        return subprocess.run(
+            ["bash", "-c", script], capture_output=True, text=True, timeout=10,
+        )
+
+
+def rc_of(r):
+    m = re.search(r"^RC=(\d+)$", r.stdout, re.M)
+    assert m, "no RC line in output: %r / %r" % (r.stdout, r.stderr)
+    return int(m.group(1))
+
+
+class MaskTmpMountForZfs(unittest.TestCase):
+    # ------------------------------------------------------------ normal ----
+    def test_tmp_dataset_present_masks_tmp_mount(self):
+        r = run(zfs_body='printf "/\\n/home\\n/tmp\\n";')
+        self.assertIn("SYSTEMCTL: mask tmp.mount", r.stdout)
+        self.assertEqual(rc_of(r), 0)
+
+    # ---------------------------------------------------------- boundary ----
+    def test_no_tmp_dataset_is_a_no_op(self):
+        r = run(zfs_body='printf "/\\n/home\\n/var\\n";')
+        self.assertNotIn("SYSTEMCTL:", r.stdout)
+        self.assertEqual(rc_of(r), 0)
+
+    def test_tmp_prefix_dataset_does_not_match(self):
+        # /tmp/scratch or /tmpfoo must not trigger the mask — exact match only.
+        r = run(zfs_body='printf "/\\n/tmp/scratch\\n/tmpfoo\\n";')
+        self.assertNotIn("SYSTEMCTL:", r.stdout)
+        self.assertEqual(rc_of(r), 0)
+
+    def test_legacy_and_none_mountpoints_do_not_match(self):
+        r = run(zfs_body='printf "legacy\\nnone\\n-\\n";')
+        self.assertNotIn("SYSTEMCTL:", r.stdout)
+        self.assertEqual(rc_of(r), 0)
+
+    # ------------------------------------------------------------- error ----
+    def test_zfs_binary_absent_is_a_silent_no_op(self):
+        r = run(zfs_body=None, strip_zfs_from_path=True)
+        self.assertNotIn("SYSTEMCTL:", r.stdout)
+        self.assertNotIn("WARN:", r.stdout)
+        self.assertEqual(rc_of(r), 0)
+
+    def test_zfs_list_failure_is_a_no_op_not_a_crash(self):
+        r = run(zfs_body='echo "no pools available" >&2; return 1;')
+        self.assertNotIn("SYSTEMCTL:", r.stdout)
+        self.assertEqual(rc_of(r), 0)
+        self.assertNotIn("no pools available", r.stdout,
+                         "zfs stderr noise must not leak into output")
+
+
+if __name__ == "__main__":
+    unittest.main()