diff options
Diffstat (limited to 'tests')
| -rw-r--r-- | tests/gallery-tokens/test_gen_tokens.py | 10 | ||||
| -rw-r--r-- | tests/gallery-widgets/test-gallery-widget.el | 58 | ||||
| -rw-r--r-- | tests/installer-steps/test_ensure_nvme_early_module.py | 95 | ||||
| -rw-r--r-- | tests/installer-steps/test_grub_cmdline.py | 121 | ||||
| -rw-r--r-- | tests/installer-steps/test_pacman_hook_order.py | 72 | ||||
| -rw-r--r-- | tests/net-scenarios/test_run_net_scenarios.py | 102 | ||||
| -rw-r--r-- | tests/vm-framework/test_vm_utils.py | 130 |
7 files changed, 571 insertions, 17 deletions
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/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_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/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/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() |