diff options
Diffstat (limited to 'tests')
21 files changed, 1437 insertions, 18 deletions
diff --git a/tests/cmail/test_cmail_setup.py b/tests/cmail/test_cmail_setup.py new file mode 100644 index 0000000..3330ee6 --- /dev/null +++ b/tests/cmail/test_cmail_setup.py @@ -0,0 +1,79 @@ +"""Test that cmail-setup-finish decrypts the mail password without a +world-readable window. + +gpg creates its --output file at the process umask (typically 0644), so a plain +`gpg --decrypt --output f` followed by `chmod 600 f` leaves the plaintext mail +password world-readable in the gap between the two. decrypt_to_secure wraps the +decrypt in a 0077 umask subshell so the file is 0600 from creation. + +Method: sed-extract decrypt_to_secure from scripts/cmail-setup-finish.sh, run it +with a fake gpg that (a) records the effective umask at call time and (b) writes +the --output file the way real gpg would (respecting the umask). Assert the umask +was tight during the write, and that the resulting file is 0600. + +Run from repo root: + python3 -m unittest tests.cmail.test_cmail_setup +""" + +import os +import stat +import subprocess +import tempfile +import textwrap +import unittest + +REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")) +SCRIPT = os.path.join(REPO_ROOT, "scripts", "cmail-setup-finish.sh") + + +def run_decrypt(): + """Run decrypt_to_secure with a fake gpg. Returns (plain_mode, umask_str).""" + tmp = tempfile.mkdtemp() + enc = os.path.join(tmp, "in.gpg") + plain = os.path.join(tmp, "cmailpass") + umask_rec = os.path.join(tmp, "umask") + with open(enc, "w") as fh: + fh.write("ciphertext") + script = textwrap.dedent(f"""\ + UMASK_REC={umask_rec!r} + gpg() {{ + umask > "$UMASK_REC" + local out="" + while [ $# -gt 0 ]; do + case "$1" in --output) out="$2"; shift;; esac + shift + done + printf 'SECRET' > "$out" + }} + source <(sed -n '/^decrypt_to_secure() {{/,/^}}/p' {SCRIPT!r}) + decrypt_to_secure {enc!r} {plain!r} + """) + subprocess.run(["bash", "-c", script], check=True, + capture_output=True, text=True, timeout=10) + mode = stat.S_IMODE(os.stat(plain).st_mode) + with open(umask_rec) as fh: + umask_str = fh.read().strip() + return mode, umask_str + + +class CmailDecryptPermissions(unittest.TestCase): + def test_decrypt_runs_under_tight_umask(self): + # The security property: gpg writes the plaintext under a 0077 umask, so + # it is 0600 from creation, not world-readable before the chmod. + _mode, umask_str = run_decrypt() + self.assertEqual( + int(umask_str, 8), 0o077, + "the mail-password decrypt must run under a 0077 umask so the " + "plaintext is never world-readable", + ) + + def test_plaintext_is_owner_only(self): + mode, _umask = run_decrypt() + self.assertEqual(mode, 0o600, + f"cmailpass ended at {oct(mode)}, expected 0600") + self.assertFalse(mode & 0o077, + "cmailpass must not be group- or world-accessible") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/gallery-tokens/test_gen_tokens.py b/tests/gallery-tokens/test_gen_tokens.py index b9fbb50..2968119 100644 --- a/tests/gallery-tokens/test_gen_tokens.py +++ b/tests/gallery-tokens/test_gen_tokens.py @@ -181,11 +181,15 @@ class ReplaceBetweenMarkers(unittest.TestCase): def test_start_marker_on_final_line_without_newline(self): # Defensive branch: no newline after the start marker. The guard must # not let text[:si-of-newline+1] collapse to "" and wipe the prefix. + # This input cannot occur in the real HTML (markers always sit on + # their own newline-terminated lines), so the exact-string assert is a + # characterization: the prefix survives, and the reconstruction's + # duplicated start marker is pinned deliberately so any change to the + # branch surfaces here instead of passing unnoticed. src = f"keep\n{self.START} {self.END}" out = gt.replace_between_markers(src, self.START, self.END, "X") - self.assertTrue(out.startswith("keep\n")) - self.assertIn(self.END, out) - self.assertIn("X", out) + self.assertEqual( + out, f"keep\n{self.START} X\n{self.START} {self.END}") class RealTokensJson(unittest.TestCase): diff --git a/tests/gallery-widgets/test-gallery-widget.el b/tests/gallery-widgets/test-gallery-widget.el index 166cd59..c0d38b4 100644 --- a/tests/gallery-widgets/test-gallery-widget.el +++ b/tests/gallery-widgets/test-gallery-widget.el @@ -73,10 +73,12 @@ (let ((xml (test-gallery-widget--svg-string 42))) ;; arc path stroked in the wash token (should (string-match-p (format "path[^>]*stroke=\"%s\"" (gallery-widget-token 'wash)) xml)) - ;; exactly three ticks - (should (= 3 (cl-count-if (lambda (_) t) - (split-string xml "class=\"tick\"" t) - :start 1))) + ;; exactly three ticks, counted as direct occurrences (the earlier + ;; split-string arithmetic depended on omit-nulls edge behavior) + (should (= 3 (let ((n 0) (pos 0)) + (while (string-match "class=\"tick\"" xml pos) + (setq n (1+ n) pos (match-end 0))) + n))) ;; needle + hub in the amber tokens (should (string-match-p (format "class=\"needle\"[^>]*stroke=\"%s\"" (gallery-widget-token 'gold-hi)) @@ -110,5 +112,53 @@ "The readout shows a rounded integer percent, matching the web card." (should (string-match-p ">67%<" (test-gallery-widget--svg-string 66.6)))) +(ert-deftest gallery-widget-gauge-out-of-range-readout-clamps () + "Readout and needle agree at out-of-range values: both clamp. +Regression: the needle clamped to the dial ends while the readout rendered +the raw value, so a gauge fed 150 pinned the needle at +60 degrees under a +\"150%\" label." + (let ((over (test-gallery-widget--svg-string 150))) + (should (string-match-p ">100%<" over)) + (should-not (string-match-p ">150%<" over))) + (let ((under (test-gallery-widget--svg-string -5))) + (should (string-match-p ">0%<" under)) + (should-not (string-match-p ">-5%<" under)))) + +(ert-deftest gallery-widget-source-dir-load-and-fallback () + "The token-file directory resolves from load context, else default-directory. +Regression: `file-name-directory' received nil when the load form was re-evaluated +outside a load and outside a file buffer." + (let ((load-file-name "/tmp/proto/gallery-widget.el")) + (should (equal (gallery-widget--source-dir) "/tmp/proto/"))) + (with-temp-buffer + (let ((load-file-name nil) + (default-directory "/tmp/elsewhere/")) + (should (equal (gallery-widget--source-dir) "/tmp/elsewhere/"))))) + +(ert-deftest gallery-widget-write-svg-writes-file-and-returns-path () + "The output helper writes the SVG document and returns the path." + (let ((file (make-temp-file "gallery-widget-test-" nil ".svg"))) + (unwind-protect + (let ((ret (gallery-widget-write-svg + (gallery-widget-needle-gauge 42) file))) + (should (equal ret file)) + (should (file-exists-p file)) + (with-temp-buffer + (insert-file-contents file) + (should (string-match-p "\\`<svg" (buffer-string))))) + (delete-file file)))) + +(ert-deftest gallery-widget-requires-cl-lib-at-load-time () + "Loading the module alone brings in cl-lib, not just an autoload cookie. +A cold byte-compile or a changed autoload would otherwise break +`gallery-widget--node' with a void-function error on first call." + (with-temp-buffer + (let ((emacs (expand-file-name invocation-name invocation-directory)) + (widget (expand-file-name "docs/prototypes/gallery-widget.el" + test-gallery-widget--root))) + (call-process emacs nil t nil "--batch" "-l" widget + "--eval" "(princ (if (featurep 'cl-lib) \"cl-lib:yes\" \"cl-lib:no\"))") + (should (string-match-p "cl-lib:yes" (buffer-string)))))) + (provide 'test-gallery-widget) ;;; test-gallery-widget.el ends here diff --git a/tests/import-wireguard-configs/fake-nmcli b/tests/import-wireguard-configs/fake-nmcli index 45b88cd..30de62f 100644 --- a/tests/import-wireguard-configs/fake-nmcli +++ b/tests/import-wireguard-configs/fake-nmcli @@ -38,6 +38,9 @@ case "$1 $2" in "connection modify") exit "${FAKE_NMCLI_MODIFY_RC:-0}" ;; +"connection down") + exit "${FAKE_NMCLI_DOWN_RC:-0}" + ;; *) echo "fake-nmcli: unexpected args: $*" >&2 exit 99 diff --git a/tests/import-wireguard-configs/test_import_wireguard_configs.py b/tests/import-wireguard-configs/test_import_wireguard_configs.py index 0307041..45afa54 100644 --- a/tests/import-wireguard-configs/test_import_wireguard_configs.py +++ b/tests/import-wireguard-configs/test_import_wireguard_configs.py @@ -162,6 +162,28 @@ class ImportWireguardConfigs(unittest.TestCase): imports = [ln for ln in self.log_lines() if ln.startswith("connection import")] self.assertEqual(len(imports), 1) + def test_tunnel_is_brought_down_before_the_modify(self): + # nmcli import auto-activates a full-tunnel (0.0.0.0/0) profile. The + # down must run before the rename/modify so a failed modify under set -e + # can never leave a live unasked-for VPN up. + self.write_conf("USNY") + r = self.run_script() + self.assertEqual(r.returncode, 0, r.stderr) + verbs = [ln.split()[1] for ln in self.log_lines() + if ln.startswith("connection ")] + self.assertEqual(verbs, ["import", "down", "modify"], verbs) + + def test_modify_failure_still_left_the_tunnel_down(self): + # Even when the modify fails and aborts the run, the down already ran, + # so no live tunnel survives. + self.write_conf("USNY") + r = self.run_script(env_extra={"FAKE_NMCLI_MODIFY_RC": "4"}) + self.assertNotEqual(r.returncode, 0) + verbs = [ln.split()[1] for ln in self.log_lines() + if ln.startswith("connection ")] + self.assertIn("down", verbs, "the tunnel must be downed before the modify aborts") + self.assertLess(verbs.index("down"), verbs.index("modify")) + if __name__ == "__main__": unittest.main() diff --git a/tests/installer-steps/test_check_disk_space.py b/tests/installer-steps/test_check_disk_space.py new file mode 100644 index 0000000..deec6d9 --- /dev/null +++ b/tests/installer-steps/test_check_disk_space.py @@ -0,0 +1,94 @@ +"""Test the disk-space pre-flight gate. + +Two bugs the extracted check_disk_space fixes: + +1. `df /` wraps to two lines when the source device name is long (common on a + live ISO / device-mapper root). The old `df / | awk 'NR==2 {print $4}'` + then reads the device-name line, gets an empty $4, and wrongly aborts on a + disk with plenty of space. `df -P /` forces POSIX single-line output. +2. The old code truncated KB -> GB with integer division before comparing, + biasing the gate against the user near the threshold. The check now compares + available KB against the minimum in KB directly. + +Method: sed-extract check_disk_space from the real `archsetup`, run it with a +fake df (controlled output, and wrapped output for the plain-df regression) and +assert the gate passes/aborts correctly. + +Run from repo root: + python3 -m unittest tests.installer-steps.test_check_disk_space +""" + +import os +import subprocess +import textwrap +import unittest + +REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")) +ARCHSETUP = os.path.join(REPO_ROOT, "archsetup") + +# Fake df: single, correct POSIX line for `df -P`; a WRAPPED two-line body for +# plain `df`. If check_disk_space ever reverts to plain `df`, the wrapped body +# yields an empty Available field and the ample-space test fails -- that is the +# regression guard for bug 1. +FAKE_DF = textwrap.dedent("""\ + df() {{ + if [ "$1" = "-P" ]; then + printf '%s\\n' "Filesystem 1024-blocks Used Available Capacity Mounted on" + printf '%s\\n' "/dev/mapper/root {avail} 100 {avail} 1% /" + else + printf '%s\\n' "Filesystem 1024-blocks Used Available Capacity Mounted on" + printf '%s\\n' "/dev/mapper/a-very-long-device-mapper-name-that-wraps" + printf '%s\\n' " {avail} 100 {avail} 1% /" + fi + }} +""") + + +def run_check(available_kb, min_gb=20): + """Run check_disk_space with fake df reporting available_kb free.""" + script = textwrap.dedent(f"""\ + min_disk_space_gb={min_gb} + {FAKE_DF.format(avail=available_kb)} + source <(sed -n '/^check_disk_space() {{/,/^}}/p' "{ARCHSETUP}") + check_disk_space + echo "REACHED-END" + """) + return subprocess.run( + ["bash", "-c", script], capture_output=True, text=True, timeout=10, + ) + + +class CheckDiskSpace(unittest.TestCase): + # min 20GB -> 20 * 1024 * 1024 = 20971520 KB + MIN_KB = 20 * 1024 * 1024 + + def test_ample_space_passes(self): + # ~95GB free; also the wrapped-df regression guard (plain df would + # yield an empty Available and wrongly abort). + r = run_check(99000000) + self.assertIn("REACHED-END", r.stdout) + self.assertIn("[OK] Disk space", r.stdout) + + def test_insufficient_space_aborts(self): + r = run_check(5 * 1024 * 1024) # 5GB + self.assertIn("Insufficient disk space", r.stdout) + self.assertNotIn("REACHED-END", r.stdout) + + def test_boundary_exact_minimum_passes(self): + r = run_check(self.MIN_KB) # exactly 20GB in KB + self.assertIn("REACHED-END", r.stdout) + + def test_boundary_one_kb_under_minimum_aborts(self): + r = run_check(self.MIN_KB - 1) + self.assertIn("Insufficient disk space", r.stdout) + self.assertNotIn("REACHED-END", r.stdout) + + def test_empty_df_output_aborts(self): + # df field unparseable -> treat as zero, abort rather than crash. + r = run_check("") + self.assertIn("Insufficient disk space", r.stdout) + self.assertNotIn("REACHED-END", r.stdout) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/installer-steps/test_ensure_nvme_early_module.py b/tests/installer-steps/test_ensure_nvme_early_module.py new file mode 100644 index 0000000..8e3f036 --- /dev/null +++ b/tests/installer-steps/test_ensure_nvme_early_module.py @@ -0,0 +1,95 @@ +"""Test the nvme early-module step. + +Two bugs the extracted ensure_nvme_early_module fixes: + +1. The MODULES=(... nvme) edit was only compiled into the initramfs by a + mkinitcpio -P that ran behind `if ! is_zfs_root`, so on ZFS-root machines + the early-load hardening never landed. The step now rebuilds whenever it + changed the file, unconditionally. +2. The already-present check grepped the whole file for "nvme", so a comment + or an unrelated token (nvme_tcp) anywhere in mkinitcpio.conf skipped the + edit. The check now looks for the nvme word on the MODULES line only. + +Method: sed-extract ensure_nvme_early_module from the real `archsetup`, point +it at a temp mkinitcpio.conf, and fake has_nvme_drives/backup_system_file/ +mkinitcpio/display/error_warn. + +Run from repo root: + python3 -m unittest tests.installer-steps.test_ensure_nvme_early_module +""" + +import os +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(conf_body, has_nvme=True): + with tempfile.TemporaryDirectory() as d: + conf = os.path.join(d, "mkinitcpio.conf") + with open(conf, "w") as f: + f.write(conf_body) + script = textwrap.dedent(f"""\ + logfile=/dev/null + action="" + display() {{ :; }} + backup_system_file() {{ :; }} + error_warn() {{ echo "WARN: $1"; return 1; }} + has_nvme_drives() {{ return {0 if has_nvme else 1}; }} + # stdout is redirected into $logfile at the call site, so the fake + # records its invocation in a side file instead. + mkinitcpio() {{ echo "MKINITCPIO $*" >> "{d}/mk.log"; }} + source <(sed -n '/^ensure_nvme_early_module() {{/,/^}}/p' "{ARCHSETUP}") + ensure_nvme_early_module "{conf}" + echo "RC=$?" + echo "CONF:[$(cat "{conf}")]" + [ -f "{d}/mk.log" ] && cat "{d}/mk.log" + """) + return subprocess.run( + ["bash", "-c", script], capture_output=True, text=True, timeout=10, + ) + + +class EnsureNvmeEarlyModule(unittest.TestCase): + def test_empty_modules_gets_nvme_and_rebuilds(self): + r = run("MODULES=()\nHOOKS=(base udev)\n") + self.assertIn("MODULES=(nvme)", r.stdout) + self.assertIn("MKINITCPIO -P", r.stdout, + "the initramfs must rebuild after the MODULES edit") + + def test_populated_modules_appends_nvme_and_rebuilds(self): + r = run("MODULES=(btrfs)\n") + self.assertIn("MODULES=(btrfs nvme)", r.stdout) + self.assertIn("MKINITCPIO -P", r.stdout) + + def test_nvme_already_present_no_edit_no_rebuild(self): + r = run("MODULES=(nvme)\n") + self.assertIn("MODULES=(nvme)", r.stdout) + self.assertNotIn("MODULES=(nvme nvme)", r.stdout) + self.assertNotIn("MKINITCPIO", r.stdout, + "an unchanged conf must not trigger a rebuild (resume idempotence)") + + def test_nvme_mention_elsewhere_does_not_skip_the_edit(self): + # The old whole-file grep skipped the edit when any line mentioned + # nvme; a comment must not mask a missing MODULES entry. + r = run("# early nvme notes: nvme_load=YES\nMODULES=(btrfs)\n") + self.assertIn("MODULES=(btrfs nvme)", r.stdout) + self.assertIn("MKINITCPIO -P", r.stdout) + + def test_nvme_tcp_module_does_not_count_as_nvme(self): + r = run("MODULES=(nvme_tcp)\n") + self.assertIn("MODULES=(nvme_tcp nvme)", r.stdout) + + def test_no_nvme_drives_is_a_noop(self): + r = run("MODULES=()\n", has_nvme=False) + self.assertIn("RC=0", r.stdout) + self.assertIn("CONF:[MODULES=()", r.stdout) + self.assertNotIn("MKINITCPIO", r.stdout) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/installer-steps/test_grub_cmdline.py b/tests/installer-steps/test_grub_cmdline.py new file mode 100644 index 0000000..8c0b5b0 --- /dev/null +++ b/tests/installer-steps/test_grub_cmdline.py @@ -0,0 +1,121 @@ +"""Test the GRUB cmdline merge. + +Bug (P2): configure_grub rewrote the whole GRUB_CMDLINE_LINUX_DEFAULT line +with a fixed string. A base install that had set cryptdevice=/resume=/zfs= +(or any other boot-critical token) lost it, and the following grub-mkconfig +baked an unbootable config. + +The fix is a merge: every pre-existing token survives, archsetup's tokens are +added, and where both set the same key archsetup's value wins. A safety +assert refuses to write if any existing token's key would vanish. + +Method: sed-extract merge_grub_cmdline (pure) and update_grub_cmdline +(file-level, run against temp grub files with a fake error_warn). + +Run from repo root: + python3 -m unittest tests.installer-steps.test_grub_cmdline +""" + +import os +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") + +EXTRACT = ( + "source <(sed -n '/^merge_grub_cmdline() {{/,/^}}/p;" + "/^update_grub_cmdline() {{/,/^}}/p' \"{a}\")" +).format(a=ARCHSETUP) + + +def run(body): + script = f"logfile=/dev/null\nerror_warn() {{ echo \"WARN: $1\"; return 1; }}\n{EXTRACT}\n{body}\n" + return subprocess.run(["bash", "-c", script], + capture_output=True, text=True, timeout=10) + + +def merge(existing, desired): + r = run(f'merge_grub_cmdline "{existing}" "{desired}"') + return r.stdout.strip() + + +class MergeGrubCmdline(unittest.TestCase): + DESIRED = "rw loglevel=2 quiet splash" + + def test_empty_existing_yields_desired(self): + self.assertEqual(merge("", self.DESIRED), self.DESIRED) + + def test_boot_critical_tokens_survive(self): + out = merge("cryptdevice=UUID=abc:root resume=/dev/nvme0n1p3 root=/dev/mapper/root", + self.DESIRED) + for tok in ("cryptdevice=UUID=abc:root", "resume=/dev/nvme0n1p3", + "root=/dev/mapper/root", "loglevel=2", "quiet", "splash"): + self.assertIn(tok, out.split(), f"{tok} missing from: {out}") + + def test_same_key_desired_value_wins_once(self): + out = merge("loglevel=7 quiet", self.DESIRED).split() + self.assertIn("loglevel=2", out) + self.assertNotIn("loglevel=7", out) + self.assertEqual(out.count("loglevel=2"), 1) + self.assertEqual(out.count("quiet"), 1) + + def test_zfs_token_survives(self): + out = merge("zfs=zroot/ROOT/default", self.DESIRED).split() + self.assertIn("zfs=zroot/ROOT/default", out) + + +class UpdateGrubCmdline(unittest.TestCase): + def run_update(self, grub_body): + with tempfile.TemporaryDirectory() as d: + path = os.path.join(d, "grub") + with open(path, "w") as f: + f.write(grub_body) + r = run(f'update_grub_cmdline "{path}"; echo "RC=$?"') + with open(path) as f: + return r, f.read() + + def line(self, content): + return [ln for ln in content.splitlines() + if ln.startswith('GRUB_CMDLINE_LINUX_DEFAULT=')] + + def test_existing_tokens_preserved_in_file(self): + _, content = self.run_update(textwrap.dedent("""\ + GRUB_TIMEOUT=5 + GRUB_CMDLINE_LINUX_DEFAULT="loglevel=3 cryptdevice=UUID=abc:root resume=/dev/sda2" + GRUB_DISABLE_RECOVERY=true + """)) + lines = self.line(content) + self.assertEqual(len(lines), 1) + val = lines[0] + self.assertIn("cryptdevice=UUID=abc:root", val) + self.assertIn("resume=/dev/sda2", val) + self.assertIn("loglevel=2", val) # archsetup's value wins + self.assertNotIn("loglevel=3", val) + self.assertIn("quiet", val) + # other lines untouched + self.assertIn("GRUB_TIMEOUT=5", content) + self.assertIn("GRUB_DISABLE_RECOVERY=true", content) + + def test_quoting_stays_a_single_pair(self): + _, content = self.run_update('GRUB_CMDLINE_LINUX_DEFAULT="quiet"\n') + val = self.line(content)[0] + self.assertEqual(val.count('"'), 2) + self.assertRegex(val, r'^GRUB_CMDLINE_LINUX_DEFAULT="[^"]*"$') + + def test_commented_line_gets_active_line_appended(self): + _, content = self.run_update('#GRUB_CMDLINE_LINUX_DEFAULT="quiet"\n') + self.assertEqual(len(self.line(content)), 1) + self.assertIn('#GRUB_CMDLINE_LINUX_DEFAULT="quiet"', content) + + def test_idempotent_on_rerun(self): + body = 'GRUB_CMDLINE_LINUX_DEFAULT="cryptdevice=UUID=abc:root"\n' + _, once = self.run_update(body) + _, twice = self.run_update(once) + self.assertEqual(once, twice) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/installer-steps/test_idempotency_cluster.py b/tests/installer-steps/test_idempotency_cluster.py new file mode 100644 index 0000000..ecb279d --- /dev/null +++ b/tests/installer-steps/test_idempotency_cluster.py @@ -0,0 +1,119 @@ +"""Test the resume-idempotency cluster: three extracted installer helpers. + +Each fix targets a step that misbehaves on a resume re-run or with unexpected +system state: + +- crontab_append_once: the log-cleanup cron line was appended unconditionally, + so a resume re-run of essential_services stacked a duplicate line each time. + The helper appends only when no existing line matches. +- zfs_scrub_timer_units: the scrub timer used `zpool list | head -1`, which + picks an arbitrary pool when several exist and yields the malformed unit + `zfs-scrub-weekly@.timer` when none do. The helper emits one unit per pool + and nothing for no pools. +- enable_user_service: gamemode was enabled via `systemctl --user enable`, + which the script itself documents fails at install time (no user bus). The + helper creates the wants symlink directly, the pattern syncthing already uses. + +Method: sed-extract each helper from the real `archsetup` and drive it with +plain stdin/args (the first two are pure) or a temp HOME (the third). + +Run from repo root: + python3 -m unittest tests.installer-steps.test_idempotency_cluster +""" + +import os +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") +ME = os.environ.get("USER") or __import__("getpass").getuser() + + +def extract(func): + return ( + "source <(sed -n '/^{f}() {{/,/^}}/p' \"{a}\")".format(f=func, a=ARCHSETUP) + ) + + +def run(func, body, stdin=None): + script = f"{extract(func)}\n{body}\n" + return subprocess.run( + ["bash", "-c", script], capture_output=True, text=True, timeout=10, + input=stdin, + ) + + +class CrontabAppendOnce(unittest.TestCase): + def call(self, existing, match, line): + body = f'crontab_append_once {match!r} {line!r}' + return run("crontab_append_once", body, stdin=existing).stdout + + def test_appends_when_absent(self): + out = self.call("", "log-cleanup", "0 12 * * * run-cleanup") + self.assertIn("0 12 * * * run-cleanup", out) + + def test_preserves_existing_and_appends(self): + out = self.call("0 0 * * * other-job\n", "log-cleanup", "0 12 * * * cleanup") + self.assertIn("other-job", out) + self.assertIn("0 12 * * * cleanup", out) + + def test_does_not_duplicate_when_present(self): + existing = "0 12 * * * $HOME/.local/bin/cron/log-cleanup\n" + out = self.call(existing, "cron/log-cleanup", "0 12 * * * dup") + self.assertNotIn("0 12 * * * dup", out) + self.assertEqual(out.count("log-cleanup"), 1) + + +class ZfsScrubTimerUnits(unittest.TestCase): + def units(self, pools_text): + out = run("zfs_scrub_timer_units", "zfs_scrub_timer_units", stdin=pools_text).stdout + return [ln for ln in out.splitlines() if ln] + + def test_single_pool(self): + self.assertEqual(self.units("zroot\n"), + ["zfs-scrub-weekly@zroot.timer"]) + + def test_no_pool_yields_no_unit(self): + self.assertEqual(self.units(""), []) + + def test_multiple_pools_each_get_a_unit(self): + self.assertEqual( + self.units("zroot\ntank\n"), + ["zfs-scrub-weekly@zroot.timer", "zfs-scrub-weekly@tank.timer"], + ) + + def test_blank_lines_are_skipped(self): + self.assertEqual(self.units("zroot\n\n"), + ["zfs-scrub-weekly@zroot.timer"]) + + +class EnableUserService(unittest.TestCase): + def test_creates_wants_symlink_in_user_config(self): + with tempfile.TemporaryDirectory() as home: + body = ( + f'enable_user_service {ME!r} gamemoded.service ' + f'/usr/lib/systemd/user/gamemoded.service {home!r}\n' + f'link="{home}/.config/systemd/user/default.target.wants/gamemoded.service"\n' + f'[ -L "$link" ] && echo "LINK=yes" || echo "LINK=no"\n' + f'echo "TARGET=$(readlink "$link")"' + ) + out = run("enable_user_service", body).stdout + self.assertIn("LINK=yes", out) + self.assertIn("TARGET=/usr/lib/systemd/user/gamemoded.service", out) + + def test_idempotent_on_rerun(self): + with tempfile.TemporaryDirectory() as home: + svc = "/usr/lib/systemd/user/gamemoded.service" + one = ( + f'enable_user_service {ME!r} gamemoded.service {svc!r} {home!r}\n' + f'enable_user_service {ME!r} gamemoded.service {svc!r} {home!r}\n' + f'echo "RC=$?"' + ) + out = run("enable_user_service", one).stdout + self.assertIn("RC=0", out) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/installer-steps/test_install_executable.py b/tests/installer-steps/test_install_executable.py new file mode 100644 index 0000000..6a2ecfd --- /dev/null +++ b/tests/installer-steps/test_install_executable.py @@ -0,0 +1,66 @@ +"""Test the guarded executable install helper. + +With `set -e` off, the installer's bare `cp script /usr/local/bin/...` followed +by `chmod +x` failed silently when the source was missing or partial, leaving a +systemd service with a dead ExecStart or a pacman hook pointing at a script that +was never installed. install_executable guards the copy and the chmod, warns on +failure, and skips the chmod when the copy did not land. + +Method: sed-extract install_executable from the real `archsetup` and run it +against real temp files (no fakes needed) with a fake error_warn recorder. + +Run from repo root: + python3 -m unittest tests.installer-steps.test_install_executable +""" + +import os +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(src, dest): + script = textwrap.dedent(f"""\ + logfile=/dev/null + error_warn() {{ echo "WARN: $1"; return 1; }} + source <(sed -n '/^install_executable() {{/,/^}}/p' "{ARCHSETUP}") + install_executable "{src}" "{dest}" + echo "RC=$?" + """) + return subprocess.run( + ["bash", "-c", script], capture_output=True, text=True, timeout=10, + ) + + +class InstallExecutable(unittest.TestCase): + def test_copies_and_makes_executable(self): + with tempfile.TemporaryDirectory() as d: + src = os.path.join(d, "myscript") + dest = os.path.join(d, "bin", "myscript") + os.mkdir(os.path.join(d, "bin")) + with open(src, "w") as f: + f.write("#!/bin/sh\necho hi\n") + r = run(src, dest) + self.assertIn("RC=0", r.stdout) + self.assertTrue(os.path.exists(dest)) + self.assertTrue(os.access(dest, os.X_OK), "dest must be executable") + self.assertNotIn("WARN", r.stdout) + + def test_missing_source_warns_and_skips_chmod(self): + with tempfile.TemporaryDirectory() as d: + src = os.path.join(d, "does-not-exist") + dest = os.path.join(d, "bin", "myscript") + os.mkdir(os.path.join(d, "bin")) + r = run(src, dest) + self.assertIn("WARN: copying", r.stdout, + "a failed copy must warn, not silently continue") + self.assertNotIn("RC=0", r.stdout, "helper returns non-zero on copy failure") + self.assertFalse(os.path.exists(dest)) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/installer-steps/test_pacman_hook_order.py b/tests/installer-steps/test_pacman_hook_order.py index 49acdbe..45058dd 100644 --- a/tests/installer-steps/test_pacman_hook_order.py +++ b/tests/installer-steps/test_pacman_hook_order.py @@ -1,26 +1,78 @@ -"""Regression tests for the pacman hook ordering safety invariant.""" +"""Regression tests for the pacman hook ordering safety invariant. + +pacman runs hooks in filename sort order, so the safety hooks archsetup +installs (the pre-pacman ZFS snapshot and the live-update guard, both +PreTransaction) must carry names that sort before mkinitcpio's stock +60-mkinitcpio-remove.hook. Otherwise a blocked transaction can remove the +current initramfs before the guard aborts, or after the snapshot window. + +The earlier version of this test asserted the ordering by comparing two string +literals to each other, which is constant-true and never looked at the source. +These tests extract the hook filenames the installer actually writes and +compare those, so a rename in the source flows into the assertion. +""" import os +import re import unittest REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")) ARCHSETUP = os.path.join(REPO_ROOT, "archsetup") +MKINITCPIO_REMOVE = "60-mkinitcpio-remove.hook" # stock mkinitcpio hook name + +WRITE_RE = re.compile(r">\s*/etc/pacman\.d/hooks/(\S+\.hook)\b") + + +def written_hooks(text): + """Hook filenames the source writes (redirections into the hooks dir).""" + names = [] + for line in text.splitlines(): + if "rm -f" in line: + continue + m = WRITE_RE.search(line) + if m: + names.append(m.group(1)) + return names + class PacmanHookOrderTests(unittest.TestCase): - def test_safety_hooks_precede_mkinitcpio_removal(self): + @classmethod + def setUpClass(cls): with open(ARCHSETUP, encoding="utf-8") as source: - text = source.read() + cls.text = source.read() + cls.hooks = written_hooks(cls.text) + def hook_named(self, suffix): + matches = [h for h in self.hooks if h.endswith(suffix)] + self.assertEqual( + len(matches), 1, + f"expected exactly one written hook ending {suffix}, got {matches}") + return matches[0] + + def test_safety_hooks_sort_before_mkinitcpio_removal(self): + # The invariant pacman actually enforces: filename sort order. + for suffix in ("zfs-snapshot.hook", "hypr-live-update-guard.hook"): + name = self.hook_named(suffix) + self.assertLess( + name, MKINITCPIO_REMOVE, + f"{name} must sort before {MKINITCPIO_REMOVE} or the guard " + "runs after the initramfs removal") + + def test_every_written_hook_has_an_ordering_prefix(self): + # Ordering is deliberate; a hook without a numeric prefix sorts by + # accident of its first letter. + self.assertTrue(self.hooks, "extraction found no written hooks") + for name in self.hooks: + self.assertRegex(name, r"^\d{2}-", + f"{name} lacks the two-digit ordering prefix") + + def test_stale_unprefixed_hooks_are_cleaned_up(self): + # Migration from installs made before the numeric prefixes. + self.assertIn("rm -f /etc/pacman.d/hooks/zfs-snapshot.hook", self.text) self.assertIn( - "cat << 'EOF' > /etc/pacman.d/hooks/05-zfs-snapshot.hook", text) - self.assertIn( - "cat > /etc/pacman.d/hooks/10-hypr-live-update-guard.hook", text) - self.assertLess("05-zfs-snapshot.hook", "60-mkinitcpio-remove.hook") - self.assertLess("10-hypr-live-update-guard.hook", "60-mkinitcpio-remove.hook") - self.assertIn("rm -f /etc/pacman.d/hooks/zfs-snapshot.hook", text) - self.assertIn("rm -f /etc/pacman.d/hooks/hypr-live-update-guard.hook", text) + "rm -f /etc/pacman.d/hooks/hypr-live-update-guard.hook", self.text) if __name__ == "__main__": diff --git a/tests/installer-steps/test_replace_sudoers_pacnew.py b/tests/installer-steps/test_replace_sudoers_pacnew.py new file mode 100644 index 0000000..8bfa914 --- /dev/null +++ b/tests/installer-steps/test_replace_sudoers_pacnew.py @@ -0,0 +1,76 @@ +"""Test the guarded sudoers.pacnew replacement. + +Bug (Minor, hard consequence): the installer did + [ -f /etc/sudoers.pacnew ] && cp /etc/sudoers.pacnew /etc/sudoers +with no validation. A malformed pacnew replaces /etc/sudoers with a file sudo +refuses to parse, locking out privilege escalation right before the NOPASSWD +rule is appended. replace_sudoers_pacnew now runs `visudo -cf` on the pacnew +first and only copies a file that validates; a failure warns (non-fatal) and +leaves the working sudoers in place. + +Method: sed-extract replace_sudoers_pacnew from the real `archsetup`, pass a +temp pacnew + temp target (the function's positional-arg testability seam), and +drive it with a fake visudo (controlled validation result) and a fake +error_warn recorder. + +Run from repo root: + python3 -m unittest tests.installer-steps.test_replace_sudoers_pacnew +""" + +import os +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(*, pacnew_exists, visudo_rc): + """Run replace_sudoers_pacnew with a temp pacnew/target and fake visudo.""" + with tempfile.TemporaryDirectory() as d: + pacnew = os.path.join(d, "sudoers.pacnew") + target = os.path.join(d, "sudoers") + with open(target, "w") as f: + f.write("ORIGINAL\n") + if pacnew_exists: + with open(pacnew, "w") as f: + f.write("NEW\n") + script = textwrap.dedent(f"""\ + logfile=/dev/null + display() {{ :; }} + visudo() {{ return {visudo_rc}; }} + error_warn() {{ echo "WARN: $1"; return 1; }} + source <(sed -n '/^replace_sudoers_pacnew() {{/,/^}}/p' "{ARCHSETUP}") + replace_sudoers_pacnew "{pacnew}" "{target}" + echo "RC=$?" + echo "TARGET=[$(cat "{target}")]" + """) + return subprocess.run( + ["bash", "-c", script], capture_output=True, text=True, timeout=10, + ) + + +class ReplaceSudoersPacnew(unittest.TestCase): + def test_valid_pacnew_is_copied(self): + r = run(pacnew_exists=True, visudo_rc=0) + self.assertIn("TARGET=[NEW]", r.stdout) + self.assertNotIn("WARN", r.stdout) + + def test_invalid_pacnew_warns_and_keeps_original(self): + r = run(pacnew_exists=True, visudo_rc=1) + self.assertIn("WARN", r.stdout, + "a pacnew that fails visudo must warn, not clobber sudoers") + self.assertIn("TARGET=[ORIGINAL]", r.stdout, + "the working sudoers must survive a malformed pacnew") + + def test_no_pacnew_is_a_noop(self): + r = run(pacnew_exists=False, visudo_rc=0) + self.assertIn("RC=0", r.stdout) + self.assertIn("TARGET=[ORIGINAL]", r.stdout) + self.assertNotIn("WARN", r.stdout) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/installer-steps/test_required_software.py b/tests/installer-steps/test_required_software.py new file mode 100644 index 0000000..85b01d8 --- /dev/null +++ b/tests/installer-steps/test_required_software.py @@ -0,0 +1,58 @@ +"""Test that install_required_software declares the expected base packages. + +inetutils provides /usr/bin/ftp, which TRAMP's /ftp: method (ange-ftp) shells +out to; Arch ships no command-line ftp client by default. It must be in the +base install so every machine gets it (Craig's dirvish config has an FTP +quick-access entry that depends on it). + +Method mirrors test_orchestrators: sed-extract install_required_software from +the real `archsetup`, stub pacman_install as a recorder that echoes the package +name, run, and assert membership. + +Run from repo root: + python3 -m unittest tests.installer-steps.test_required_software +""" + +import os +import subprocess +import textwrap +import unittest + +REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")) +ARCHSETUP = os.path.join(REPO_ROOT, "archsetup") + + +def required_packages(): + """Return (exit_code, [package, ...]) declared by install_required_software.""" + script = textwrap.dedent(f"""\ + display() {{ :; }} + pacman_install() {{ echo "$1"; }} + source <(sed -n '/^install_required_software() {{/,/^}}/p' "{ARCHSETUP}") + install_required_software + """) + result = subprocess.run( + ["bash", "-c", script], capture_output=True, text=True, timeout=10, + ) + return result.returncode, result.stdout.split() + + +class RequiredSoftware(unittest.TestCase): + def test_installs_inetutils_for_ftp(self): + rc, pkgs = required_packages() + self.assertEqual(rc, 0) + self.assertIn( + "inetutils", pkgs, + "inetutils (provides /usr/bin/ftp for TRAMP's /ftp: method) must be " + "in the base install", + ) + + def test_core_base_packages_present(self): + # Guard a few load-bearing entries so a bad edit to the list is caught. + rc, pkgs = required_packages() + self.assertEqual(rc, 0) + for p in ("base-devel", "git", "stow", "openssh", "curl"): + self.assertIn(p, pkgs, f"{p} dropped from the base install") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/installer-steps/test_run_step.py b/tests/installer-steps/test_run_step.py new file mode 100644 index 0000000..994e8c5 --- /dev/null +++ b/tests/installer-steps/test_run_step.py @@ -0,0 +1,92 @@ +"""Test run_step's state-marker and scope behavior. + +Bug (Major): run_step marked a step complete only when its function returned 0. +But error_fatal exits the script outright, so a step function that *returns* has +already survived every fatal check -- a non-zero return can only come from a +trailing non-fatal warning (error_warn returns 1). The old code read that as a +step failure and withheld the state marker, so the step re-ran on every resume. +run_step now records the marker whenever the function returns. + +Bug (Minor): run_step's step_name/step_func leaked to global scope. + +Method: sed-extract run_step + step_completed + mark_complete from the real +`archsetup`, point state_dir at a temp dir, and drive it with fake step +functions. + +Run from repo root: + python3 -m unittest tests.installer-steps.test_run_step +""" + +import os +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") + +EXTRACT = ( + "source <(sed -n '/^run_step() {{/,/^}}/p;" + "/^step_completed() {{/,/^}}/p;" + "/^mark_complete() {{/,/^}}/p' \"{archsetup}\")" +).format(archsetup=ARCHSETUP) + + +def run(step_body, pre_marked=False): + """Run run_step against a fake step function whose body is step_body.""" + with tempfile.TemporaryDirectory() as state_dir: + marker_setup = ( + f'mkdir -p "{state_dir}"; echo pre > "{state_dir}/mystep"\n' + if pre_marked else "" + ) + script = textwrap.dedent(f"""\ + state_dir="{state_dir}" + {marker_setup} + mystep() {{ + {step_body} + }} + {EXTRACT} + run_step "mystep" "mystep" + rc=$? + echo "RUN_STEP_RC=$rc" + # marker present? + [ -f "$state_dir/mystep" ] && echo "MARKED=yes" || echo "MARKED=no" + # locals must not leak + echo "LEAK_step_name=[${{step_name:-}}]" + echo "LEAK_step_func=[${{step_func:-}}]" + """) + return subprocess.run( + ["bash", "-c", script], capture_output=True, text=True, timeout=10, + ) + + +class RunStep(unittest.TestCase): + def test_success_marks_complete(self): + r = run("return 0") + self.assertIn("RUN_STEP_RC=0", r.stdout) + self.assertIn("MARKED=yes", r.stdout) + + def test_nonfatal_warning_return_still_marks_complete(self): + # The bug fix: a step ending in a non-fatal warning returns 1 but has + # done its work; it must be marked so resume does not re-run it. + r = run("return 1") + self.assertIn("MARKED=yes", r.stdout, + "a returned-non-zero step must still record its marker") + self.assertIn("RUN_STEP_RC=0", r.stdout, + "run_step returns 0 so the main loop treats it as handled") + + def test_already_completed_skips_and_does_not_rerun(self): + # Body would create a sentinel if it ran; pre-marked step must skip it. + r = run("echo RAN-BODY", pre_marked=True) + self.assertIn("Skipping", r.stdout) + self.assertNotIn("RAN-BODY", r.stdout) + + def test_locals_do_not_leak_to_global_scope(self): + r = run("return 0") + self.assertIn("LEAK_step_name=[]", r.stdout) + self.assertIn("LEAK_step_func=[]", r.stdout) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/installer-steps/test_set_user_password.py b/tests/installer-steps/test_set_user_password.py new file mode 100644 index 0000000..9f21b3a --- /dev/null +++ b/tests/installer-steps/test_set_user_password.py @@ -0,0 +1,56 @@ +"""Test that the primary user's password is set under a guard. + +archsetup runs with `set -e` OFF (line 21), so an unguarded `chpasswd` that +fails silently leaves the primary user with no password and no log entry -- an +unloggable account on a fresh install. set_user_password guards the chpasswd +with error_fatal so a failure aborts loudly instead of passing unnoticed. + +Method: sed-extract set_user_password from the real `archsetup`, run it with a +fake chpasswd (controlled exit) and a fake error_fatal recorder. Assert the +guard fires on failure and stays quiet on success. + +Run from repo root: + python3 -m unittest tests.installer-steps.test_set_user_password +""" + +import os +import subprocess +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_set_password(chpasswd_rc): + """Run set_user_password with a fake chpasswd exiting chpasswd_rc.""" + script = textwrap.dedent(f"""\ + password="hunter2" + logfile=/dev/null + display() {{ :; }} + chpasswd() {{ cat >/dev/null; return {chpasswd_rc}; }} + error_fatal() {{ echo "FATAL: $1"; exit 9; }} + source <(sed -n '/^set_user_password() {{/,/^}}/p' "{ARCHSETUP}") + set_user_password "alice" + echo "REACHED-END" + """) + return subprocess.run( + ["bash", "-c", script], capture_output=True, text=True, timeout=10, + ) + + +class SetUserPassword(unittest.TestCase): + def test_success_does_not_abort(self): + r = run_set_password(0) + self.assertIn("REACHED-END", r.stdout) + self.assertNotIn("FATAL", r.stdout) + + def test_failure_aborts_with_error_fatal(self): + r = run_set_password(1) + self.assertIn("FATAL: setting password for alice", r.stdout, + "a chpasswd failure must abort via error_fatal") + self.assertNotIn("REACHED-END", r.stdout) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/installer-steps/test_validate_yesno.py b/tests/installer-steps/test_validate_yesno.py new file mode 100644 index 0000000..da78fc5 --- /dev/null +++ b/tests/installer-steps/test_validate_yesno.py @@ -0,0 +1,57 @@ +"""Test the validate_yesno config-validation helper. + +The installer had four near-identical blocks validating that AUTOLOGIN, +NO_GPU_DRIVERS, INSTALL_CLAUDE_CODE, and INSTALL_DEVICE_UDEV_RULES are empty or +exactly yes/no. validate_yesno collapses them into one testable helper: empty +passes (the default), yes/no pass, anything else fails with a named error. + +Method: sed-extract validate_yesno from the real `archsetup` and drive it with +plain args. + +Run from repo root: + python3 -m unittest tests.installer-steps.test_validate_yesno +""" + +import os +import subprocess +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(name, value): + script = textwrap.dedent(f"""\ + source <(sed -n '/^validate_yesno() {{/,/^}}/p' "{ARCHSETUP}") + validate_yesno {name!r} {value!r} + echo "RC=$?" + """) + return subprocess.run( + ["bash", "-c", script], capture_output=True, text=True, timeout=10, + ) + + +class ValidateYesno(unittest.TestCase): + def test_yes_passes(self): + self.assertIn("RC=0", run("AUTOLOGIN", "yes").stdout) + + def test_no_passes(self): + self.assertIn("RC=0", run("AUTOLOGIN", "no").stdout) + + def test_empty_passes(self): + self.assertIn("RC=0", run("AUTOLOGIN", "").stdout) + + def test_other_value_fails_with_named_error(self): + r = run("NO_GPU_DRIVERS", "maybe") + self.assertNotIn("RC=0", r.stdout) + self.assertIn("NO_GPU_DRIVERS", r.stderr) + self.assertIn("maybe", r.stderr) + + def test_capitalized_yes_fails(self): + # The values are compared exactly; "Yes" is not accepted. + self.assertNotIn("RC=0", run("AUTOLOGIN", "Yes").stdout) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/net-scenarios/test_run_net_scenarios.py b/tests/net-scenarios/test_run_net_scenarios.py new file mode 100644 index 0000000..1d92185 --- /dev/null +++ b/tests/net-scenarios/test_run_net_scenarios.py @@ -0,0 +1,102 @@ +"""Test run-net-scenarios.sh scenario accounting. + +Bug: the scenario_diagnose_expect else-branch printed a FAIL line but never +forced a non-zero subshell exit, so a scenario whose only failure was the +diagnose check counted as passed and the run printed "all scenarios passed" +with exit 0. A net-doctor diagnosis regression would ship behind a green run. + +Method: run the real script against a temp scenario dir (NET_SCENARIO_DIR) +with stub ssh/rsync/jq on PATH, driving one fake scenario whose check results +are controlled per test. Assert on the script's exit code and summary line. + +Run from repo root: + python3 -m unittest tests.net-scenarios.test_run_net_scenarios +""" + +import os +import shutil +import subprocess +import tempfile +import textwrap +import unittest + +REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")) +SCRIPT = os.path.join(REPO_ROOT, "scripts", "testing", "run-net-scenarios.sh") + +STUB = "#!/bin/sh\ncat >/dev/null 2>&1 || true\nexit 0\n" + + +class RunNetScenarios(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.mkdtemp(prefix="net-scen-test-") + self.addCleanup(shutil.rmtree, self.tmp, ignore_errors=True) + self.bindir = os.path.join(self.tmp, "bin") + os.mkdir(self.bindir) + for tool in ("ssh", "rsync", "jq"): + p = os.path.join(self.bindir, tool) + with open(p, "w") as f: + f.write(STUB) + os.chmod(p, 0o755) + self.scen_dir = os.path.join(self.tmp, "scenarios") + os.mkdir(self.scen_dir) + + def write_scenario(self, *, brk=0, diagnose=0, fix=0, assert_rc=0, + with_diagnose=True): + diag_fn = ( + f"scenario_diagnose_expect() {{ return {diagnose}; }}\n" + if with_diagnose else "" + ) + body = textwrap.dedent(f"""\ + SCENARIO_DESC="fake scenario for harness tests" + scenario_break() {{ return {brk}; }} + {diag_fn}scenario_fix() {{ return {fix}; }} + scenario_assert() {{ return {assert_rc}; }} + """) + with open(os.path.join(self.scen_dir, "fake-scenario.sh"), "w") as f: + f.write(body) + + def run_script(self): + env = dict(os.environ) + env["PATH"] = self.bindir + os.pathsep + env["PATH"] + env["NET_SCENARIO_DIR"] = self.scen_dir + env["DOTFILES"] = self.tmp # rsync is stubbed; path just has to exist + return subprocess.run( + ["bash", SCRIPT, "--target", "root@fake-vm"], + capture_output=True, text=True, timeout=20, env=env, + ) + + def test_all_checks_pass_exits_zero(self): + self.write_scenario() + r = self.run_script() + self.assertEqual(r.returncode, 0, r.stdout + r.stderr) + self.assertIn("all scenarios passed", r.stdout) + + def test_diagnose_failure_alone_fails_the_run(self): + # The bug: break/fix/assert all pass, only the diagnose check fails. + self.write_scenario(diagnose=1) + r = self.run_script() + self.assertIn("diagnose did NOT name it", r.stdout) + self.assertNotEqual(r.returncode, 0, + "a diagnose regression must not exit green") + self.assertIn("1 scenario(s) failed", r.stdout) + + def test_assert_failure_fails_the_run(self): + self.write_scenario(assert_rc=1) + r = self.run_script() + self.assertNotEqual(r.returncode, 0) + self.assertIn("NOT repaired", r.stdout) + + def test_break_failure_fails_the_run(self): + self.write_scenario(brk=1) + r = self.run_script() + self.assertNotEqual(r.returncode, 0) + self.assertIn("break failed", r.stdout) + + def test_scenario_without_diagnose_hook_still_passes(self): + self.write_scenario(with_diagnose=False) + r = self.run_script() + self.assertEqual(r.returncode, 0, r.stdout + r.stderr) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/normalize-notify/fake-ffmpeg b/tests/normalize-notify/fake-ffmpeg new file mode 100644 index 0000000..cdfcb6f --- /dev/null +++ b/tests/normalize-notify/fake-ffmpeg @@ -0,0 +1,25 @@ +#!/bin/bash +# Fake ffmpeg for the normalize-notify-sounds tests. Two modes: +# measure (args include the `volumedetect` filter): print a mean_volume line +# to stderr, driven by FAKE_MEAN (default -20.0). +# encode (otherwise): write to the output file (the last argument), driven +# by FAKE_FFMPEG_EMPTY (1 -> zero-byte output) and FAKE_FFMPEG_FAIL +# (1 -> exit non-zero without writing). +set -uo pipefail + +for a in "$@"; do + if [ "$a" = "volumedetect" ]; then + echo "mean_volume: ${FAKE_MEAN:--20.0} dB" >&2 + exit 0 + fi +done + +out="${*: -1}" +if [ "${FAKE_FFMPEG_FAIL:-0}" = 1 ]; then + exit 1 +fi +if [ "${FAKE_FFMPEG_EMPTY:-0}" = 1 ]; then + : > "$out" +else + echo ENCODED > "$out" +fi diff --git a/tests/normalize-notify/test_normalize_notify_sounds.py b/tests/normalize-notify/test_normalize_notify_sounds.py new file mode 100644 index 0000000..ce3084b --- /dev/null +++ b/tests/normalize-notify/test_normalize_notify_sounds.py @@ -0,0 +1,117 @@ +"""Tests for normalize-notify-sounds.sh temp handling and atomic write. + +The script re-encodes each .ogg in place. SOUND_DIR is often the stow-symlinked +~/.local copy, so the write must land on the real repo file and preserve the +symlink. Two bugs the fix addresses: + +- A failed or empty re-encode used to truncate the target with `cat "$tmp" > + "$f"`, corrupting a repo-tracked sound to zero bytes. +- The mktemp had no EXIT trap, so an interrupted encode leaked a temp file. + +ffmpeg/ffprobe are faked via stubs on PATH (this directory) so the encode +result is controllable without real audio work. + +Run from repo root: + python3 -m unittest tests.normalize-notify.test_normalize_notify_sounds +""" + +import glob +import os +import shutil +import subprocess +import tempfile +import unittest + +HERE = os.path.dirname(os.path.abspath(__file__)) +REPO_ROOT = os.path.abspath(os.path.join(HERE, "..", "..")) +SCRIPT = os.path.join(REPO_ROOT, "scripts", "normalize-notify-sounds.sh") + + +class NormalizeNotifySounds(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.mkdtemp(prefix="norm-notify-test-") + self.addCleanup(shutil.rmtree, self.tmp, ignore_errors=True) + self.bindir = os.path.join(self.tmp, "bin") + os.mkdir(self.bindir) + for tool in ("ffmpeg", "ffprobe"): + fake = os.path.join(HERE, "fake-" + tool) + if not os.path.exists(fake): + fake = os.path.join(HERE, "fake-ffmpeg") # ffprobe reuses stub + dst = os.path.join(self.bindir, tool) + shutil.copy(fake, dst) + os.chmod(dst, 0o755) + # sound dir with one .ogg, plus a "repo" the sound symlinks into + self.repo = os.path.join(self.tmp, "repo") + os.mkdir(self.repo) + self.sounddir = os.path.join(self.tmp, "sounds") + os.mkdir(self.sounddir) + + def run_script(self, env_extra=None): + env = dict(os.environ) + env["PATH"] = self.bindir + os.pathsep + env["PATH"] + if env_extra: + env.update(env_extra) + return subprocess.run( + ["bash", SCRIPT, self.sounddir], + capture_output=True, text=True, timeout=20, env=env, + ) + + def make_sound(self, name="chime.ogg", body="ORIGINAL\n", symlink=False): + if symlink: + real = os.path.join(self.repo, name) + with open(real, "w") as f: + f.write(body) + link = os.path.join(self.sounddir, name) + os.symlink(real, link) + return link, real + path = os.path.join(self.sounddir, name) + with open(path, "w") as f: + f.write(body) + return path, path + + def leftover_temps(self, directory): + return glob.glob(os.path.join(directory, "*.ogg.*")) + \ + [p for p in glob.glob(os.path.join(directory, "tmp*")) if p.endswith(".ogg")] + + # --- Normal ---------------------------------------------------------- + + def test_reencodes_regular_file(self): + path, _ = self.make_sound() + r = self.run_script() + self.assertEqual(r.returncode, 0, r.stderr) + with open(path) as f: + self.assertEqual(f.read(), "ENCODED\n") + + def test_preserves_symlink_and_updates_target(self): + link, real = self.make_sound(symlink=True) + r = self.run_script() + self.assertEqual(r.returncode, 0, r.stderr) + self.assertTrue(os.path.islink(link), "the stow symlink must survive") + with open(real) as f: + self.assertEqual(f.read(), "ENCODED\n", + "the re-encode must land on the real repo file") + + # --- Error / corruption guard --------------------------------------- + + def test_empty_reencode_does_not_corrupt_target(self): + path, _ = self.make_sound(body="ORIGINAL\n") + self.run_script(env_extra={"FAKE_FFMPEG_EMPTY": "1"}) + with open(path) as f: + self.assertEqual(f.read(), "ORIGINAL\n", + "an empty re-encode must not truncate the tracked file") + + def test_failed_encode_leaves_no_temp_in_target_dir(self): + self.make_sound(symlink=True) + self.run_script(env_extra={"FAKE_FFMPEG_FAIL": "1"}) + self.assertEqual(self.leftover_temps(self.repo), [], + "a failed encode must not leak a temp file") + + def test_failed_encode_does_not_corrupt_target(self): + _, real = self.make_sound(symlink=True, body="ORIGINAL\n") + self.run_script(env_extra={"FAKE_FFMPEG_FAIL": "1"}) + with open(real) as f: + self.assertEqual(f.read(), "ORIGINAL\n") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/nvidia-preflight/test_nvidia_preflight.py b/tests/nvidia-preflight/test_nvidia_preflight.py index bdacfd5..191e983 100644 --- a/tests/nvidia-preflight/test_nvidia_preflight.py +++ b/tests/nvidia-preflight/test_nvidia_preflight.py @@ -50,7 +50,8 @@ class NvidiaPreflightHarness(unittest.TestCase): "#!/bin/bash\n" 'ARCHSETUP="$1"; shift\n' "source <(sed -n " - "'/^nvidia_preflight_report() {/,/^}/p' \"$ARCHSETUP\")\n" + "'/^NVIDIA_MIN_DRIVER=/p;" + "/^nvidia_preflight_report() {/,/^}/p' \"$ARCHSETUP\")\n" "nvidia_preflight_report\n" ) os.chmod(self.wrapper, 0o755) diff --git a/tests/vm-framework/test_vm_utils.py b/tests/vm-framework/test_vm_utils.py new file mode 100644 index 0000000..579955b --- /dev/null +++ b/tests/vm-framework/test_vm_utils.py @@ -0,0 +1,130 @@ +"""Tests for the VM test framework's vm-utils helpers. + +Two robustness bugs under test: + +1. init_vm_paths suffixed DISK_PATH and OVMF_VARS by FS_PROFILE but left + PID_FILE, MONITOR_SOCK, and SERIAL_LOG unsuffixed, so a concurrent btrfs + and zfs run shared them -- the second run's stop/is-running logic read the + first run's PID and could kill the other VM. All runtime paths now carry + the profile suffix. +2. kill_qemu sent kill -9 and deleted the PID file without waiting for the + process to die, so a force-kill fallback could run qemu-img snapshot + against a qcow2 the dying qemu still held locked. kill_qemu now waits for + the process to be dead (or reaped) before returning. + +Method: sed-extract init_vm_paths / kill_qemu / _cleanup_qemu_files from +lib/vm-utils.sh and drive them with temp dirs and real background processes. + +Run from repo root: + python3 -m unittest tests.vm-framework.test_vm_utils +""" + +import os +import subprocess +import tempfile +import textwrap +import unittest + +REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")) +VM_UTILS = os.path.join(REPO_ROOT, "scripts", "testing", "lib", "vm-utils.sh") + +EXTRACT = ( + "source <(sed -n '/^init_vm_paths() {{/,/^}}/p;" + "/^kill_qemu() {{/,/^}}/p;" + "/^_cleanup_qemu_files() {{/,/^}}/p' \"{lib}\")" +).format(lib=VM_UTILS) + + +def run(body): + script = f"fatal() {{ echo \"FATAL: $*\"; exit 1; }}\nwarn() {{ :; }}\n{EXTRACT}\n{body}\n" + return subprocess.run( + ["bash", "-c", script], capture_output=True, text=True, timeout=30, + ) + + +class InitVmPaths(unittest.TestCase): + def paths_for(self, profile): + with tempfile.TemporaryDirectory() as d: + body = textwrap.dedent(f"""\ + FS_PROFILE={profile} + init_vm_paths "{d}" + echo "DISK=$DISK_PATH" + echo "VARS=$OVMF_VARS" + echo "PID=$PID_FILE" + echo "MON=$MONITOR_SOCK" + echo "SER=$SERIAL_LOG" + """) + return run(body).stdout + + def test_zfs_profile_suffixes_all_runtime_paths(self): + out = self.paths_for("zfs") + self.assertIn("archsetup-base-zfs.qcow2", out) + self.assertIn("OVMF_VARS-zfs.fd", out) + # The bug: these three shared one name across profiles. + self.assertIn("PID=", out) + self.assertRegex(out, r"PID=.*-zfs", + "PID_FILE must carry the profile suffix") + self.assertRegex(out, r"MON=.*-zfs", + "MONITOR_SOCK must carry the profile suffix") + self.assertRegex(out, r"SER=.*-zfs", + "SERIAL_LOG must carry the profile suffix") + + def test_btrfs_profile_keeps_legacy_unsuffixed_names(self): + out = self.paths_for("btrfs") + self.assertIn("DISK=", out) + self.assertIn("archsetup-base.qcow2", out) + self.assertNotIn("-btrfs", out) + + def test_invalid_profile_fatals(self): + out = self.paths_for("ext4") + self.assertIn("FATAL", out) + + +class KillQemu(unittest.TestCase): + def test_process_is_dead_before_return(self): + # Contract pin for the snapshot-restore race: when kill_qemu returns, + # the process must be dead or reaped -- state Z or /proc entry gone -- + # so a following qemu-img call can't hit a still-held qcow2 lock. + with tempfile.TemporaryDirectory() as d: + body = textwrap.dedent(f"""\ + PID_FILE="{d}/qemu.pid" + MONITOR_SOCK="{d}/qemu-monitor.sock" + sleep 60 & victim=$! + echo "$victim" > "$PID_FILE" + kill_qemu + state=$(awk '{{print $3}}' "/proc/$victim/stat" 2>/dev/null || echo GONE) + echo "STATE=$state" + [ -f "$PID_FILE" ] && echo "PIDFILE=present" || echo "PIDFILE=removed" + """) + r = run(body) + self.assertRegex(r.stdout, r"STATE=(Z|GONE)", + f"process still live after kill_qemu: {r.stdout}") + self.assertIn("PIDFILE=removed", r.stdout) + + def test_stale_pid_file_is_cleaned_quietly(self): + with tempfile.TemporaryDirectory() as d: + body = textwrap.dedent(f"""\ + PID_FILE="{d}/qemu.pid" + MONITOR_SOCK="{d}/qemu-monitor.sock" + echo 99999999 > "$PID_FILE" + kill_qemu + echo "RC=$?" + [ -f "$PID_FILE" ] && echo "PIDFILE=present" || echo "PIDFILE=removed" + """) + r = run(body) + self.assertIn("RC=0", r.stdout) + self.assertIn("PIDFILE=removed", r.stdout) + + def test_no_pid_file_is_a_noop(self): + with tempfile.TemporaryDirectory() as d: + body = textwrap.dedent(f"""\ + PID_FILE="{d}/qemu.pid" + MONITOR_SOCK="{d}/qemu-monitor.sock" + kill_qemu + echo "RC=$?" + """) + self.assertIn("RC=0", run(body).stdout) + + +if __name__ == "__main__": + unittest.main() |