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
|
"""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()
|
