From 3becfac66dc569e71d0f6085fb56e92cfa6d626a Mon Sep 17 00:00:00 2001 From: Craig Jennings Date: Fri, 24 Jul 2026 12:19:04 -0500 Subject: fix(installer): eight fixes from an overnight bug-hunt and its review I squashed these because the per-bug reasoning lives in todo.org, which this commit carries. Two could cost a machine. configure_initramfs_hook swapped the udev hook for systemd on a LUKS root, leaving a standalone encrypt hook under an init that never runs it. The rebuild succeeds and the installer exits clean, then the root won't unlock at the next boot. trim_firmware ran pacman -Rdd against twelve firmware packages behind a DMI gate reading product_name, where "Framework" never appears. That left it dead on the hardware it targets, and dangerous to fix the obvious way: this machine is a Framework Desktop whose Ryzen iGPU needs the amdgpu firmware. It refuses on PCI modalias evidence now. wipedisk discarded before it checked. blkdiscard ran with -f, which disables the exclusive open, so picking the wrong disk destroyed a live filesystem and then reported that nothing had happened. Four more are smaller. The NVIDIA preflight aborted dwm and headless installs over a driver floor they never need. zfs-replicate exited 0 after every dataset failed. Unattended installs blocked on two prompts, and the first fix for that inherited a [Y/n] default into passwordless console login. A fresh install left the dotfiles repo permanently dirty. The review found a pattern worth more than any single fix. Helpers had thorough tests and none proved they were called. Deleting the call left five suites green, including the guard on that pacman -Rdd. CALL_SITES now pins nine caller/callee pairs. The suite runs 341 tests at exit 0, with no new shellcheck findings. I proved every guard by deleting it and watching the intended test go red. --- tests/installer-steps/test_configure_autologin.py | 218 ++++++++++++++++ .../test_framework_firmware_trim.py | 233 ++++++++++++++++++ .../installer-steps/test_mark_volatile_configs.py | 216 ++++++++++++++++ tests/installer-steps/test_orchestrators.py | 58 +++++ tests/installer-steps/test_select_locale.py | 160 ++++++++++++ tests/installer-steps/test_switch_udev_hook.py | 178 ++++++++++++++ .../nvidia-preflight/test_nvidia_preflight_gate.py | 173 +++++++++++++ tests/wipedisk/test_wipedisk.py | 273 +++++++++++++++++++++ tests/zfs-replicate/test_zfs_replicate.py | 174 +++++++++++++ 9 files changed, 1683 insertions(+) create mode 100644 tests/installer-steps/test_configure_autologin.py create mode 100644 tests/installer-steps/test_framework_firmware_trim.py create mode 100644 tests/installer-steps/test_mark_volatile_configs.py create mode 100644 tests/installer-steps/test_select_locale.py create mode 100644 tests/installer-steps/test_switch_udev_hook.py create mode 100644 tests/nvidia-preflight/test_nvidia_preflight_gate.py create mode 100644 tests/wipedisk/test_wipedisk.py create mode 100644 tests/zfs-replicate/test_zfs_replicate.py (limited to 'tests') diff --git a/tests/installer-steps/test_configure_autologin.py b/tests/installer-steps/test_configure_autologin.py new file mode 100644 index 0000000..b4bc1c4 --- /dev/null +++ b/tests/installer-steps/test_configure_autologin.py @@ -0,0 +1,218 @@ +"""Tests for configure_autologin in the archsetup installer. + +configure_autologin decides whether tty1 gets an agetty --autologin drop-in. +Three inputs decide it: --autologin / --no-autologin (or AUTOLOGIN=yes/no in +the config file), which are explicit; and, when neither is set, an auto-detect +branch that prompts -- but only on an encrypted root, where the user has +already authenticated at boot. + +That prompt is the reason these tests exist. --config-file is the documented +unattended mode and AUTOLOGIN is optional in archsetup.conf.example, so an +unattended install of an encrypted machine reaches an interactive read with +nobody there to answer it. The step runs from boot_ux, the last step in STEPS, +so it blocks at the end of a 40-60 minute run. The prompt is advisory (it +carries its own [Y/n] default), so unattended must answer it rather than block +-- the same ruling nvidia_preflight already makes for its rc-10 prompt. + +These tests exercise the REAL function body, extracted from the `archsetup` +script at run time (not a copy), with is_encrypted_root stubbed to a chosen +answer and the drop-in written under a temp dir. + +Run from repo root: + python3 -m unittest tests.installer-steps.test_configure_autologin +""" + +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") + + +class ConfigureAutologinHarness(unittest.TestCase): + """Source configure_autologin out of the real archsetup script.""" + + def setUp(self): + self.tmp = tempfile.mkdtemp(prefix="autologin-test-") + self.dropin_dir = os.path.join(self.tmp, "getty@tty1.service.d") + 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 " + "'/^configure_autologin() {/,/^}/p' \"$ARCHSETUP\")\n" + # The step's only other collaborators: a display helper and the + # encryption probe. Both stubbed so the body is what runs. + 'display() { echo "DISPLAY:$2"; }\n' + "is_encrypted_root() { return \"${STUB_ENCRYPTED_RC:-1}\"; }\n" + 'configure_autologin "$AUTOLOGIN_DIR"\n' + 'echo "RETURNED=$?"\n' + # Whatever the prompt did not consume is still on stdin. + 'if IFS= read -r leftover; then echo "LEFTOVER:$leftover"; ' + 'else echo "LEFTOVER:"; fi\n' + ) + os.chmod(self.wrapper, 0o755) + + def tearDown(self): + shutil.rmtree(self.tmp, ignore_errors=True) + + def run_step(self, enable_autologin="", encrypted=False, config_file="", + stdin="", username="testuser"): + env = dict(os.environ) + env["enable_autologin"] = enable_autologin + env["config_file"] = config_file + env["username"] = username + env["AUTOLOGIN_DIR"] = self.dropin_dir + env["STUB_ENCRYPTED_RC"] = "0" if encrypted else "1" + return subprocess.run( + ["bash", self.wrapper, ARCHSETUP], + input=stdin, capture_output=True, text=True, env=env, + ) + + def dropin(self): + """The written drop-in's contents, or None when it wasn't written.""" + path = os.path.join(self.dropin_dir, "autologin.conf") + if not os.path.exists(path): + return None + with open(path) as f: + return f.read() + + def assertReturnedZero(self, r): + self.assertIn("RETURNED=0", r.stdout, + "step did not return 0; stdout=%r stderr=%r" + % (r.stdout, r.stderr)) + + # ---------------------------------------------------------- normal ---- + def test_explicit_true_writes_dropin_without_prompting(self): + r = self.run_step(enable_autologin="true", stdin="sentinel\n") + self.assertReturnedZero(r) + self.assertIn("--autologin testuser", self.dropin() or "") + self.assertIn("LEFTOVER:sentinel", r.stdout) + + def test_explicit_false_writes_nothing_without_prompting(self): + r = self.run_step(enable_autologin="false", encrypted=True, + stdin="sentinel\n") + self.assertReturnedZero(r) + self.assertIsNone(self.dropin()) + self.assertIn("LEFTOVER:sentinel", r.stdout) + + def test_autodetect_unencrypted_root_skips_silently(self): + r = self.run_step(encrypted=False, stdin="sentinel\n") + self.assertReturnedZero(r) + self.assertIsNone(self.dropin()) + self.assertIn("LEFTOVER:sentinel", r.stdout) + + # -------------------------------------------------------- boundary ---- + def test_autodetect_encrypted_interactive_yes_writes_dropin(self): + # bash emits `read -p`'s prompt only when stdin is a terminal, so the + # prompt text is unassertable here. What is assertable is that the read + # consumed exactly the answer line and left the rest of stdin alone. + r = self.run_step(encrypted=True, stdin="y\nsentinel\n") + self.assertReturnedZero(r) + self.assertIn("--autologin testuser", self.dropin() or "") + self.assertIn("LEFTOVER:sentinel", r.stdout) + + def test_autodetect_encrypted_bare_enter_takes_the_yes_default(self): + r = self.run_step(encrypted=True, stdin="\nsentinel\n") + self.assertReturnedZero(r) + self.assertIn("--autologin testuser", self.dropin() or "") + self.assertIn("LEFTOVER:sentinel", r.stdout) + + def test_unattended_config_file_does_not_block_on_the_prompt(self): + # --config-file is the documented unattended mode and AUTOLOGIN is + # optional in archsetup.conf.example, so this is the ordinary + # unattended install of an encrypted machine. It must not read stdin. + r = self.run_step(encrypted=True, config_file="/etc/archsetup.conf", + stdin="sentinel\n") + self.assertReturnedZero(r) + self.assertIn("LEFTOVER:sentinel", r.stdout) + + def test_unattended_leaves_autologin_off(self): + # A prompt default is calibrated for someone who can override it in the + # moment. Unattended there is no override, so the default stops being + # advisory and becomes policy -- and this policy is passwordless console + # login. "Encrypted root means already authenticated" holds only at + # boot: it does not hold for a stolen powered-on machine, where tty1 is + # one Ctrl-Alt-F1 away and a screen locker does not lock the console. + # + # The error costs are asymmetric. Wrongly ON is a silent authentication + # downgrade nobody notices. Wrongly OFF is one password prompt, noticed + # immediately, fixed with a config key that already exists. + r = self.run_step(encrypted=True, config_file="/etc/archsetup.conf") + self.assertReturnedZero(r) + self.assertIsNone(self.dropin()) + + def test_unattended_says_why_it_left_autologin_off(self): + # Silently declining is its own trap: the user gets a login prompt they + # did not expect and no idea which knob controls it. + r = self.run_step(encrypted=True, config_file="/etc/archsetup.conf") + self.assertIn("AUTOLOGIN", r.stdout) + + def test_unattended_explicit_yes_still_enables_it(self): + # Declining by default must not make the feature unreachable + # unattended. AUTOLOGIN=yes is the opt-in and it still works. + r = self.run_step(enable_autologin="true", encrypted=True, + config_file="/etc/archsetup.conf", stdin="sentinel\n") + self.assertReturnedZero(r) + self.assertIn("--autologin testuser", self.dropin() or "") + self.assertIn("LEFTOVER:sentinel", r.stdout) + + def test_unattended_explicit_no_still_wins_over_the_default(self): + # Unattended does not mean "always autologin" -- AUTOLOGIN=no is an + # answer, and the auto-detect default must not override it. + r = self.run_step(enable_autologin="false", encrypted=True, + config_file="/etc/archsetup.conf", stdin="sentinel\n") + self.assertReturnedZero(r) + self.assertIsNone(self.dropin()) + self.assertIn("LEFTOVER:sentinel", r.stdout) + + def test_unattended_unencrypted_root_still_writes_nothing(self): + r = self.run_step(encrypted=False, config_file="/etc/archsetup.conf", + stdin="sentinel\n") + self.assertReturnedZero(r) + self.assertIsNone(self.dropin()) + self.assertIn("LEFTOVER:sentinel", r.stdout) + + # ----------------------------------------------------------- error ---- + def test_autodetect_encrypted_interactive_no_writes_nothing(self): + r = self.run_step(encrypted=True, stdin="n\nsentinel\n") + self.assertReturnedZero(r) + self.assertIsNone(self.dropin()) + self.assertIn("LEFTOVER:sentinel", r.stdout) + + def test_autodetect_encrypted_interactive_capital_no_writes_nothing(self): + r = self.run_step(encrypted=True, stdin="No\nsentinel\n") + self.assertReturnedZero(r) + self.assertIsNone(self.dropin()) + self.assertIn("LEFTOVER:sentinel", r.stdout) + + def test_dropin_overrides_the_packaged_execstart(self): + # A drop-in that adds a second ExecStart without clearing the first is + # a unit systemd refuses to start -- tty1 would have no getty at all. + r = self.run_step(enable_autologin="true") + self.assertReturnedZero(r) + body = self.dropin() or "" + self.assertIn("ExecStart=\n", body) + self.assertLess(body.index("ExecStart=\n"), + body.index("ExecStart=-/sbin/agetty")) + + def test_closed_stdin_leaves_autologin_off(self): + # The same "nobody can answer" state reached without --config-file: a + # piped or redirected stdin at EOF. Pre-fix the read failed, response + # was empty, and `case *` enabled autologin silently. + r = self.run_step(encrypted=True, stdin="") + self.assertReturnedZero(r) + self.assertIsNone(self.dropin()) + + def test_closed_stdin_says_why(self): + r = self.run_step(encrypted=True, stdin="") + self.assertIn("stdin closed", r.stdout) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/installer-steps/test_framework_firmware_trim.py b/tests/installer-steps/test_framework_firmware_trim.py new file mode 100644 index 0000000..cdb6a60 --- /dev/null +++ b/tests/installer-steps/test_framework_firmware_trim.py @@ -0,0 +1,233 @@ +"""Tests for is_framework_intel_laptop, the gate on trim_firmware. + +trim_firmware runs `pacman -Rdd` against twelve linux-firmware packages, +keeping only the three a Framework 13 Intel needs (intel, atheros, realtek). +It is the most destructive conditional in the installer, so the condition is +what these tests pin. + +The gate has to be exact in both directions. Too narrow and the step is dead +code: "framework" lives in DMI's sys_vendor, never in product_name, so a gate +reading product_name alone matches no Framework machine at all. Too wide and it +removes linux-firmware-amdgpu from a Framework Desktop, whose Ryzen AI Max iGPU +needs it to bring up a display. Both machines are real: velox is +Framework/"Laptop (13th Gen Intel Core)" and must trim; ratio is +Framework/"Desktop (AMD Ryzen AI Max 300 Series)" and must not. + +These tests exercise the REAL function body, extracted from the `archsetup` +script at run time (not a copy), against a fixture DMI directory. + +Run from repo root: + python3 -m unittest tests.installer-steps.test_framework_firmware_trim +""" + +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") + + +class FrameworkIntelGateHarness(unittest.TestCase): + """Source is_framework_intel_laptop out of the real archsetup script.""" + + def setUp(self): + self.tmp = tempfile.mkdtemp(prefix="fw-trim-test-") + self.dmi = os.path.join(self.tmp, "dmi") + os.mkdir(self.dmi) + 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 " + "'/^is_framework_intel_laptop() {/,/^}/p' \"$ARCHSETUP\")\n" + 'if is_framework_intel_laptop "$DMI_DIR"; then echo "TRIM=yes"; ' + 'else echo "TRIM=no"; fi\n' + ) + os.chmod(self.wrapper, 0o755) + + def tearDown(self): + shutil.rmtree(self.tmp, ignore_errors=True) + + def dmi_write(self, **fields): + for name, value in fields.items(): + with open(os.path.join(self.dmi, name), "w") as f: + f.write(value + "\n") + + def trims(self, dmi_dir=None): + env = dict(os.environ) + env["DMI_DIR"] = self.dmi if dmi_dir is None else dmi_dir + r = subprocess.run(["bash", self.wrapper, ARCHSETUP], + capture_output=True, text=True, env=env) + if "TRIM=yes" in r.stdout: + return True + if "TRIM=no" in r.stdout: + return False + self.fail("no TRIM line; stdout=%r stderr=%r" % (r.stdout, r.stderr)) + + # ---------------------------------------------------------- normal ---- + def test_velox_framework_13_intel_trims(self): + # The exact strings read off velox on 2026-07-24. + self.dmi_write(sys_vendor="Framework", + product_name="Laptop (13th Gen Intel Core)") + self.assertTrue(self.trims()) + + def test_framework_12_intel_trims(self): + self.dmi_write(sys_vendor="Framework", + product_name="Laptop 12 (13th Gen Intel Core)") + self.assertTrue(self.trims()) + + # -------------------------------------------------------- boundary ---- + def test_ratio_framework_desktop_amd_does_not_trim(self): + # The exact strings read off ratio on 2026-07-24. Trimming here would + # remove linux-firmware-amdgpu from the machine whose iGPU needs it. + self.dmi_write(sys_vendor="Framework", + product_name="Desktop (AMD Ryzen AI Max 300 Series)") + self.assertFalse(self.trims()) + + def test_framework_16_amd_laptop_does_not_trim(self): + self.dmi_write(sys_vendor="Framework", + product_name="Laptop 16 (AMD Ryzen 7040 Series)") + self.assertFalse(self.trims()) + + def test_non_framework_intel_laptop_does_not_trim(self): + # The firmware set is only known for Framework hardware. The model + # deliberately carries both "Laptop" and "Intel", so the vendor check + # is the only condition that can reject it -- a fixture that failed the + # model checks too would pass no matter what the vendor check did. + self.dmi_write(sys_vendor="Dell Inc.", + product_name="Laptop 15 (Intel Core i7)") + self.assertFalse(self.trims()) + + def test_framework_desktop_naming_an_intel_cpu_does_not_trim(self): + # No such machine ships today -- the Framework Desktop is AMD -- but + # "an Intel Framework desktop will never exist" is exactly the kind of + # assumption that made the old gate wrong. The model must say Laptop, + # so a desktop is refused on the model rather than on its CPU vendor. + self.dmi_write(sys_vendor="Framework", + product_name="Desktop (Intel Core Ultra)") + self.assertFalse(self.trims()) + + def test_vendor_match_is_case_insensitive(self): + self.dmi_write(sys_vendor="framework", + product_name="Laptop (13th Gen intel Core)") + self.assertTrue(self.trims()) + + def test_vendor_with_surrounding_whitespace_still_matches(self): + # DMI strings are vendor-supplied and often padded. + self.dmi_write(sys_vendor=" Framework ", + product_name="Laptop (13th Gen Intel Core)") + self.assertTrue(self.trims()) + + # ----------------------------------------------------------- error ---- + def test_missing_dmi_directory_does_not_trim(self): + self.assertFalse(self.trims(dmi_dir=os.path.join(self.tmp, "nope"))) + + def test_missing_product_name_does_not_trim(self): + self.dmi_write(sys_vendor="Framework") + self.assertFalse(self.trims()) + + def test_missing_sys_vendor_does_not_trim(self): + self.dmi_write(product_name="Laptop (13th Gen Intel Core)") + self.assertFalse(self.trims()) + + def test_empty_dmi_values_do_not_trim(self): + self.dmi_write(sys_vendor="", product_name="") + self.assertFalse(self.trims()) + +class GpuVendorEvidence(unittest.TestCase): + """trim_firmware deletes GPU firmware, so what it needs to know is which + GPU is physically present -- and the kernel already answers that. + + The DMI model string cannot. It is a marketing name: it does not know a + swapped card, and it cannot see hardware the allowlist has never met. The + same file already reads PCI modalias twice (install_gpu_drivers, and + nvidia_preflight_report, which parameterises its globs for exactly this + kind of test). detect_gpu_vendors is that read, made reusable, so the trim + can refuse on evidence rather than on a name. + """ + + def setUp(self): + self.tmp = tempfile.mkdtemp(prefix="gpu-vendor-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.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 " + "'/^detect_gpu_vendors() {/,/^}/p' \"$ARCHSETUP\")\n" + 'detect_gpu_vendors "$DRM_GLOB" "$PCI_GLOB"\n' + ) + os.chmod(self.wrapper, 0o755) + + def tearDown(self): + shutil.rmtree(self.tmp, ignore_errors=True) + + def card(self, root, name, modalias): + d = os.path.join(root, name, "device") + os.makedirs(d, exist_ok=True) + with open(os.path.join(d, "modalias"), "w") as f: + f.write(modalias + "\n") + + def vendors(self): + env = dict(os.environ) + env["DRM_GLOB"] = os.path.join(self.drm, "card*", "device", "modalias") + env["PCI_GLOB"] = os.path.join(self.pci, "*", "device", "modalias") + r = subprocess.run(["bash", self.wrapper, ARCHSETUP], + capture_output=True, text=True, env=env) + return sorted(v for v in r.stdout.split() if v) + + # ---------------------------------------------------------- normal ---- + def test_intel_igpu_detected(self): + self.card(self.drm, "card0", "pci:v00008086d00007D55sv00001414bc03sc00i00") + self.assertEqual(self.vendors(), ["intel"]) + + def test_amd_igpu_detected(self): + self.card(self.drm, "card0", "pci:v00001002d0000150Esv00001414bc03sc00i00") + self.assertEqual(self.vendors(), ["amd"]) + + # -------------------------------------------------------- boundary ---- + def test_nvidia_detected_in_either_hex_case(self): + self.card(self.drm, "card0", "pci:v000010DEd00002684bc03sc00i00") + self.assertEqual(self.vendors(), ["nvidia"]) + shutil.rmtree(os.path.join(self.drm, "card0")) + self.card(self.drm, "card0", "pci:v000010ded00002684bc03sc00i00") + self.assertEqual(self.vendors(), ["nvidia"]) + + def test_hybrid_machine_reports_both(self): + self.card(self.drm, "card0", "pci:v00008086d00007D55bc03sc00i00") + self.card(self.drm, "card1", "pci:v000010DEd00002684bc03sc00i00") + self.assertEqual(self.vendors(), ["intel", "nvidia"]) + + def test_falls_back_to_pci_when_drm_is_empty(self): + # Early boot / chroot: no DRM nodes yet, same ruling install_gpu_drivers + # already makes. + self.card(self.pci, "0000:00:02.0", "pci:v00001002d0000150Ebc03sc00i00") + self.assertEqual(self.vendors(), ["amd"]) + + def test_pci_fallback_ignores_non_display_devices(self): + # An NVIDIA audio function on the same card is bc04, not bc03. + self.card(self.pci, "0000:01:00.1", "pci:v000010DEd000022BAbc04sc03i00") + self.assertEqual(self.vendors(), []) + + # ----------------------------------------------------------- error ---- + def test_no_gpu_anywhere_reports_nothing(self): + self.assertEqual(self.vendors(), []) + + def test_unreadable_modalias_is_skipped_not_fatal(self): + d = os.path.join(self.drm, "card0", "device") + os.makedirs(d) + os.mkdir(os.path.join(d, "modalias")) # a directory, not a file + self.assertEqual(self.vendors(), []) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/installer-steps/test_mark_volatile_configs.py b/tests/installer-steps/test_mark_volatile_configs.py new file mode 100644 index 0000000..27eac61 --- /dev/null +++ b/tests/installer-steps/test_mark_volatile_configs.py @@ -0,0 +1,216 @@ +"""Tests for mark_volatile_configs, the tail of the dotfiles install. + +Four stowed configs get rewritten in place by the apps that own them (btop, +qalculate, calibre, waypaper). The dotfiles repo handles that with git's +skip-worktree bit, applied by its own `skip-volatile` make target, which +`make stow` runs as its last step. + +archsetup stows inline rather than through that Makefile, and it never ran the +step -- so a machine installed by archsetup alone starts with a dotfiles repo +that goes dirty as soon as those apps run, and every later `git pull --ff-only` +trips over paths the user never edited. Both daily drivers carry the bits +today, but archsetup contains no code that sets them: they were applied by a +hand-run `make stow`. + +The fix calls the dotfiles Makefile's target rather than reimplementing the +logic here, so the volatile list stays in one place. That is what these tests +pin: that archsetup delegates, and that it degrades quietly when the checkout +it was handed cannot answer. + +These tests exercise the REAL function body, extracted from the `archsetup` +script at run time (not a copy), against a fixture git repo, with sudo stubbed +on PATH. + +Run from repo root: + python3 -m unittest tests.installer-steps.test_mark_volatile_configs +""" + +import os +import shutil +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") + +# The real target, copied from the dotfiles Makefile. The point of the fixture +# is that archsetup calls *a* skip-volatile target rather than doing the work +# itself; the body here mirrors the real one so the assertions are meaningful. +MAKEFILE = textwrap.dedent("""\ + DOTFILES := $(shell pwd) + VOLATILE := $(shell sed 's/#.*//' $(DOTFILES)/volatile-configs 2>/dev/null) + + skip-volatile: + \t@for f in $(VOLATILE); do \\ + \t\tif git -C $(DOTFILES) ls-files --error-unmatch "$$f" >/dev/null 2>&1; then \\ + \t\t\tgit -C $(DOTFILES) update-index --skip-worktree "$$f" && echo " skip-worktree: $$f"; \\ + \t\telse \\ + \t\t\techo " WARN: not tracked, skipping: $$f"; \\ + \t\tfi; \\ + \tdone +""") + +# sudo isn't available (and isn't wanted) in a unit test. Drop `-u ` and +# exec the rest, so the function's real command line is what runs. +SUDO_STUB = textwrap.dedent("""\ + #!/bin/bash + # Record the argv archsetup handed sudo, then drop `-u ` and exec the + # rest, so the function's real command line is what runs. + printf '%s\\n' "$*" >> "$SUDO_CALLS" + if [ "$1" = "-u" ]; then shift 2; fi + exec "$@" +""") + + +class MarkVolatileConfigsHarness(unittest.TestCase): + """Source mark_volatile_configs out of the real archsetup script.""" + + def setUp(self): + self.tmp = tempfile.mkdtemp(prefix="volatile-test-") + self.repo = os.path.join(self.tmp, "dotfiles") + os.mkdir(self.repo) + self.bindir = os.path.join(self.tmp, "bin") + os.mkdir(self.bindir) + sudo = os.path.join(self.bindir, "sudo") + with open(sudo, "w") as f: + f.write(SUDO_STUB) + os.chmod(sudo, 0o755) + self.logfile = os.path.join(self.tmp, "install.log") + self.sudo_calls = os.path.join(self.tmp, "sudo-calls") + 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 " + "'/^mark_volatile_configs() {/,/^}/p' \"$ARCHSETUP\")\n" + 'display() { echo "DISPLAY:$2"; }\n' + 'error_warn() { echo "WARN:$1"; return 1; }\n' + 'mark_volatile_configs "$REPO" testuser\n' + 'echo "RETURNED=$?"\n' + ) + os.chmod(self.wrapper, 0o755) + + def tearDown(self): + shutil.rmtree(self.tmp, ignore_errors=True) + + def git(self, *args): + return subprocess.run(["git", "-C", self.repo, *args], + capture_output=True, text=True) + + def make_repo(self, tracked=("common/.config/btop/btop.conf",), + volatile=None, makefile: "str | None" = MAKEFILE): + self.git("init", "-q") + self.git("config", "user.email", "t@example.com") + self.git("config", "user.name", "t") + for rel in tracked: + path = os.path.join(self.repo, rel) + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "w") as f: + f.write("original\n") + listed = tracked if volatile is None else volatile + with open(os.path.join(self.repo, "volatile-configs"), "w") as f: + f.write("# apps rewrite these in place\n") + f.write("".join(rel + "\n" for rel in listed)) + if makefile is not None: + with open(os.path.join(self.repo, "Makefile"), "w") as f: + f.write(makefile) + self.git("add", "-A") + self.git("commit", "-q", "-m", "fixture") + + def run_step(self): + env = dict(os.environ) + env["PATH"] = self.bindir + os.pathsep + env["PATH"] + env["REPO"] = self.repo + env["logfile"] = self.logfile + env["SUDO_CALLS"] = self.sudo_calls + return subprocess.run(["bash", self.wrapper, ARCHSETUP], + capture_output=True, text=True, env=env) + + def sudo_argv(self): + """The argument lines sudo was called with, one per invocation.""" + if not os.path.exists(self.sudo_calls): + return [] + with open(self.sudo_calls) as f: + return [line.rstrip("\n") for line in f if line.strip()] + + def skipped(self): + """Paths carrying the skip-worktree bit.""" + r = self.git("ls-files", "-v") + return sorted(line[2:] for line in r.stdout.splitlines() + if line.startswith("S ")) + + # ---------------------------------------------------------- normal ---- + def test_sets_skip_worktree_on_every_volatile_path(self): + paths = ["common/.config/btop/btop.conf", + "hyprland/.config/waypaper/config.ini"] + self.make_repo(tracked=paths) + self.assertEqual(self.skipped(), []) # the bit is not set by default + r = self.run_step() + self.assertIn("RETURNED=0", r.stdout, + "stdout=%r stderr=%r" % (r.stdout, r.stderr)) + self.assertEqual(self.skipped(), sorted(paths)) + + def test_runs_make_as_the_user_not_as_root(self): + # archsetup runs as root against a user-owned checkout. Letting root + # write .git/index leaves it root-owned, and the user's next git + # command then cannot update the index at all. + self.make_repo() + self.run_step() + self.assertEqual(self.sudo_argv(), + ["-u testuser make -C %s skip-volatile" % self.repo]) + + def test_leaves_non_volatile_tracked_files_alone(self): + self.make_repo(tracked=["common/.config/btop/btop.conf", + "common/.bashrc"], + volatile=["common/.config/btop/btop.conf"]) + self.run_step() + self.assertEqual(self.skipped(), ["common/.config/btop/btop.conf"]) + + # -------------------------------------------------------- boundary ---- + def test_no_makefile_is_a_quiet_no_op(self): + # DOTFILES_DIR can point at a checkout that predates the target, or at + # a fork. Nothing to delegate to is not an error. + self.make_repo(makefile=None) + r = self.run_step() + self.assertIn("RETURNED=0", r.stdout) + self.assertNotIn("WARN:", r.stdout) + self.assertEqual(self.skipped(), []) + + def test_empty_volatile_list_sets_nothing_and_does_not_warn(self): + self.make_repo(tracked=["common/.config/btop/btop.conf"], volatile=[]) + r = self.run_step() + self.assertIn("RETURNED=0", r.stdout) + self.assertNotIn("WARN:", r.stdout) + self.assertEqual(self.skipped(), []) + + def test_untracked_path_in_the_list_does_not_fail_the_step(self): + # The Makefile warns and carries on; a stale entry must not take the + # install down, and the tracked entries still get their bit. + self.make_repo(tracked=["common/.config/btop/btop.conf"], + volatile=["common/.config/btop/btop.conf", + "common/.config/gone/removed.conf"]) + r = self.run_step() + self.assertIn("RETURNED=0", r.stdout) + self.assertEqual(self.skipped(), ["common/.config/btop/btop.conf"]) + + # ----------------------------------------------------------- error ---- + def test_makefile_without_the_target_warns_and_returns(self): + self.make_repo(makefile="all:\n\t@true\n") + r = self.run_step() + self.assertIn("WARN:", r.stdout) + self.assertEqual(self.skipped(), []) + + def test_make_output_goes_to_the_logfile_not_the_console(self): + self.make_repo(tracked=["common/.config/btop/btop.conf"]) + r = self.run_step() + self.assertNotIn("skip-worktree:", r.stdout) + with open(self.logfile) as f: + self.assertIn("skip-worktree:", f.read()) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/installer-steps/test_orchestrators.py b/tests/installer-steps/test_orchestrators.py index 2a771ba..8de75cc 100644 --- a/tests/installer-steps/test_orchestrators.py +++ b/tests/installer-steps/test_orchestrators.py @@ -147,3 +147,61 @@ class MaintenanceConfigDispatch(unittest.TestCase): if __name__ == "__main__": unittest.main() + + +# Helper -> the function that must call it. Every one of these helpers has its +# own suite proving it behaves correctly; none of those suites proves it is +# wired in. Deleting the call left every test green in five separate suites, +# including the one guarding a `pacman -Rdd` of twelve firmware packages. +# +# This is a static check on purpose. The behavioural harness above stubs only +# the names it is given and runs the rest of the body for real, which is safe +# for an orchestrator whose body is nothing but calls, and emphatically not +# safe for a leaf like trim_firmware. Cheaper and machine-independent too: the +# alternative depends on what /proc/cpuinfo says on the box running the tests. +CALL_SITES = { + "trim_firmware": ["detect_gpu_vendors", "is_framework_intel_laptop"], + "finalize_dotfiles": ["mark_volatile_configs"], + "preflight_checks": ["select_locale", "nvidia_preflight", "check_disk_space"], + "configure_encrypted_autologin": ["configure_autologin"], + "configure_initramfs_hook": ["switch_udev_hook_to_systemd"], + "add_nvme_early_module": ["ensure_nvme_early_module"], + "configure_snapshots": ["configure_zfs_snapshots", "configure_btrfs_snapshots"], + "configure_zfs_snapshots": ["zfs_scrub_timer_units"], + "install_maintenance_config": ["install_maintenance_thresholds"], +} + + +def function_body(name): + """The body of `name` from archsetup, comments and blank lines stripped, so + a match is a real invocation rather than a mention in a comment.""" + with open(ARCHSETUP) as f: + src = f.read() + start = src.index(f"\n{name}() {{\n") + end = src.index("\n}\n", start) + lines = src[start:end].splitlines()[1:] + keep = [] + for line in lines: + stripped = line.strip() + if not stripped or stripped.startswith("#"): + continue + keep.append(line.split(" #", 1)[0]) + return "\n".join(keep) + + +class CallSiteWiring(unittest.TestCase): + def test_every_guarded_helper_is_actually_called(self): + for caller, callees in CALL_SITES.items(): + body = function_body(caller) + for callee in callees: + with self.subTest(caller=caller, callee=callee): + self.assertRegex( + body, rf"(^|[\s;&|(]){callee}\b", + f"{caller} no longer calls {callee}; its own suite " + f"cannot see this") + + def test_the_check_would_notice_a_missing_call(self): + # The check is only worth having if it fails when the call goes away. + body = function_body("trim_firmware").replace("detect_gpu_vendors", "") + with self.assertRaises(AssertionError): + self.assertRegex(body, r"(^|[\s;&|(])detect_gpu_vendors\b") diff --git a/tests/installer-steps/test_select_locale.py b/tests/installer-steps/test_select_locale.py new file mode 100644 index 0000000..b9b9ce7 --- /dev/null +++ b/tests/installer-steps/test_select_locale.py @@ -0,0 +1,160 @@ +"""Tests for select_locale in the archsetup installer. + +select_locale resolves the system locale during pre-flight. Three inputs, in +precedence order: an existing LANG= in locale.conf wins outright, then LOCALE +from the config file, then a numbered menu. + +That menu is why these tests exist. It carries its own "[1]" default, LOCALE is +optional in archsetup.conf.example, and archsetup does not require an archangel +install -- so an unattended run (--config-file) can reach the menu's bare read +with nobody there to answer it. configure_build_environment's own "no LANG= in +locale.conf" branch shows archsetup expects that state to occur. Unattended +therefore takes the menu's default, the same ruling nvidia_preflight and +configure_autologin make for their prompts. + +These tests exercise the REAL function body, extracted from the `archsetup` +script at run time (not a copy), against a fixture locale.conf. + +Run from repo root: + python3 -m unittest tests.installer-steps.test_select_locale +""" + +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") + + +class SelectLocaleHarness(unittest.TestCase): + """Source select_locale out of the real archsetup script.""" + + def setUp(self): + self.tmp = tempfile.mkdtemp(prefix="select-locale-test-") + self.conf = os.path.join(self.tmp, "locale.conf") + 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 " + "'/^select_locale() {/,/^}/p' \"$ARCHSETUP\")\n" + 'select_locale "$LOCALE_CONF"\n' + 'echo "RESOLVED=$locale"\n' + # Whatever the prompt did not consume is still on stdin. + 'if IFS= read -r leftover; then echo "LEFTOVER:$leftover"; ' + 'else echo "LEFTOVER:"; fi\n' + ) + os.chmod(self.wrapper, 0o755) + + def tearDown(self): + shutil.rmtree(self.tmp, ignore_errors=True) + + def write_conf(self, body): + with open(self.conf, "w") as f: + f.write(body) + + def run_select(self, locale="", config_file="", stdin=""): + env = dict(os.environ) + env["locale"] = locale + env["config_file"] = config_file + env["LOCALE_CONF"] = self.conf + return subprocess.run( + ["bash", self.wrapper, ARCHSETUP], + input=stdin, capture_output=True, text=True, env=env, + ) + + def resolved(self, r): + for line in r.stdout.splitlines(): + if line.startswith("RESOLVED="): + return line.split("=", 1)[1] + self.fail("no RESOLVED line; stdout=%r stderr=%r" % (r.stdout, r.stderr)) + + # ---------------------------------------------------------- normal ---- + def test_existing_locale_conf_wins_and_does_not_prompt(self): + self.write_conf("LANG=de_DE.UTF-8\n") + r = self.run_select(stdin="sentinel\n") + self.assertIn("[OK] Locale: de_DE.UTF-8", r.stdout) + # locale.conf is authoritative, so the global is deliberately left + # unset -- configure_build_environment reads LANG out of the file. + self.assertEqual(self.resolved(r), "") + self.assertIn("LEFTOVER:sentinel", r.stdout) + + def test_config_file_locale_used_when_conf_has_no_lang(self): + self.write_conf("") + r = self.run_select(locale="fr_FR.UTF-8", stdin="sentinel\n") + self.assertIn("(from config)", r.stdout) + self.assertEqual(self.resolved(r), "fr_FR.UTF-8") + self.assertIn("LEFTOVER:sentinel", r.stdout) + + def test_menu_choice_selects_that_locale(self): + self.write_conf("") + r = self.run_select(stdin="3\nsentinel\n") + self.assertEqual(self.resolved(r), "de_DE.UTF-8") + self.assertIn("LEFTOVER:sentinel", r.stdout) + + # -------------------------------------------------------- boundary ---- + def test_missing_locale_conf_falls_through_to_the_menu(self): + # No file at all, not just no LANG= line. + r = self.run_select(stdin="2\nsentinel\n") + self.assertEqual(self.resolved(r), "en_GB.UTF-8") + self.assertIn("LEFTOVER:sentinel", r.stdout) + + def test_bare_enter_takes_choice_one(self): + self.write_conf("") + r = self.run_select(stdin="\nsentinel\n") + self.assertEqual(self.resolved(r), "en_US.UTF-8") + self.assertIn("LEFTOVER:sentinel", r.stdout) + + def test_unattended_config_file_takes_the_default_without_reading(self): + # THE BUG. --config-file is the documented unattended mode and LOCALE + # is optional, so this is an ordinary unattended install on a machine + # whose locale.conf carries no LANG=. + self.write_conf("") + r = self.run_select(config_file="/etc/archsetup.conf", + stdin="sentinel\n") + self.assertEqual(self.resolved(r), "en_US.UTF-8") + self.assertIn("LEFTOVER:sentinel", r.stdout) + + def test_unattended_still_prefers_an_existing_locale_conf(self): + self.write_conf("LANG=ja_JP.UTF-8\n") + r = self.run_select(config_file="/etc/archsetup.conf", + stdin="sentinel\n") + self.assertIn("[OK] Locale: ja_JP.UTF-8", r.stdout) + self.assertIn("LEFTOVER:sentinel", r.stdout) + + def test_unattended_still_prefers_the_configured_locale(self): + # Unattended does not mean "always en_US" -- LOCALE is an answer, and + # the menu default must not override it. + self.write_conf("") + r = self.run_select(locale="pt_BR.UTF-8", + config_file="/etc/archsetup.conf", stdin="sentinel\n") + self.assertEqual(self.resolved(r), "pt_BR.UTF-8") + self.assertIn("LEFTOVER:sentinel", r.stdout) + + # ----------------------------------------------------------- error ---- + def test_unrecognised_choice_falls_back_to_en_us(self): + self.write_conf("") + r = self.run_select(stdin="42\nsentinel\n") + self.assertEqual(self.resolved(r), "en_US.UTF-8") + self.assertIn("LEFTOVER:sentinel", r.stdout) + + def test_manual_entry_reads_a_second_line(self): + self.write_conf("") + r = self.run_select(stdin="9\nnl_NL.UTF-8\nsentinel\n") + self.assertEqual(self.resolved(r), "nl_NL.UTF-8") + self.assertIn("LEFTOVER:sentinel", r.stdout) + + def test_manual_entry_left_empty_falls_back_to_en_us(self): + self.write_conf("") + r = self.run_select(stdin="9\n\nsentinel\n") + self.assertEqual(self.resolved(r), "en_US.UTF-8") + self.assertIn("LEFTOVER:sentinel", r.stdout) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/installer-steps/test_switch_udev_hook.py b/tests/installer-steps/test_switch_udev_hook.py new file mode 100644 index 0000000..54969e6 --- /dev/null +++ b/tests/installer-steps/test_switch_udev_hook.py @@ -0,0 +1,178 @@ +"""Test the udev -> systemd initramfs hook swap. + +configure_initramfs_hook swaps the `udev` initramfs hook for `systemd` so fsck +output routes through systemd. mkinitcpio's `systemd` hook symlinks /init to +systemd, which never executes the ash `run_hook` scripts the busybox init +drives -- so any busybox-only hook left in HOOKS is installed into the image +and then never runs. + +`encrypt` is the boot-critical member of that set: it is what unlocks a LUKS +root, and its `cryptdevice=` cmdline parameter (which merge_grub_cmdline +already treats as boot-critical) is read by nothing else. mkinitcpio has no +conflict check, so `mkinitcpio -P` succeeds, archsetup reports success, and the +machine fails to unlock its root at the next boot. The systemd equivalent is +`sd-encrypt` driven by `rd.luks.*`, and converting between the two means +rewriting the kernel cmdline against the volume's UUID -- so the step refuses +the swap rather than attempting a migration. + +Method: sed-extract switch_udev_hook_to_systemd + hooks_need_busybox_init from +the real `archsetup`, point them at a temp mkinitcpio.conf, and fake +backup_system_file/display/error_warn. + +Run from repo root: + python3 -m unittest tests.installer-steps.test_switch_udev_hook +""" + +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") + +# The two HOOKS lines Arch's own mkinitcpio.conf documents for an encrypted +# root -- the busybox pairing and the systemd pairing. They are the fixtures +# that matter, because the bug is turning the first into neither. +BUSYBOX_ENCRYPT = ( + "HOOKS=(base udev microcode modconf keyboard keymap consolefont block " + "mdadm_udev encrypt filesystems fsck)\n" +) +SYSTEMD_ENCRYPT = ( + "HOOKS=(base systemd autodetect microcode modconf kms keyboard sd-vconsole " + "sd-encrypt block filesystems fsck)\n" +) +PLAIN_UDEV = "HOOKS=(base udev autodetect microcode modconf block filesystems fsck)\n" + + +def run(conf_body, missing_conf=False): + with tempfile.TemporaryDirectory() as d: + conf = os.path.join(d, "mkinitcpio.conf") + if not missing_conf: + with open(conf, "w") as f: + f.write(conf_body) + script = textwrap.dedent(f"""\ + logfile=/dev/null + action="" + display() {{ echo "DISPLAY: $2"; }} + backup_system_file() {{ echo "BACKUP: $1" >> "{d}/backup.log"; }} + error_warn() {{ echo "WARN: $1"; return 1; }} + source <(sed -n '/^hooks_need_busybox_init() {{/,/^}}/p;/^switch_udev_hook_to_systemd() {{/,/^}}/p' "{ARCHSETUP}") + switch_udev_hook_to_systemd "{conf}" + echo "RC=$?" + echo "CONF:[$(cat "{conf}" 2>/dev/null)]" + [ -f "{d}/backup.log" ] && cat "{d}/backup.log" + exit 0 + """) + return subprocess.run( + ["bash", "-c", script], capture_output=True, text=True, timeout=10, + ) + + +def rc_of(r): + """The step's exact return code. Substring-matching "RC=1" in stdout would + also match "RC=127" (the code bash gives a missing function), so a broken + extraction would read as a passing test.""" + 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 SwitchUdevHookToSystemd(unittest.TestCase): + # ------------------------------------------------------------ normal ---- + def test_plain_udev_is_swapped_and_reports_change(self): + r = run(PLAIN_UDEV) + self.assertIn("HOOKS=(base systemd autodetect", r.stdout) + self.assertNotIn("udev", r.stdout.split("CONF:[")[1]) + self.assertEqual(rc_of(r), 0, "a conf that changed must report success") + self.assertIn("BACKUP:", r.stdout, "the conf must be backed up before the edit") + + def test_already_systemd_is_left_alone_and_reports_no_change(self): + body = "HOOKS=(base systemd autodetect block filesystems fsck)\n" + r = run(body) + self.assertIn(body.strip(), r.stdout) + self.assertEqual(rc_of(r), 1, + "an unchanged conf must not report a change (no needless rebuild)") + + # ---------------------------------------------------------- boundary ---- + def test_busybox_encrypt_hook_blocks_the_swap(self): + # THE BUG: swapping here leaves `encrypt` under a systemd init that + # never runs it, so the LUKS root never unlocks and the machine will + # not boot. + r = run(BUSYBOX_ENCRYPT) + self.assertIn(" udev ", r.stdout, "udev must survive next to a busybox encrypt hook") + self.assertIn(" encrypt ", r.stdout) + self.assertNotIn("systemd", r.stdout.split("CONF:[")[1]) + self.assertEqual(rc_of(r), 1) + self.assertNotIn("BACKUP:", r.stdout, "a declined swap must not touch the file") + + def test_tab_separated_encrypt_hook_blocks_the_swap(self): + # The token class was [( ] / [ )], which is a literal space and does + # not include a tab. mkinitcpio's HOOKS is a bash array literal, and + # bash word-splits on IFS -- tabs included -- so a tab-separated HOOKS + # line is valid, loads correctly, and slipped straight past the guard + # into the exact unbootable state it exists to prevent. + r = run("HOOKS=(base udev block\tencrypt\tfilesystems fsck)\n") + self.assertEqual(rc_of(r), 1, "a tab-separated encrypt must block the swap") + self.assertIn("encrypt", r.stdout) + self.assertNotIn("systemd", r.stdout.split("CONF:[")[1]) + self.assertNotIn("BACKUP:", r.stdout) + + def test_tab_separated_sd_encrypt_is_still_not_the_busybox_hook(self): + # Widening the class must not widen what counts as the busybox hook: + # sd-encrypt IS the systemd one and has to keep swapping. + r = run("HOOKS=(base\tsystemd\tsd-encrypt\tfilesystems)\n") + self.assertNotIn("BACKUP:", r.stdout) # already systemd, nothing to do + self.assertEqual(rc_of(r), 1) + + def test_tab_separated_udev_without_encrypt_still_swaps(self): + r = run("HOOKS=(base\tudev\tblock\tfilesystems\tfsck)\n") + self.assertEqual(rc_of(r), 0, "a tab-separated conf with no encrypt must swap") + self.assertIn("systemd", r.stdout.split("CONF:[")[1]) + + def test_plymouth_encrypt_hook_blocks_the_swap(self): + r = run("HOOKS=(base udev block plymouth-encrypt filesystems)\n") + self.assertIn(" udev ", r.stdout) + self.assertEqual(rc_of(r), 1) + + def test_sd_encrypt_is_not_mistaken_for_the_busybox_hook(self): + # sd-encrypt IS the systemd hook -- it must not be read as a reason to + # refuse. (No udev here, so the swap is a no-op either way; what this + # pins is that the substring match does not over-trigger.) + r = run(SYSTEMD_ENCRYPT) + self.assertIn("sd-encrypt", r.stdout) + self.assertNotIn("HOOKS carries encrypt", r.stdout) + + def test_encrypt_as_last_token_is_detected(self): + r = run("HOOKS=(base udev block encrypt)\n") + self.assertIn(" udev ", r.stdout) + self.assertEqual(rc_of(r), 1) + + def test_encrypt_as_first_token_is_detected(self): + r = run("HOOKS=(encrypt udev block)\n") + self.assertIn("(encrypt udev block)", r.stdout) + self.assertEqual(rc_of(r), 1) + + def test_only_the_hooks_line_is_rewritten(self): + # A commented example line mentioning udev must survive untouched. + body = "# HOOKS=(base udev block filesystems fsck)\n" + PLAIN_UDEV + r = run(body) + self.assertIn("# HOOKS=(base udev block filesystems fsck)", r.stdout) + self.assertIn("HOOKS=(base systemd autodetect", r.stdout) + + # ------------------------------------------------------------- error ---- + def test_conf_without_a_hooks_line_is_a_no_op(self): + r = run("MODULES=(nvme)\n") + self.assertEqual(rc_of(r), 1) + self.assertIn("MODULES=(nvme)", r.stdout) + + def test_missing_conf_is_a_no_op_not_a_crash(self): + r = run("", missing_conf=True) + self.assertEqual(rc_of(r), 1) + self.assertEqual(r.returncode, 0) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/nvidia-preflight/test_nvidia_preflight_gate.py b/tests/nvidia-preflight/test_nvidia_preflight_gate.py new file mode 100644 index 0000000..1d76689 --- /dev/null +++ b/tests/nvidia-preflight/test_nvidia_preflight_gate.py @@ -0,0 +1,173 @@ +"""Tests for the nvidia_preflight policy wrapper in the archsetup installer. + +nvidia_preflight_report (covered by test_nvidia_preflight.py) is the pure +detection core. nvidia_preflight is the policy around it: it decides whether +the check applies at all, runs the report, aborts on a too-old driver, and +confirms an otherwise-supported card with the user. + +The check is Wayland-specific -- every line it prints names Wayland/Hyprland, +and its own failure hint tells the user to install with DESKTOP_ENV=dwm +instead. So it must not fire on a dwm (X11) or headless (none) install, and it +must not fire when --no-gpu-drivers / NO_GPU_DRIVERS=yes says the user handles +the driver themselves -- the same ruling install_gpu_drivers already makes. + +These tests exercise the REAL function body, extracted from the `archsetup` +script at run time (not a copy), with nvidia_preflight_report stubbed to a +chosen verdict. + +Run from repo root: + python3 -m unittest tests.nvidia-preflight.test_nvidia_preflight_gate +""" + +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_preflight_report's documented return codes. +RC_NO_NVIDIA = 0 +RC_NVIDIA_OK = 10 +RC_NVIDIA_TOO_OLD = 11 + + +class NvidiaPreflightGateHarness(unittest.TestCase): + """Source nvidia_preflight out of the real archsetup script.""" + + def setUp(self): + self.tmp = tempfile.mkdtemp(prefix="nvidia-gate-test-") + self.marker = os.path.join(self.tmp, "report-calls") + 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() {/,/^}/p' \"$ARCHSETUP\")\n" + # Stub the detection core: record the call, answer with $STUB_RC. + "nvidia_preflight_report() {\n" + ' echo "called" >> "$MARKER"\n' + ' return "${STUB_RC:-0}"\n' + "}\n" + "nvidia_preflight\n" + 'echo "PREFLIGHT_RETURNED"\n' + # Whatever the prompt did not consume is still on stdin. + 'if IFS= read -r leftover; then echo "LEFTOVER:$leftover"; ' + 'else echo "LEFTOVER:"; fi\n' + ) + os.chmod(self.wrapper, 0o755) + + def tearDown(self): + shutil.rmtree(self.tmp, ignore_errors=True) + + def run_gate(self, rc, desktop_env="hyprland", skip_gpu_drivers="false", + config_file="", stdin=""): + env = dict(os.environ) + env["MARKER"] = self.marker + env["STUB_RC"] = str(rc) + env["desktop_env"] = desktop_env + env["skip_gpu_drivers"] = skip_gpu_drivers + env["config_file"] = config_file + return subprocess.run( + ["bash", self.wrapper, ARCHSETUP], + input=stdin, capture_output=True, text=True, env=env, + ) + + def report_calls(self): + if not os.path.exists(self.marker): + return 0 + with open(self.marker) as f: + return len([line for line in f if line.strip()]) + + def assertContinued(self, r): + self.assertIn("PREFLIGHT_RETURNED", r.stdout, + "preflight aborted; stdout=%r stderr=%r" % (r.stdout, r.stderr)) + self.assertEqual(r.returncode, 0) + + def assertAborted(self, r): + self.assertNotIn("PREFLIGHT_RETURNED", r.stdout) + self.assertEqual(r.returncode, 1) + + # ---------------------------------------------------------- normal ---- + def test_hyprland_no_nvidia_continues_without_prompting(self): + r = self.run_gate(RC_NO_NVIDIA, stdin="sentinel\n") + self.assertContinued(r) + self.assertEqual(self.report_calls(), 1) + self.assertIn("LEFTOVER:sentinel", r.stdout) + + def test_hyprland_supported_driver_prompts_and_continues_on_yes(self): + # bash emits `read -p`'s prompt only when stdin is a terminal, so the + # prompt text is unassertable here. What is assertable -- and is the + # thing that matters -- is that the read consumed exactly the answer + # line and left the rest of stdin alone. + r = self.run_gate(RC_NVIDIA_OK, stdin="y\nsentinel\n") + self.assertContinued(r) + self.assertIn("Continuing on NVIDIA", r.stdout) + self.assertIn("LEFTOVER:sentinel", r.stdout) + + def test_hyprland_old_driver_still_aborts(self): + # The check must keep working where it actually applies. + r = self.run_gate(RC_NVIDIA_TOO_OLD) + self.assertAborted(r) + self.assertEqual(self.report_calls(), 1) + + # -------------------------------------------------------- boundary ---- + def test_dwm_old_driver_does_not_abort(self): + # An X11 install does not care about the Wayland driver floor -- and + # the abort's own hint tells the user to install with DESKTOP_ENV=dwm. + r = self.run_gate(RC_NVIDIA_TOO_OLD, desktop_env="dwm") + self.assertContinued(r) + self.assertEqual(self.report_calls(), 0) + + def test_none_old_driver_does_not_abort(self): + # A headless install never runs a compositor at all. + r = self.run_gate(RC_NVIDIA_TOO_OLD, desktop_env="none") + self.assertContinued(r) + self.assertEqual(self.report_calls(), 0) + + def test_skip_gpu_drivers_old_driver_does_not_abort(self): + # --no-gpu-drivers means the user installs the driver themselves, so + # the repo's candidate version is not archsetup's call to veto. + r = self.run_gate(RC_NVIDIA_TOO_OLD, skip_gpu_drivers="true") + self.assertContinued(r) + self.assertEqual(self.report_calls(), 0) + + def test_dwm_supported_driver_does_not_prompt(self): + r = self.run_gate(RC_NVIDIA_OK, desktop_env="dwm", stdin="sentinel\n") + self.assertContinued(r) + self.assertIn("LEFTOVER:sentinel", r.stdout) + + def test_unattended_config_file_answers_the_prompt_itself(self): + # --config-file is the documented unattended mode. The rc-10 prompt is + # advisory (the driver requirement is met), so it must not block on a + # read -- the same failure aur_install's --answerdiff/--answerclean + # comment already rules out. + r = self.run_gate(RC_NVIDIA_OK, config_file="/etc/archsetup.conf", + stdin="sentinel\n") + self.assertContinued(r) + self.assertIn("Continuing on NVIDIA", r.stdout) + self.assertIn("LEFTOVER:sentinel", r.stdout) + + # ----------------------------------------------------------- error ---- + def test_hyprland_supported_driver_aborts_on_no(self): + r = self.run_gate(RC_NVIDIA_OK, stdin="n\n") + self.assertAborted(r) + self.assertIn("Aborted at NVIDIA preflight.", r.stdout) + + def test_hyprland_supported_driver_aborts_on_capital_no(self): + r = self.run_gate(RC_NVIDIA_OK, stdin="No\n") + self.assertAborted(r) + + def test_unattended_old_driver_still_aborts(self): + # Unattended does not mean "never fail" -- a driver below the floor is + # still a hard stop on a hyprland install. + r = self.run_gate(RC_NVIDIA_TOO_OLD, config_file="/etc/archsetup.conf") + self.assertAborted(r) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/wipedisk/test_wipedisk.py b/tests/wipedisk/test_wipedisk.py new file mode 100644 index 0000000..95c7df8 --- /dev/null +++ b/tests/wipedisk/test_wipedisk.py @@ -0,0 +1,273 @@ +"""Test scripts/wipedisk, the manual disk-erase helper. + +Three defects, all of which make the script's final word untrue: + +1. `sgdisk --zap-all` had its result discarded and the script printed + "Disk erased." and exited 0 unconditionally. sgdisk fails on a busy device + -- a disk with a mounted filesystem or an active md/LVM/ZFS holder, which is + exactly the case a user hits when they pick the wrong disk -- so the tool + claimed an erase it had not performed. + +2. The prompt says "Select the disk id to use", and then listed every entry in + /dev/disk/by-id, two thirds of which are partitions (18 entries on this + machine, 12 of them -partN). The menu promised disks and offered partitions. + +3. "Disk erased." overstates what the tool does. `sgdisk --zap-all` destroys + partition tables, not data, and `blkdiscard -f || true` deliberately + tolerates a device that cannot discard. On a disk without discard support + the script erased the partition table and every byte of data remained + readable -- while telling the user the disk was erased. + +Method: run the REAL script with WIPEDISK_BY_ID pointed at a fixture directory +and fake blkdiscard/sgdisk on PATH, feeding the select menu and the confirm +prompt on stdin. The env-overridable path follows nvidia_preflight_report's +NVIDIA_DRM_GLOB. + +Run from repo root: + python3 -m unittest tests.wipedisk.test_wipedisk +""" + +import os +import shutil +import stat +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", "wipedisk") + +# A realistic by-id listing: two whole disks, each with two partitions, plus +# the eui alias the kernel also publishes. +# Names that cannot collide with a real device. The first cut of this suite +# copied this machine's actual by-id entries, so a test that lost the +# WIPEDISK_BY_ID seam would have driven a menu of real disks and still passed. +DISKS = [ + "wipedisk-fixture-disk-A", + "wipedisk-fixture-disk-B", +] +PARTS = [d + p for d in DISKS for p in ("-part1", "-part2")] + + +class Wipedisk(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.mkdtemp(prefix="wipedisk-test-") + self.bin = os.path.join(self.tmp, "bin") + self.byid = os.path.join(self.tmp, "by-id") + os.makedirs(self.bin) + os.makedirs(self.byid) + for name in DISKS + PARTS: + open(os.path.join(self.byid, name), "w").close() + + def tearDown(self): + shutil.rmtree(self.tmp, ignore_errors=True) + + def fake(self, name, body): + path = os.path.join(self.bin, name) + with open(path, "w") as f: + f.write("#!/bin/bash\n" + body) + os.chmod(path, os.stat(path).st_mode | stat.S_IEXEC | stat.S_IXGRP | stat.S_IXOTH) + + def install_fakes(self, discard_ok=True, sgdisk_ok=True, + discard_err="blkdiscard: BLKDISCARD ioctl failed: " + "Operation not supported"): + log = os.path.join(self.tmp, "calls.log") + self.fake("blkdiscard", + 'echo "blkdiscard $*" >> "%s"\n' + '[ %d -eq 0 ] || echo "%s" >&2\nexit %d\n' + % (log, 0 if discard_ok else 1, discard_err, + 0 if discard_ok else 1)) + self.fake("sgdisk", 'echo "sgdisk $*" >> "%s"\nexit %d\n' + % (log, 0 if sgdisk_ok else 1)) + self.calls_log = log + + def calls(self): + if not os.path.exists(self.calls_log): + return "" + with open(self.calls_log) as f: + return f.read() + + def run_script(self, stdin): + env = dict(os.environ) + env["PATH"] = self.bin + os.pathsep + env["PATH"] + env["WIPEDISK_BY_ID"] = self.byid + return subprocess.run( + ["bash", SCRIPT], input=stdin, capture_output=True, text=True, + env=env, timeout=20, + ) + + # ---------------------------------------------------------- normal ---- + def test_successful_wipe_exits_zero_and_calls_both_tools(self): + self.install_fakes() + r = self.run_script("1\ny\n") + self.assertEqual(r.returncode, 0, r.stdout + r.stderr) + # Assert the fixture PATH, not just the tool name: this is what + # catches a lost WIPEDISK_BY_ID seam driving real devices. + self.assertIn("blkdiscard %s/%s" % (self.byid, DISKS[0]), self.calls()) + self.assertIn("sgdisk --zap-all %s/%s" % (self.byid, DISKS[0]), + self.calls()) + + def test_declining_the_confirmation_touches_nothing(self): + self.install_fakes() + r = self.run_script("1\nn\n") + self.assertNotEqual(r.returncode, 0) + self.assertEqual(self.calls(), "", + "a declined confirmation must not run either tool") + + # -------------------------------------------------------- boundary ---- + def test_menu_offers_whole_disks_only(self): + # The prompt says "Select the disk id"; partitions are not disks. + self.install_fakes() + r = self.run_script("1\nn\n") + menu = r.stdout + r.stderr + for part in PARTS: + self.assertNotIn(part, menu, + "the disk menu must not offer partition %s" % part) + for disk in DISKS: + self.assertIn(disk, menu) + + def test_discard_unsupported_does_not_claim_the_data_is_gone(self): + # blkdiscard failing is tolerated by design (not every device can + # discard) -- but then only the partition table went, and the message + # must not tell the user the disk was erased. + self.install_fakes(discard_ok=False) + r = self.run_script("1\ny\n") + out = r.stdout + r.stderr + self.assertIn("sgdisk", self.calls(), + "a failed discard must not stop the partition-table zap") + self.assertNotIn("Disk erased.", out, + "data was not erased -- only the partition table was") + + # ----------------------------------------------------------- error ---- + def test_failed_zap_exits_non_zero(self): + # sgdisk fails on a busy device. Claiming success here is the defect. + self.install_fakes(sgdisk_ok=False) + r = self.run_script("1\ny\n") + self.assertNotEqual(r.returncode, 0, + "a failed zap must not report success") + self.assertNotIn("Disk erased.", r.stdout + r.stderr) + + def test_empty_by_id_directory_says_so(self): + # Without the guard this still exits non-zero and runs nothing -- the + # empty select menu falls through to the confirm prompt, which reads + # EOF and declines. So asserting only on the exit code and the call log + # is not a gate; it passes with the guard deleted. What the guard + # actually buys is saying why, instead of an empty menu the user has to + # break out of. + self.install_fakes() + for name in os.listdir(self.byid): + os.unlink(os.path.join(self.byid, name)) + r = self.run_script("1\ny\n") + self.assertNotEqual(r.returncode, 0) + self.assertEqual(self.calls(), "", + "no disks means nothing to erase, not an unguarded run") + self.assertIn("No disks found", r.stdout + r.stderr, + "an empty device list must be reported, not shown as an empty menu") + + +if __name__ == "__main__": + unittest.main() + + +class WipediskBusyDisk(unittest.TestCase): + """The wrong-disk case, which is the whole reason this tool checks anything. + + blkdiscard opens the device O_EXCL by default (util-linux >= 2.36) and + refuses a disk something is holding -- a mounted filesystem, an md member, + an LVM PV. That refusal is the safety gate, so it has to come first. Forcing + past it with -f destroyed a live filesystem and only then let sgdisk fail, + leaving the user told that nothing had happened and to try again. + + A discard that fails because the hardware cannot discard is a different + answer entirely, and must not stop the run. + """ + + def setUp(self): + self.tmp = tempfile.mkdtemp(prefix="wipedisk-busy-test-") + self.bin = os.path.join(self.tmp, "bin") + self.byid = os.path.join(self.tmp, "by-id") + os.makedirs(self.bin) + os.makedirs(self.byid) + for name in DISKS + PARTS: + open(os.path.join(self.byid, name), "w").close() + self.log = os.path.join(self.tmp, "calls.log") + + def tearDown(self): + shutil.rmtree(self.tmp, ignore_errors=True) + + def fake(self, name, body): + path = os.path.join(self.bin, name) + with open(path, "w") as f: + f.write("#!/bin/bash\n" + body) + os.chmod(path, 0o755) + + def install(self, discard_rc, discard_err, sgdisk_rc=0): + self.fake("blkdiscard", + 'echo "blkdiscard $*" >> "%s"\n' + '[ %d -eq 0 ] || echo "%s" >&2\nexit %d\n' + % (self.log, discard_rc, discard_err, discard_rc)) + self.fake("sgdisk", 'echo "sgdisk $*" >> "%s"\nexit %d\n' + % (self.log, sgdisk_rc)) + + def calls(self): + if not os.path.exists(self.log): + return "" + with open(self.log) as f: + return f.read() + + def run_script(self, stdin="1\ny\n"): + env = dict(os.environ) + env["PATH"] = self.bin + os.pathsep + env["PATH"] + env["WIPEDISK_BY_ID"] = self.byid + return subprocess.run(["bash", SCRIPT], input=stdin, + capture_output=True, text=True, env=env, + timeout=20) + + # ---------------------------------------------------------- normal ---- + def test_blkdiscard_is_not_forced_past_the_exclusive_open(self): + # -f is what turned "refuse a busy disk" into "destroy it anyway". + self.install(0, "") + self.run_script() + lines = [l for l in self.calls().splitlines() + if l.startswith("blkdiscard ")] + self.assertEqual(len(lines), 1, self.calls()) + # Exact argv tokens: "-f" is a substring of the fixture disk name, so + # assertNotIn would pass on the fixture path and prove nothing. + args = lines[0].split()[1:] + self.assertNotIn("-f", args) + self.assertNotIn("--force", args) + + # -------------------------------------------------------- boundary ---- + def test_a_device_that_cannot_discard_still_gets_its_table_cleared(self): + self.install(1, "blkdiscard: BLKDISCARD ioctl failed: " + "Operation not supported") + r = self.run_script() + self.assertEqual(r.returncode, 0, r.stdout + r.stderr) + self.assertIn("sgdisk", self.calls()) + self.assertIn("still", r.stdout) # the honest data-still-present note + + # ----------------------------------------------------------- error ---- + def test_a_busy_disk_stops_before_sgdisk_is_ever_called(self): + self.install(1, "blkdiscard: /dev/sda: BLKDISCARD ioctl failed: " + "Device or resource busy") + r = self.run_script() + self.assertNotEqual(r.returncode, 0) + self.assertNotIn("sgdisk", self.calls(), + "a refused discard must not fall through to sgdisk") + + def test_a_busy_disk_says_nothing_was_erased(self): + self.install(1, "blkdiscard: /dev/sda: BLKDISCARD ioctl failed: " + "Device or resource busy") + r = self.run_script() + out = r.stdout + r.stderr + self.assertIn("Nothing was erased", out) + self.assertNotIn("Disk erased", out) + + def test_a_busy_disk_never_reports_a_cleared_partition_table(self): + # The defect this class exists for: the old order discarded first and + # then said "could not clear the partition table ... run this again", + # which reads as "nothing happened" after data was already gone. + self.install(1, "blkdiscard: /dev/sda: BLKDISCARD ioctl failed: " + "Device or resource busy") + r = self.run_script() + self.assertNotIn("Partition table cleared", r.stdout + r.stderr) diff --git a/tests/zfs-replicate/test_zfs_replicate.py b/tests/zfs-replicate/test_zfs_replicate.py new file mode 100644 index 0000000..f316ce6 --- /dev/null +++ b/tests/zfs-replicate/test_zfs_replicate.py @@ -0,0 +1,174 @@ +"""Test scripts/zfs-replicate, the nightly ZFS replication to TrueNAS. + +Two defects the script had, both of which make a failing backup look like a +working one. It runs as a systemd oneshot (zfs-replicate.service) on a nightly +timer, so its exit code and its journal output are the only signals anyone ever +sees. + +1. The full-replication loop caught each syncoid failure, warned, and carried + on -- then printed "Replication complete." and exited 0 regardless. With + every dataset failing, systemd recorded a clean success. A backup that had + not run for months was indistinguishable from one that had. + +2. determine_host runs inside a command substitution, and its error() wrote to + stdout. So on an unreachable TrueNAS the message was captured into + TRUENAS_HOST and discarded, and `set -e` killed the script with exit 1 and + no output whatsoever -- a nightly service failing silently with nothing in + the journal to say why. + +Both are fixed by sending every diagnostic to stderr (which also keeps the +command substitution's stdout clean, so only the hostname can land in +TRUENAS_HOST) and by exiting non-zero when any dataset failed. + +Method: run the REAL script with a fake syncoid and a fake ping on PATH, in the +style of tests/zfs-pre-snapshot's fake-zfs. + +Run from repo root: + python3 -m unittest tests.zfs-replicate.test_zfs_replicate +""" + +import os +import shutil +import stat +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", "zfs-replicate") + +# The four datasets the script replicates, in order. +DATASETS = ["zroot/ROOT/default", "zroot/home", "zroot/media", "zroot/vms"] + + +class ZfsReplicate(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.mkdtemp(prefix="zfs-replicate-test-") + self.bin = os.path.join(self.tmp, "bin") + os.makedirs(self.bin) + + def tearDown(self): + shutil.rmtree(self.tmp, ignore_errors=True) + + def fake(self, name, body): + path = os.path.join(self.bin, name) + with open(path, "w") as f: + f.write("#!/bin/bash\n" + body) + os.chmod(path, os.stat(path).st_mode | stat.S_IEXEC | stat.S_IXGRP | stat.S_IXOTH) + + def install_fakes(self, reachable="local", fail_datasets=(), no_syncoid=False): + """ping answers for the named host only; syncoid fails for the listed + datasets (the last argument-but-one is the source dataset).""" + if reachable == "local": + ping = 'case "$3" in truenas.local) exit 0 ;; *) exit 1 ;; esac\n' + elif reachable == "tailscale": + ping = 'case "$3" in truenas) exit 0 ;; *) exit 1 ;; esac\n' + else: + ping = "exit 1\n" + # `ping -c 1 -W 2 ` puts the host in $5, but be permissive and + # scan every argument so an option-order change does not silently make + # every host unreachable (which would make these tests pass for the + # wrong reason). + ping = 'for a in "$@"; do case "$a" in\n' + \ + (" truenas.local) exit 0 ;;\n" if reachable == "local" else "") + \ + (" truenas) exit 0 ;;\n" if reachable == "tailscale" else "") + \ + "esac; done\nexit 1\n" + self.fake("ping", ping) + if not no_syncoid: + arms = "".join( + ' %s) echo "fake syncoid failing for %s" >&2; exit 1 ;;\n' % (d, d) + for d in fail_datasets + ) + self.fake("syncoid", ( + 'for a in "$@"; do case "$a" in\n' + arms + "esac; done\nexit 0\n" + )) + + def run_script(self, *args): + env = dict(os.environ) + # Front of PATH so the fakes win; keep the rest so bash/coreutils work. + env["PATH"] = self.bin + os.pathsep + env["PATH"] + return subprocess.run( + ["bash", SCRIPT, *args], + capture_output=True, text=True, env=env, timeout=20, + ) + + # ---------------------------------------------------------- normal ---- + def test_all_datasets_succeed_exits_zero(self): + self.install_fakes() + r = self.run_script() + self.assertEqual(r.returncode, 0, r.stdout + r.stderr) + self.assertIn("Replication complete", r.stdout + r.stderr) + + def test_single_dataset_mode_succeeds(self): + self.install_fakes() + r = self.run_script("zroot/home") + self.assertEqual(r.returncode, 0, r.stdout + r.stderr) + + def test_falls_back_to_tailscale_when_local_is_unreachable(self): + self.install_fakes(reachable="tailscale") + r = self.run_script() + self.assertEqual(r.returncode, 0, r.stdout + r.stderr) + out = r.stdout + r.stderr + self.assertIn("Using TrueNAS host: truenas", out) + self.assertNotIn("truenas.local", out) + + # -------------------------------------------------------- boundary ---- + def test_every_dataset_failing_exits_non_zero(self): + # THE BUG: this used to print "Replication complete." and exit 0, so a + # nightly oneshot recorded success while backing nothing up. + self.install_fakes(fail_datasets=DATASETS) + r = self.run_script() + self.assertNotEqual(r.returncode, 0, + "a run where every dataset failed must not report success") + + def test_one_dataset_failing_exits_non_zero_and_still_tries_the_rest(self): + self.install_fakes(fail_datasets=["zroot/media"]) + r = self.run_script() + self.assertNotEqual(r.returncode, 0) + out = r.stdout + r.stderr + # The failure must not abort the run -- the dataset after it still goes. + self.assertIn("zroot/vms", out, + "a mid-loop failure must not stop the remaining datasets") + + def test_failure_count_is_reported(self): + self.install_fakes(fail_datasets=["zroot/home", "zroot/vms"]) + r = self.run_script() + self.assertIn("2", (r.stdout + r.stderr).split("complete")[-1] or "", + "the summary must say how many datasets failed") + + # ----------------------------------------------------------- error ---- + def test_unreachable_truenas_reports_why(self): + # THE OTHER BUG: error() wrote to stdout inside a command substitution, + # so this exited 1 with completely empty output. + self.install_fakes(reachable="none") + r = self.run_script() + self.assertNotEqual(r.returncode, 0) + self.assertIn("Cannot reach TrueNAS", r.stdout + r.stderr, + "the reason must reach the journal, not the command substitution") + + def test_missing_syncoid_reports_why(self): + self.install_fakes(no_syncoid=True) + # PATH is the fake bin only, so `command -v syncoid` fails whether or + # not this machine has a real syncoid. bash must then be invoked by + # absolute path -- a PATH that cannot find syncoid cannot find bash + # either, and subprocess would fail to launch instead of testing + # anything. The script's syncoid check runs before any PATH lookup it + # actually needs, so nothing else breaks. + env = dict(os.environ) + env["PATH"] = self.bin + r = subprocess.run([shutil.which("bash") or "/usr/bin/bash", SCRIPT], + capture_output=True, text=True, env=env, timeout=20) + self.assertNotEqual(r.returncode, 0) + self.assertIn("syncoid not found", r.stdout + r.stderr) + + def test_diagnostics_do_not_land_on_stdout_of_the_host_probe(self): + # Whatever determine_host prints on stdout becomes TRUENAS_HOST, so a + # stray diagnostic there corrupts every destination path. + self.install_fakes() + r = self.run_script() + self.assertIn("Using TrueNAS host: truenas.local", r.stdout + r.stderr) + self.assertNotIn("[INFO] Using TrueNAS host: [", r.stdout + r.stderr) + + +if __name__ == "__main__": + unittest.main() -- cgit v1.2.3