diff options
Diffstat (limited to 'tests')
| -rw-r--r-- | tests/backup-system-file/test_backup_system_file.py | 161 | ||||
| -rw-r--r-- | tests/hypr-live-update-guard/test_hypr_live_update_guard.py | 165 | ||||
| -rw-r--r-- | tests/import-wireguard-configs/fake-nmcli | 45 | ||||
| -rw-r--r-- | tests/import-wireguard-configs/test_import_wireguard_configs.py | 167 | ||||
| -rw-r--r-- | tests/installer-steps/test_orchestrators.py | 149 | ||||
| -rw-r--r-- | tests/installer-steps/test_pacman_install.py | 95 | ||||
| -rw-r--r-- | tests/maint-scenarios/test_scenario_plan.py | 273 | ||||
| -rw-r--r-- | tests/network-diagnostics/test_network_diagnostics.py | 215 | ||||
| -rw-r--r-- | tests/nvidia-preflight/test_nvidia_preflight.py | 162 | ||||
| -rw-r--r-- | tests/run-task/test_run_task.py | 172 | ||||
| -rwxr-xr-x | tests/zfs-pre-snapshot/fake-zfs | 14 | ||||
| -rw-r--r-- | tests/zfs-pre-snapshot/test_zfs_pre_snapshot.py | 116 |
12 files changed, 1734 insertions, 0 deletions
diff --git a/tests/backup-system-file/test_backup_system_file.py b/tests/backup-system-file/test_backup_system_file.py new file mode 100644 index 0000000..5d48d03 --- /dev/null +++ b/tests/backup-system-file/test_backup_system_file.py @@ -0,0 +1,161 @@ +"""Tests for the backup_system_file helper in the archsetup installer. + +backup_system_file snapshots a pre-existing system file to +`<path>.archsetup.bak` before archsetup edits it in place, so a botched +in-place edit (fstab, mkinitcpio.conf, sudoers, ...) is recoverable. It is +idempotent: it never overwrites an existing backup, so the pristine original +survives repeated edits within a run and across re-runs of the installer. It +no-ops (success) when the target does not exist. + +These tests exercise the REAL function body, extracted from the `archsetup` +script at run time (not a copy), so the production code path is what runs. +Edits run against real temp files the test creates and tears down. + +Run from repo root: + python3 -m unittest tests.backup-system-file.test_backup_system_file +""" + +import os +import shutil +import stat +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 BackupHarness(unittest.TestCase): + """Source backup_system_file out of the real archsetup script and invoke it.""" + + def setUp(self): + self.tmp = tempfile.mkdtemp(prefix="backup-system-file-test-") + # A bash wrapper that extracts just the backup_system_file function from + # the real installer and invokes it with the test's arg. Sourcing the + # sed-extracted function means we test the production code path, not a + # reimplementation. The helper is self-contained (prints its own + # warnings), so no logger stub is needed. + 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 '/^backup_system_file() {/,/^}/p' \"$ARCHSETUP\")\n" + 'backup_system_file "$@"\n' + ) + os.chmod(self.wrapper, 0o755) + + def tearDown(self): + # Restore writability in case a test made a dir read-only. + for root, dirs, _ in os.walk(self.tmp): + for d in dirs: + os.chmod(os.path.join(root, d), 0o755) + shutil.rmtree(self.tmp, ignore_errors=True) + + def run_backup(self, target): + return subprocess.run( + ["bash", self.wrapper, ARCHSETUP, target], + capture_output=True, text=True, timeout=10, + ) + + def write(self, name, content, mode=None): + path = os.path.join(self.tmp, name) + with open(path, "w") as f: + f.write(content) + if mode is not None: + os.chmod(path, mode) + return path + + +# ----------------------------------------------------------------------------- +# Normal cases +# ----------------------------------------------------------------------------- + +class TestBackupNormal(BackupHarness): + + def test_existing_file_is_backed_up_with_same_content(self): + target = self.write("fstab", "UUID=abc / ext4 defaults 0 1\n") + result = self.run_backup(target) + self.assertEqual(result.returncode, 0, msg=result.stderr) + backup = target + ".archsetup.bak" + self.assertTrue(os.path.isfile(backup), "backup should be created") + with open(backup) as f: + self.assertEqual(f.read(), "UUID=abc / ext4 defaults 0 1\n") + + def test_backup_preserves_mode(self): + # sudoers ships 0440; a restored backup must keep restrictive perms. + target = self.write("sudoers", "root ALL=(ALL) ALL\n", mode=0o440) + result = self.run_backup(target) + self.assertEqual(result.returncode, 0, msg=result.stderr) + backup = target + ".archsetup.bak" + self.assertEqual(stat.S_IMODE(os.stat(backup).st_mode), 0o440) + + +# ----------------------------------------------------------------------------- +# Boundary cases +# ----------------------------------------------------------------------------- + +class TestBackupBoundary(BackupHarness): + + def test_existing_backup_is_not_overwritten(self): + # The pristine original must survive a later edit + second backup call. + target = self.write("pacman.conf", "PRISTINE\n") + self.assertEqual(self.run_backup(target).returncode, 0) + # Simulate archsetup editing the file in place, then backing up again. + with open(target, "w") as f: + f.write("EDITED\n") + result = self.run_backup(target) + self.assertEqual(result.returncode, 0, msg=result.stderr) + with open(target + ".archsetup.bak") as f: + self.assertEqual(f.read(), "PRISTINE\n", "backup must stay pristine") + + def test_missing_target_is_a_quiet_noop(self): + target = os.path.join(self.tmp, "never-existed.conf") + result = self.run_backup(target) + self.assertEqual(result.returncode, 0, msg=result.stderr) + self.assertFalse(os.path.exists(target + ".archsetup.bak")) + + def test_second_call_same_run_is_a_noop(self): + # A file edited twice in one run (e.g. mkinitcpio MODULES then HOOKS) + # gets backed up once; the second call must not error or re-copy. + target = self.write("mkinitcpio.conf", "HOOKS=(base udev)\n") + self.assertEqual(self.run_backup(target).returncode, 0) + backup = target + ".archsetup.bak" + first_mtime = os.stat(backup).st_mtime_ns + result = self.run_backup(target) + self.assertEqual(result.returncode, 0, msg=result.stderr) + self.assertEqual(os.stat(backup).st_mtime_ns, first_mtime, + "backup must not be rewritten on the second call") + + +# ----------------------------------------------------------------------------- +# Error cases +# ----------------------------------------------------------------------------- + +class TestBackupErrors(BackupHarness): + + def test_empty_target_is_refused(self): + result = self.run_backup("") + self.assertNotEqual(result.returncode, 0) + + def test_copy_failure_returns_nonzero(self): + # Target exists but its directory is read-only, so the .bak can't be + # written. The helper must report failure rather than silently skip. + subdir = os.path.join(self.tmp, "ro") + os.makedirs(subdir) + target = os.path.join(subdir, "fstab") + with open(target, "w") as f: + f.write("data\n") + os.chmod(subdir, 0o500) # r-x: owner cannot create the .bak here + try: + result = self.run_backup(target) + finally: + os.chmod(subdir, 0o755) + self.assertNotEqual(result.returncode, 0) + self.assertFalse(os.path.exists(target + ".archsetup.bak")) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/hypr-live-update-guard/test_hypr_live_update_guard.py b/tests/hypr-live-update-guard/test_hypr_live_update_guard.py new file mode 100644 index 0000000..a6c6f68 --- /dev/null +++ b/tests/hypr-live-update-guard/test_hypr_live_update_guard.py @@ -0,0 +1,165 @@ +"""Tests for the hypr-live-update-guard pacman PreTransaction hook script. + +The guard aborts a live pacman upgrade of GPU/compositor runtime libraries +(mesa, hyprland, wayland, GPU drivers) while a Hyprland session is running, +so the compositor doesn't SIGABRT when a now-"(deleted)" library is next +called. It reads the triggering package names on stdin (pacman NeedsTargets) +and exits non-zero to abort the transaction (AbortOnFail) before any package +is swapped. When Hyprland isn't running, or an override is set, it exits 0 +and the upgrade proceeds. + +Test seams (env vars the production script honors): + HYPR_GUARD_RUNNING 1/0 forces the Hyprland-running check (default: pgrep) + HYPR_ALLOW_LIVE_UPDATE 1 overrides the guard (proceed anyway) + HYPR_GUARD_SENTINEL path whose existence also overrides the guard + HYPR_GUARD_VERSIONS "pkg installed candidate" lines replacing the + pacman -Q / expac -S version lookups; when set, + a package absent from the map reads as unknown + (conservative block) + +Run from repo root: + python3 -m unittest tests.hypr-live-update-guard.test_hypr_live_update_guard +""" + +import os +import subprocess +import tempfile +import unittest + + +REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")) +GUARD = os.path.join(REPO_ROOT, "scripts", "hypr-live-update-guard") + +# Every package the legacy tests feed, mapped to a version CHANGE — those +# tests describe real upgrades, and the map keeps them hermetic (no +# pacman/expac calls against the test host). +CHANGING_VERSIONS = "\n".join(( + "mesa 25.1.0-1 26.0.0-1", + "hyprland 0.55.3-1 0.55.4-1", + "vulkan-radeon 25.1.0-1 26.0.0-1", +)) + + +def run_guard(stdin="mesa\n", running="1", allow=None, sentinel=None, + versions=CHANGING_VERSIONS): + env = dict(os.environ) + env["HYPR_GUARD_RUNNING"] = running + if allow is not None: + env["HYPR_ALLOW_LIVE_UPDATE"] = allow + # Point the sentinel at a path that does not exist unless a test sets one, + # so the host's real /run state can't leak into the result. + env["HYPR_GUARD_SENTINEL"] = sentinel if sentinel else "/nonexistent/guard-sentinel" + env["HYPR_GUARD_VERSIONS"] = versions + return subprocess.run( + ["sh", GUARD], + input=stdin, capture_output=True, text=True, timeout=10, env=env, + ) + + +class HyprLiveUpdateGuard(unittest.TestCase): + # --- Normal cases --------------------------------------------------- + + def test_running_with_dangerous_pkg_aborts(self): + r = run_guard(stdin="mesa\n", running="1") + self.assertEqual(r.returncode, 1, r.stderr) + + def test_abort_message_names_the_package_and_tty_remedy(self): + r = run_guard(stdin="mesa\n", running="1") + self.assertIn("mesa", r.stderr) + self.assertIn("TTY", r.stderr) + + def test_not_running_allows(self): + r = run_guard(stdin="mesa\n", running="0") + self.assertEqual(r.returncode, 0, r.stderr) + + def test_not_running_is_silent(self): + r = run_guard(stdin="mesa\nhyprland\n", running="0") + self.assertEqual(r.stderr.strip(), "") + + # --- Boundary cases ------------------------------------------------- + + def test_multiple_packages_all_listed(self): + r = run_guard(stdin="mesa\nhyprland\nvulkan-radeon\n", running="1") + self.assertEqual(r.returncode, 1) + for pkg in ("mesa", "hyprland", "vulkan-radeon"): + self.assertIn(pkg, r.stderr) + + def test_running_with_empty_stdin_still_guards(self): + # The hook only fires when dangerous targets exist, so an empty target + # list shouldn't normally happen; if Hyprland is up, stay safe (abort). + r = run_guard(stdin="", running="1") + self.assertEqual(r.returncode, 1) + + # --- Version awareness ------------------------------------------------ + + def test_same_version_reinstall_allows(self): + # a pure reinstall replaces identical bytes with identical bytes — + # no live-swap hazard, the guard must let it through + r = run_guard(stdin="hyprland\n", running="1", + versions="hyprland 0.55.4-1 0.55.4-1") + self.assertEqual(r.returncode, 0, r.stderr) + self.assertEqual(r.stderr.strip(), "") + + def test_all_same_version_multi_pkg_allows(self): + versions = "\n".join(("mesa 26.1.4-1 26.1.4-1", + "hyprland 0.55.4-1 0.55.4-1", + "vulkan-radeon 26.1.4-1 26.1.4-1")) + r = run_guard(stdin="mesa\nhyprland\nvulkan-radeon\n", running="1", + versions=versions) + self.assertEqual(r.returncode, 0, r.stderr) + + def test_mixed_blocks_naming_only_the_version_changing(self): + versions = "\n".join(("mesa 25.1.0-1 26.0.0-1", + "hyprland 0.55.4-1 0.55.4-1")) + r = run_guard(stdin="mesa\nhyprland\n", running="1", + versions=versions) + self.assertEqual(r.returncode, 1) + self.assertIn("- mesa", r.stderr) + # the prose mentions the Hyprland session, so assert on the + # package-list line format, not the bare word + self.assertNotIn("- hyprland", r.stderr) + + def test_unknown_versions_block_conservatively(self): + # seam set but package absent from the map = the lookup failed + # (AUR target, -U transaction, expac missing) — stay safe + r = run_guard(stdin="mesa\n", running="1", versions="") + self.assertEqual(r.returncode, 1) + self.assertIn("- mesa", r.stderr) + + # --- Override / error cases ----------------------------------------- + + def test_env_override_proceeds_even_when_running(self): + r = run_guard(stdin="mesa\n", running="1", allow="1") + self.assertEqual(r.returncode, 0, r.stderr) + + def test_sentinel_file_override_proceeds(self): + with tempfile.NamedTemporaryFile(prefix="guard-allow-") as f: + r = run_guard(stdin="mesa\n", running="1", sentinel=f.name) + self.assertEqual(r.returncode, 0, r.stderr) + + def test_sentinel_is_consumed_on_use(self): + # one touch = one transaction: if the override's cleanup never runs + # (a crashed caller), a leftover sentinel must not keep the guard + # disarmed until reboot — the hook deletes it as it honors it + fd, path = tempfile.mkstemp(prefix="guard-allow-") + os.close(fd) + try: + r = run_guard(stdin="mesa\n", running="1", sentinel=path) + self.assertEqual(r.returncode, 0, r.stderr) + self.assertFalse(os.path.exists(path)) + finally: + if os.path.exists(path): + os.unlink(path) + + def test_env_override_consumes_nothing(self): + # the env override isn't a file; nothing to consume, still proceeds + r = run_guard(stdin="mesa\n", running="1", allow="1") + self.assertEqual(r.returncode, 0, r.stderr) + + def test_override_env_zero_does_not_bypass(self): + r = run_guard(stdin="mesa\n", running="1", allow="0") + self.assertEqual(r.returncode, 1, r.stderr) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/import-wireguard-configs/fake-nmcli b/tests/import-wireguard-configs/fake-nmcli new file mode 100644 index 0000000..45b88cd --- /dev/null +++ b/tests/import-wireguard-configs/fake-nmcli @@ -0,0 +1,45 @@ +#!/bin/bash +# Fake nmcli for the import-wireguard-configs tests. +# +# Behavior is driven by env vars set by the test harness: +# FAKE_NMCLI_LOG file every invocation's args are appended to (one line +# per call; for imports the staged file's basename and +# content hash context are visible in the args) +# FAKE_NMCLI_NAMES newline-separated connection names returned by +# `nmcli -t -f NAME connection show` +# FAKE_NMCLI_IMPORT_OUT override for the import command's stdout +# (default: the real NM success line with a per-call +# deterministic UUID) +# FAKE_NMCLI_MODIFY_RC exit code for `nmcli connection modify` (default 0) +# +# Import calls also copy the staged file into $FAKE_NMCLI_LOG.d/ so tests can +# assert the temp copy was named wgpvpn.conf and carried the right content. +set -euo pipefail + +echo "$*" >>"$FAKE_NMCLI_LOG" + +case "$1 $2" in +"-t -f") + # nmcli -t -f NAME connection show + printf '%s\n' "${FAKE_NMCLI_NAMES:-}" + ;; +"connection import") + # nmcli connection import type wireguard file <path> + file="${6:?}" + mkdir -p "$FAKE_NMCLI_LOG.d" + n=$(find "$FAKE_NMCLI_LOG.d" -type f | wc -l) + cp "$file" "$FAKE_NMCLI_LOG.d/import-$n-$(basename "$file")" + if [ -n "${FAKE_NMCLI_IMPORT_OUT:-}" ]; then + echo "$FAKE_NMCLI_IMPORT_OUT" + else + printf "Connection 'wgpvpn' (%08d-aaaa-bbbb-cccc-dddddddddddd) successfully added.\n" "$n" + fi + ;; +"connection modify") + exit "${FAKE_NMCLI_MODIFY_RC:-0}" + ;; +*) + echo "fake-nmcli: unexpected args: $*" >&2 + exit 99 + ;; +esac diff --git a/tests/import-wireguard-configs/test_import_wireguard_configs.py b/tests/import-wireguard-configs/test_import_wireguard_configs.py new file mode 100644 index 0000000..0307041 --- /dev/null +++ b/tests/import-wireguard-configs/test_import_wireguard_configs.py @@ -0,0 +1,167 @@ +"""Tests for the import-wireguard-configs.sh one-time migration script. + +The script imports every assets/wireguard-config/*.conf into NetworkManager +as a wireguard connection with autoconnect forced off. NM quirks under test: +the import filename must be a valid interface name (<= 15 chars), so every +config stages through a temp copy named wgpvpn.conf and is renamed to the +real config name immediately after import — by the UUID parsed from the +import output, never by the transient wgpvpn name. A leftover connection +literally named wgpvpn (an earlier run died between import and rename, so +it still has autoconnect on) makes the script refuse to run. + +nmcli is faked via a stub on PATH (fake-nmcli in this directory) that logs +every invocation and snapshots the staged import file. + +Run from repo root: + python3 -m unittest tests.import-wireguard-configs.test_import_wireguard_configs +""" + +import os +import shutil +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", "import-wireguard-configs.sh") +FAKE_NMCLI = os.path.join(os.path.dirname(os.path.abspath(__file__)), "fake-nmcli") + + +class ImportWireguardConfigs(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.mkdtemp(prefix="import-wg-test-") + self.addCleanup(shutil.rmtree, self.tmp, ignore_errors=True) + self.confdir = os.path.join(self.tmp, "configs") + os.mkdir(self.confdir) + self.bindir = os.path.join(self.tmp, "bin") + os.mkdir(self.bindir) + shutil.copy(FAKE_NMCLI, os.path.join(self.bindir, "nmcli")) + os.chmod(os.path.join(self.bindir, "nmcli"), 0o755) + self.log = os.path.join(self.tmp, "nmcli.log") + + def write_conf(self, name, body="[Interface]\nPrivateKey = k\n"): + path = os.path.join(self.confdir, name + ".conf") + with open(path, "w") as f: + f.write(body) + return path + + def run_script(self, confdir=None, names="", env_extra=None): + env = dict(os.environ) + env["PATH"] = self.bindir + os.pathsep + env["PATH"] + env["FAKE_NMCLI_LOG"] = self.log + env["FAKE_NMCLI_NAMES"] = names + if env_extra: + env.update(env_extra) + return subprocess.run( + ["bash", SCRIPT, confdir or self.confdir], + capture_output=True, text=True, timeout=10, env=env, + ) + + def log_lines(self): + if not os.path.exists(self.log): + return [] + with open(self.log) as f: + return [ln.strip() for ln in f if ln.strip()] + + # --- Normal cases ---------------------------------------------------- + + def test_imports_every_conf_with_autoconnect_off(self): + self.write_conf("USNY") + self.write_conf("USDC") + r = self.run_script() + self.assertEqual(r.returncode, 0, r.stderr) + modifies = [ln for ln in self.log_lines() if ln.startswith("connection modify")] + self.assertEqual(len(modifies), 2) + for ln in modifies: + self.assertIn("connection.autoconnect no", ln) + self.assertIn("imported: USDC", r.stdout) + self.assertIn("imported: USNY", r.stdout) + + def test_renames_by_uuid_from_import_output_not_by_name(self): + self.write_conf("USNY") + r = self.run_script() + self.assertEqual(r.returncode, 0, r.stderr) + modify = [ln for ln in self.log_lines() if ln.startswith("connection modify")][0] + # The modify targets the UUID the import printed, and never the + # transient wgpvpn name. + self.assertIn("00000000-aaaa-bbbb-cccc-dddddddddddd", modify) + self.assertIn("connection.id USNY", modify) + self.assertNotIn("modify wgpvpn", modify) + + def test_long_name_stages_through_wgpvpn_temp_copy(self): + # switzerlan-zurich1 is 18 chars — over NM's 15-char interface-name + # limit, the reason the staging copy exists at all. + body = "[Interface]\nPrivateKey = long-name-key\n" + self.write_conf("switzerlan-zurich1", body) + r = self.run_script() + self.assertEqual(r.returncode, 0, r.stderr) + staged = os.listdir(self.log + ".d") + self.assertEqual(len(staged), 1) + self.assertTrue(staged[0].endswith("wgpvpn.conf"), staged) + with open(os.path.join(self.log + ".d", staged[0])) as f: + self.assertEqual(f.read(), body) + self.assertIn("imported: switzerlan-zurich1", r.stdout) + + # --- Idempotence ----------------------------------------------------- + + def test_already_imported_names_skip(self): + self.write_conf("USNY") + self.write_conf("USDC") + r = self.run_script(names="USNY\nsome-wifi") + self.assertEqual(r.returncode, 0, r.stderr) + self.assertIn("skip: USNY", r.stdout) + self.assertIn("imported: USDC", r.stdout) + modifies = [ln for ln in self.log_lines() if ln.startswith("connection modify")] + self.assertEqual(len(modifies), 1) + + def test_all_imported_is_a_clean_noop(self): + self.write_conf("USNY") + r = self.run_script(names="USNY") + self.assertEqual(r.returncode, 0, r.stderr) + imports = [ln for ln in self.log_lines() if ln.startswith("connection import")] + self.assertEqual(imports, []) + + # --- Boundary cases -------------------------------------------------- + + def test_empty_config_dir_fails_loudly(self): + r = self.run_script() + self.assertEqual(r.returncode, 1) + self.assertIn("no .conf files", r.stderr) + + def test_missing_config_dir_fails_loudly(self): + r = self.run_script(confdir=os.path.join(self.tmp, "nope")) + self.assertEqual(r.returncode, 1) + self.assertIn("no such config dir", r.stderr) + + # --- Error cases ----------------------------------------------------- + + def test_stale_wgpvpn_connection_refuses_to_run(self): + self.write_conf("USNY") + r = self.run_script(names="wgpvpn\nUSDC") + self.assertEqual(r.returncode, 1) + self.assertIn("stale", r.stderr) + self.assertIn("nmcli connection delete wgpvpn", r.stderr) + imports = [ln for ln in self.log_lines() if ln.startswith("connection import")] + self.assertEqual(imports, []) + + def test_unparseable_import_output_aborts(self): + self.write_conf("USNY") + r = self.run_script(env_extra={"FAKE_NMCLI_IMPORT_OUT": "something unexpected"}) + self.assertEqual(r.returncode, 1) + self.assertIn("could not parse a UUID", r.stderr) + modifies = [ln for ln in self.log_lines() if ln.startswith("connection modify")] + self.assertEqual(modifies, []) + + def test_modify_failure_aborts_the_run(self): + self.write_conf("USNY") + self.write_conf("USDC") + r = self.run_script(env_extra={"FAKE_NMCLI_MODIFY_RC": "4"}) + self.assertNotEqual(r.returncode, 0) + # set -e stops at the first failed modify — only one import attempted. + imports = [ln for ln in self.log_lines() if ln.startswith("connection import")] + self.assertEqual(len(imports), 1) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/installer-steps/test_orchestrators.py b/tests/installer-steps/test_orchestrators.py new file mode 100644 index 0000000..2a771ba --- /dev/null +++ b/tests/installer-steps/test_orchestrators.py @@ -0,0 +1,149 @@ +"""Characterization tests for the decomposed installer step orchestrators. + +The 2026 decomposition turned the giant step functions into thin +orchestrators that call one named sub-function per concern. These tests pin +the call SEQUENCE of each orchestrator: a dropped, added, or reordered +sub-step call fails the test. They guard the wiring, not the sub-functions' +own behavior (those mutate the system and are exercised by the VM harness). + +Method: sed-extract the orchestrator from the real `archsetup` (its body is +now just `display` + sub-function calls), source it with `display` silenced +and every sub-function replaced by a recorder that echoes its own name, run +it, and assert stdout is the expected ordered list. + +Run from repo root: + python3 -m unittest tests.installer-steps.test_orchestrators +""" + +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") + +# orchestrator -> exact ordered sub-step calls +ORCHESTRATORS = { + "essential_services": [ + "configure_randomness", "configure_networking", "configure_power", + "configure_ssh_server", "configure_fail2ban", "configure_firewall", + "configure_service_discovery", "configure_job_scheduling", + "configure_package_cache", "configure_snapshots", + "configure_user_lingering", + ], + "prerequisites": [ + "bootstrap_pacman_keyring", "install_required_software", + "configure_build_environment", "configure_package_mirrors", + ], + "developer_workstation": [ + "install_programming_languages", "install_editors", + "install_android_utilities", "install_vpn_tools", + "install_devops_utilities", + ], + "boot_ux": [ + "tighten_efi_permissions", "add_nvme_early_module", + "configure_initramfs_hook", "configure_encrypted_autologin", + "configure_tlp_power", "trim_firmware", "configure_grub", + "configure_pre_pacman_snapshots", + ], + "user_customizations": [ + "clone_user_repos", "stow_dotfiles", "prune_waybar_battery", + "refresh_desktop_caches", "configure_dconf_defaults", + "finalize_dotfiles", "install_maintenance_config", + "create_user_directories", + ], +} + + +def run_orchestrator(func, stubs, extra_defs=""): + """Source `func` from archsetup with `stubs` recording their names.""" + stub_defs = "\n".join(f"{s}() {{ echo {s}; }}" for s in stubs) + script = textwrap.dedent(f"""\ + display() {{ :; }} + {stub_defs} + {extra_defs} + source <(sed -n '/^{func}() {{/,/^}}/p' "{ARCHSETUP}") + {func} + """) + result = subprocess.run( + ["bash", "-c", script], + capture_output=True, text=True, timeout=10, + ) + return result + + +class OrchestratorSequence(unittest.TestCase): + def test_each_orchestrator_calls_substeps_in_order(self): + for func, expected in ORCHESTRATORS.items(): + with self.subTest(orchestrator=func): + result = run_orchestrator(func, expected) + self.assertEqual(result.returncode, 0, result.stderr) + got = result.stdout.split() + self.assertEqual(got, expected, + f"{func} call sequence drifted") + + +class SnapshotDispatch(unittest.TestCase): + """configure_snapshots branches on filesystem; pin each branch.""" + + SUBS = ["configure_zfs_snapshots", "configure_btrfs_snapshots"] + + def test_zfs_root_runs_zfs_snapshots(self): + result = run_orchestrator( + "configure_snapshots", self.SUBS, + extra_defs="is_zfs_root() { return 0; }\nis_btrfs_root() { return 1; }", + ) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertEqual(result.stdout.split(), ["configure_zfs_snapshots"]) + + def test_btrfs_root_runs_btrfs_snapshots(self): + result = run_orchestrator( + "configure_snapshots", self.SUBS, + extra_defs="is_zfs_root() { return 1; }\nis_btrfs_root() { return 0; }", + ) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertEqual(result.stdout.split(), ["configure_btrfs_snapshots"]) + + def test_other_filesystem_runs_neither(self): + result = run_orchestrator( + "configure_snapshots", self.SUBS, + extra_defs="is_zfs_root() { return 1; }\nis_btrfs_root() { return 1; }", + ) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertEqual(result.stdout.split(), []) + + +class MaintenanceConfigDispatch(unittest.TestCase): + """install_maintenance_config branches on desktop_env; pin each branch. + + The thresholds TOML installs for every environment (the CLI works + headless); the scan timers are user units in the hyprland stow tier, so + their enablement is hyprland-only. + """ + + SUBS = ["install_maintenance_thresholds", "enable_maint_timers"] + + def test_hyprland_installs_thresholds_and_enables_timers(self): + result = run_orchestrator( + "install_maintenance_config", self.SUBS, + extra_defs='desktop_env=hyprland', + ) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertEqual(result.stdout.split(), self.SUBS) + + def test_non_hyprland_installs_thresholds_only(self): + for env in ("dwm", "none"): + with self.subTest(desktop_env=env): + result = run_orchestrator( + "install_maintenance_config", self.SUBS, + extra_defs=f'desktop_env={env}', + ) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertEqual(result.stdout.split(), + ["install_maintenance_thresholds"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/installer-steps/test_pacman_install.py b/tests/installer-steps/test_pacman_install.py new file mode 100644 index 0000000..28c9a7f --- /dev/null +++ b/tests/installer-steps/test_pacman_install.py @@ -0,0 +1,95 @@ +"""Characterization tests for pacman_install's install-reason handling. + +pacman --needed skips a package that is already present as a dependency and +leaves its install reason alone. A declared package can then sit as asdeps +on an existing system, show up as an orphan once its accidental dependent +leaves, and get swept away by an orphan cleanup (expac and lm_sensors nearly +went this way on 2026-07-08). pacman_install therefore marks every declared +package explicit after a successful install. + +Method mirrors test_orchestrators: sed-extract the real functions from +`archsetup`, source them with `display`/`error_warn` silenced and `pacman` +replaced by a recorder, run, and assert the recorded calls. + +Run from repo root: + python3 -m unittest tests.installer-steps.test_pacman_install +""" + +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_pacman_install(pkg, pacman_s_rc=0, pacman_d_rc=0): + """Extract retry_install + pacman_install, run against a fake pacman. + + Returns (exit_code, recorded pacman calls as a list of strings). + """ + script = textwrap.dedent(""" + set -u + MAX_INSTALL_RETRIES=3 + logfile=/dev/null + display() { :; } + error_warn() { return 1; } + pacman() { + echo "pacman $*" >> "$CALLS" + case "$1" in + --noconfirm) return "$PACMAN_S_RC" ;; + -D) return "$PACMAN_D_RC" ;; + esac + } + %(functions)s + pacman_install "%(pkg)s" + """) + extract = subprocess.run( + ["sed", "-n", + "/^retry_install()/,/^}/p;/^pacman_install()/,/^}/p", ARCHSETUP], + capture_output=True, text=True, check=True) + calls_file = os.path.join(os.environ.get("TMPDIR", "/tmp"), + f"pacman-install-calls-{os.getpid()}") + if os.path.exists(calls_file): + os.unlink(calls_file) + open(calls_file, "w").close() + env = dict(os.environ, CALLS=calls_file, + PACMAN_S_RC=str(pacman_s_rc), PACMAN_D_RC=str(pacman_d_rc)) + proc = subprocess.run( + ["bash", "-c", script % {"functions": extract.stdout, "pkg": pkg}], + capture_output=True, text=True, env=env) + with open(calls_file) as f: + calls = [line.strip() for line in f if line.strip()] + os.unlink(calls_file) + return proc.returncode, calls + + +class PacmanInstallTests(unittest.TestCase): + def test_success_marks_package_explicit(self): + rc, calls = run_pacman_install("expac") + self.assertEqual(rc, 0) + self.assertIn("pacman --noconfirm --needed -S expac", calls) + self.assertIn("pacman -D --asexplicit expac", calls) + # the mark comes after the install, never before + self.assertGreater(calls.index("pacman -D --asexplicit expac"), + calls.index("pacman --noconfirm --needed -S expac")) + + def test_failed_install_never_marks(self): + rc, calls = run_pacman_install("expac", pacman_s_rc=1) + self.assertNotEqual(rc, 0) + self.assertNotIn("pacman -D --asexplicit expac", calls) + # all three retry attempts happened + self.assertEqual( + calls.count("pacman --noconfirm --needed -S expac"), 3) + + def test_mark_failure_does_not_fail_the_install(self): + # -D can fail in odd corners (readonly db mid-transaction); the + # install itself succeeded and must report success + rc, calls = run_pacman_install("expac", pacman_d_rc=1) + self.assertEqual(rc, 0) + self.assertIn("pacman -D --asexplicit expac", calls) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/maint-scenarios/test_scenario_plan.py b/tests/maint-scenarios/test_scenario_plan.py new file mode 100644 index 0000000..9a72db2 --- /dev/null +++ b/tests/maint-scenarios/test_scenario_plan.py @@ -0,0 +1,273 @@ +"""Tests for the maint VM scenario runner's plan layer (no VM needed). + +run-maint-scenarios.sh orchestrates break -> `maint fix` -> assert scenario +scripts over the existing qemu-img snapshot primitives (lib/vm-utils.sh). +Scenarios are grouped into non-conflicting batches that share one VM boot; +a stop -> restore -> boot cycle runs only between groups (the spec's +grouped-batch isolation policy). The runner therefore has a pure planning +layer -- enumerate scenario files, validate their contract, filter by +filesystem profile and --group, and print the batch plan -- that runs +without KVM, a base image, or root. + +These tests exercise that layer through the REAL script via `--list`: + - against the shipped scenarios directory (contract holds for every file + we actually ship); + - against fake scenario directories (MAINT_SCENARIO_DIR override) for the + validation failures a shipped tree must never contain. + +The scenario-file contract validated here: + - vars SCENARIO_DESC (non-empty), SCENARIO_GROUP (token), + SCENARIO_PROFILES (btrfs/zfs/any, space-separated); + - functions scenario_break, scenario_fix, scenario_assert; + - defining only -- sourcing a scenario file must not execute commands + (the probe sources files in a bare shell with no helpers defined). + +Run from repo root: + python3 -m unittest tests.maint-scenarios.test_scenario_plan +""" + +import os +import subprocess +import tempfile +import unittest + +REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")) +RUNNER = os.path.join(REPO_ROOT, "scripts", "testing", "run-maint-scenarios.sh") +SCENARIO_DIR = os.path.join(REPO_ROOT, "scripts", "testing", "maint-scenarios") + +GOOD_SCENARIO = """\ +SCENARIO_DESC="{desc}" +SCENARIO_GROUP="{group}" +SCENARIO_PROFILES="{profiles}" +scenario_break() {{ mexec "true"; }} +scenario_fix() {{ mfix some_remedy; }} +scenario_assert() {{ mexec "true"; }} +""" + + +def run_list(extra_args=(), scenario_dir=None, fs_profile=None): + # Hermetic against the caller's FS_PROFILE: the Makefile exports it, so + # `make test-unit FS_PROFILE=zfs` would otherwise change what --list + # shows. Tests that care pass fs_profile explicitly; everything else + # runs the runner's own default (btrfs). + env = {k: v for k, v in os.environ.items() if k != "FS_PROFILE"} + if scenario_dir is not None: + env["MAINT_SCENARIO_DIR"] = scenario_dir + if fs_profile is not None: + env["FS_PROFILE"] = fs_profile + return subprocess.run( + ["bash", RUNNER, "--list", *extra_args], + capture_output=True, text=True, env=env, cwd=REPO_ROOT, + ) + + +def write_scenario(dirpath, name, desc="a scenario", group="g1", + profiles="any", body=None): + path = os.path.join(dirpath, name) + with open(path, "w") as f: + f.write(body if body is not None + else GOOD_SCENARIO.format(desc=desc, group=group, + profiles=profiles)) + return path + + +class ShippedScenariosTests(unittest.TestCase): + """The scenarios we actually ship satisfy the contract.""" + + def test_shipped_dir_exists_and_is_nonempty(self): + files = [f for f in os.listdir(SCENARIO_DIR) if f.endswith(".sh")] + self.assertTrue(files, "no scenario files shipped") + + def test_list_exits_zero_on_shipped_scenarios(self): + proc = run_list() + self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr) + + def test_list_names_every_shipped_scenario(self): + proc = run_list() + for f in os.listdir(SCENARIO_DIR): + if not f.endswith(".sh"): + continue + name = f.split("-", 1)[1][:-3] if "-" in f else f[:-3] + self.assertIn(name, proc.stdout, + f"scenario {f} missing from --list output") + + def test_list_groups_are_headed(self): + proc = run_list() + self.assertRegex(proc.stdout, r"(?m)^group \S+:") + + def test_shipped_scenarios_define_contract_without_executing(self): + """Sourcing a scenario file in a bare bash defines the contract vars + and functions and runs nothing (no helpers exist at source time, so + any top-level command would fail loudly).""" + probe = ( + 'set -eu; source "$1"; ' + ': "${SCENARIO_DESC:?}" "${SCENARIO_GROUP:?}" ' + '"${SCENARIO_PROFILES:?}"; ' + 'case " $SCENARIO_PROFILES " in *" btrfs "*|*" zfs "*|*" any "*) ' + ';; *) echo "bad profiles: $SCENARIO_PROFILES" >&2; exit 1;; esac; ' + 'declare -f scenario_break scenario_fix scenario_assert >/dev/null' + ) + for f in sorted(os.listdir(SCENARIO_DIR)): + if not f.endswith(".sh"): + continue + path = os.path.join(SCENARIO_DIR, f) + proc = subprocess.run(["bash", "-c", probe, "probe", path], + capture_output=True, text=True) + self.assertEqual(proc.returncode, 0, + f"{f}: contract violation\n{proc.stderr}") + + +class PlanFilteringTests(unittest.TestCase): + """Profile and --group filtering over a fake scenario dir.""" + + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.dir = self.tmp.name + write_scenario(self.dir, "10-first-btrfs.sh", + group="alpha", profiles="btrfs") + write_scenario(self.dir, "20-second-any.sh", + group="beta", profiles="any") + write_scenario(self.dir, "30-third-zfs.sh", + group="gamma", profiles="zfs") + + def tearDown(self): + self.tmp.cleanup() + + def test_btrfs_profile_excludes_zfs_scenarios(self): + proc = run_list(scenario_dir=self.dir, fs_profile="btrfs") + self.assertEqual(proc.returncode, 0, proc.stderr) + self.assertIn("first-btrfs", proc.stdout) + self.assertIn("second-any", proc.stdout) + self.assertNotIn("third-zfs", proc.stdout) + + def test_zfs_profile_excludes_btrfs_scenarios(self): + proc = run_list(scenario_dir=self.dir, fs_profile="zfs") + self.assertEqual(proc.returncode, 0, proc.stderr) + self.assertNotIn("first-btrfs", proc.stdout) + self.assertIn("second-any", proc.stdout) + self.assertIn("third-zfs", proc.stdout) + + def test_group_filter_selects_one_group(self): + proc = run_list(["--group", "alpha"], + scenario_dir=self.dir, fs_profile="btrfs") + self.assertEqual(proc.returncode, 0, proc.stderr) + self.assertIn("first-btrfs", proc.stdout) + self.assertNotIn("second-any", proc.stdout) + + def test_unknown_group_is_an_error(self): + proc = run_list(["--group", "nonesuch"], + scenario_dir=self.dir, fs_profile="btrfs") + self.assertNotEqual(proc.returncode, 0) + self.assertIn("nonesuch", proc.stdout + proc.stderr) + + def test_groups_appear_in_file_order(self): + proc = run_list(scenario_dir=self.dir, fs_profile="zfs") + out = proc.stdout + self.assertLess(out.index("group beta:"), out.index("group gamma:")) + + +class ContractValidationTests(unittest.TestCase): + """Malformed scenario files fail the plan, naming the file.""" + + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.dir = self.tmp.name + + def tearDown(self): + self.tmp.cleanup() + + def assert_plan_fails_naming(self, filename): + proc = run_list(scenario_dir=self.dir) + self.assertNotEqual(proc.returncode, 0, + f"plan accepted malformed {filename}") + self.assertIn(filename, proc.stdout + proc.stderr) + + def test_missing_desc_rejected(self): + write_scenario(self.dir, "10-no-desc.sh", body=( + 'SCENARIO_GROUP="g"\nSCENARIO_PROFILES="any"\n' + "scenario_break() { :; }\nscenario_fix() { :; }\n" + "scenario_assert() { :; }\n")) + self.assert_plan_fails_naming("10-no-desc.sh") + + def test_missing_function_rejected(self): + write_scenario(self.dir, "10-no-assert.sh", body=( + 'SCENARIO_DESC="d"\nSCENARIO_GROUP="g"\nSCENARIO_PROFILES="any"\n' + "scenario_break() { :; }\nscenario_fix() { :; }\n")) + self.assert_plan_fails_naming("10-no-assert.sh") + + def test_bad_profile_token_rejected(self): + write_scenario(self.dir, "10-bad-profile.sh", profiles="ext4") + self.assert_plan_fails_naming("10-bad-profile.sh") + + def test_empty_scenario_dir_is_an_error(self): + proc = run_list(scenario_dir=self.dir) + self.assertNotEqual(proc.returncode, 0) + + +class UsageTests(unittest.TestCase): + def test_unknown_flag_is_an_error_with_usage(self): + proc = run_list(["--bogus"], scenario_dir=SCENARIO_DIR) + self.assertNotEqual(proc.returncode, 0) + self.assertIn("Usage", proc.stdout + proc.stderr) + + +NSPAWN_RUNNER = os.path.join(REPO_ROOT, "scripts", "testing", + "run-maint-nspawn.sh") + + +class NspawnPlanTests(unittest.TestCase): + """The nspawn fast lane selects exactly the pacman-level (packages) + group from the shared scenario dir.""" + + def run_nspawn_list(self, scenario_dir=None): + env = dict(os.environ) + if scenario_dir is not None: + env["MAINT_SCENARIO_DIR"] = scenario_dir + return subprocess.run( + ["bash", NSPAWN_RUNNER, "--list"], + capture_output=True, text=True, env=env, cwd=REPO_ROOT, + ) + + def test_list_exits_zero(self): + proc = self.run_nspawn_list() + self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr) + + def test_list_selects_only_the_packages_group(self): + proc = self.run_nspawn_list() + listed = set() + for f in os.listdir(SCENARIO_DIR): + if not f.endswith(".sh"): + continue + name = f.split("-", 1)[1][:-3] if "-" in f else f[:-3] + group = subprocess.run( + ["bash", "-c", f'source "{os.path.join(SCENARIO_DIR, f)}"; ' + 'printf %s "$SCENARIO_GROUP"'], + capture_output=True, text=True).stdout + if group == "packages": + self.assertIn(name, proc.stdout, + f"packages scenario {f} missing") + listed.add(name) + else: + self.assertNotIn(name, proc.stdout, + f"non-packages scenario {f} listed") + self.assertTrue(listed, "no packages-group scenarios found") + + def test_unknown_flag_is_an_error_with_usage(self): + proc = subprocess.run(["bash", NSPAWN_RUNNER, "--bogus"], + capture_output=True, text=True, cwd=REPO_ROOT) + self.assertNotEqual(proc.returncode, 0) + self.assertIn("Usage", proc.stdout + proc.stderr) + + def test_bad_profile_token_rejected_like_the_vm_lane(self): + """Both runners enforce the same scenario contract — a profile typo + must not pass the nspawn plan and only surface in the VM lane.""" + with tempfile.TemporaryDirectory() as d: + write_scenario(d, "10-bad-profile.sh", group="packages", + profiles="ext4") + proc = self.run_nspawn_list(scenario_dir=d) + self.assertNotEqual(proc.returncode, 0) + self.assertIn("10-bad-profile.sh", proc.stdout + proc.stderr) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/network-diagnostics/test_network_diagnostics.py b/tests/network-diagnostics/test_network_diagnostics.py new file mode 100644 index 0000000..1a8073f --- /dev/null +++ b/tests/network-diagnostics/test_network_diagnostics.py @@ -0,0 +1,215 @@ +"""Tests for run_network_diagnostics in the VM testing harness. + +run_network_diagnostics is the VM install pre-flight network check. It +collects read-only facts (interfaces, default route, resolver) first and +unconditionally, then runs every reachability check -- DNS, HTTP egress, +TLS egress, Arch mirror, AUR -- accumulating failures and reporting them all +at the end. Facts are printed regardless of pass/fail, so a failed install +still leaves the evidence. Generic checks (DNS/egress/TLS) are kept separate +from Arch-specific checks (mirror/AUR) so a DNS failure is named as DNS, not +misattributed to the mirror. Returns 0 when all checks pass, non-zero +otherwise, preserving the caller's success/failure contract. + +These tests exercise the REAL function body (sourced out of +network-diagnostics.sh, not a copy) with: + - stub logging functions (section/step/info/success/error/warn) that just + echo, so output is assertable; + - a fake `sshpass` on PATH that dispatches on the remote command string and + returns canned exit codes driven by FAKE_*_FAIL env vars. This is the + system boundary -- the real function shells out through + `sshpass ... ssh ... "<remote cmd>"`, and the fake stands in for the VM. + +Run from repo root: + python3 -m unittest tests.network-diagnostics.test_network_diagnostics +""" + +import os +import shutil +import subprocess +import tempfile +import unittest + + +REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")) +NETDIAG = os.path.join( + REPO_ROOT, "scripts", "testing", "lib", "network-diagnostics.sh" +) + +# A fake sshpass. The real invocation is: +# sshpass -p <pw> ssh <opts> -p <port> root@<host> "<remote cmd>" +# so the remote command is always the last argument. This stub inspects it and +# returns a canned exit code per check, driven by FAKE_*_FAIL env vars. Fact +# commands (ip/route/resolv) always succeed and print sample output so the +# evidence-collection path is exercised. +FAKE_SSHPASS = r"""#!/bin/bash +cmd="${@: -1}" +case "$cmd" in + *"ip -brief addr"*) + echo "lo UNKNOWN 127.0.0.1/8" + echo "eth0 UP 10.0.2.15/24" + exit 0 ;; + *"ip route show default"*) + echo "default via 10.0.2.2 dev eth0" + exit 0 ;; + *"resolv.conf"*) + echo "nameserver 10.0.2.3" + exit 0 ;; + *"getent hosts"*) + [ "${FAKE_DNS_FAIL:-0}" = "1" ] && exit 2 + exit 0 ;; + *"https://archlinux.org"*) + [ "${FAKE_TLS_FAIL:-0}" = "1" ] && exit 7 + exit 0 ;; + *"http://archlinux.org"*) + [ "${FAKE_HTTP_FAIL:-0}" = "1" ] && exit 7 + exit 0 ;; + *"geo.mirror.pkgbuild.com"*) + [ "${FAKE_MIRROR_FAIL:-0}" = "1" ] && exit 1 + exit 0 ;; + *"aur.archlinux.org"*) + [ "${FAKE_AUR_FAIL:-0}" = "1" ] && exit 1 + exit 0 ;; + *) + exit 0 ;; +esac +""" + +# Stub logging functions plus the sourced real file, then call the function. +WRAPPER = r"""#!/bin/bash +section() { echo "=== $1 ==="; } +step() { echo " -> $1"; } +info() { echo "[i] $1"; } +success() { echo "[OK] $1"; } +warn() { echo "[!] $1" >&2; } +error() { echo "[X] $1" >&2; } +source "$1" +run_network_diagnostics +""" + + +class NetworkDiagnosticsHarness(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.mkdtemp(prefix="netdiag-test-") + self.fakebin = os.path.join(self.tmp, "bin") + os.makedirs(self.fakebin) + sshpass = os.path.join(self.fakebin, "sshpass") + with open(sshpass, "w") as f: + f.write(FAKE_SSHPASS) + os.chmod(sshpass, 0o755) + self.wrapper = os.path.join(self.tmp, "run.sh") + with open(self.wrapper, "w") as f: + f.write(WRAPPER) + os.chmod(self.wrapper, 0o755) + + def tearDown(self): + shutil.rmtree(self.tmp, ignore_errors=True) + + def run_diag(self, results_dir=None, **fail_flags): + env = dict(os.environ) + env["PATH"] = self.fakebin + os.pathsep + env["PATH"] + # Keep the harness deterministic regardless of the host's SSH config. + env["SSH_OPTS"] = "-o StrictHostKeyChecking=no" + env["ROOT_PASSWORD"] = "archsetup" + env["SSH_PORT"] = "22" + env["VM_IP"] = "localhost" + if results_dir is not None: + env["TEST_RESULTS_DIR"] = results_dir + for k, v in fail_flags.items(): + env[k] = v + return subprocess.run( + ["bash", self.wrapper, NETDIAG], + capture_output=True, text=True, timeout=20, env=env, + ) + + # --- Normal case: everything reachable ----------------------------- + + def test_all_checks_pass_returns_zero(self): + r = self.run_diag() + self.assertEqual(r.returncode, 0, r.stdout + r.stderr) + self.assertIn("all checks passed", r.stdout) + + def test_facts_collected_on_success(self): + r = self.run_diag() + out = r.stdout + r.stderr + self.assertIn("10.0.2.15/24", out) # interface fact + self.assertIn("default via 10.0.2.2", out) # route fact + self.assertIn("nameserver 10.0.2.3", out) # resolver fact + + # --- DNS-failure case ---------------------------------------------- + + def test_dns_failure_returns_nonzero(self): + r = self.run_diag(FAKE_DNS_FAIL="1") + self.assertNotEqual(r.returncode, 0) + + def test_dns_failure_names_dns_not_mirror(self): + r = self.run_diag(FAKE_DNS_FAIL="1") + out = r.stdout + r.stderr + self.assertIn("DNS resolution failed", out) + # A DNS failure must not be misreported as a mirror failure. With only + # DNS failing, the mirror check still runs and passes. + self.assertNotIn("Cannot reach Arch mirrors", out) + + def test_dns_failure_still_collects_evidence(self): + # The whole point of the change: evidence is gathered before any check + # can bail, so a DNS failure still leaves the facts in the output. + r = self.run_diag(FAKE_DNS_FAIL="1") + out = r.stdout + r.stderr + self.assertIn("10.0.2.15/24", out) + self.assertIn("default via 10.0.2.2", out) + self.assertIn("nameserver 10.0.2.3", out) + + def test_dns_failure_summary_lists_the_failure(self): + r = self.run_diag(FAKE_DNS_FAIL="1") + out = r.stdout + r.stderr + self.assertIn("found 1 failure", out) + self.assertIn("getent hosts archlinux.org", out) + + # --- Mirror-only-failure case -------------------------------------- + + def test_mirror_only_failure_returns_nonzero(self): + r = self.run_diag(FAKE_MIRROR_FAIL="1") + self.assertNotEqual(r.returncode, 0) + + def test_mirror_only_failure_generic_checks_pass(self): + r = self.run_diag(FAKE_MIRROR_FAIL="1") + out = r.stdout + r.stderr + # Generic checks are healthy; only the Arch-specific mirror check fails. + self.assertIn("DNS resolution OK", out) + self.assertIn("HTTP egress OK", out) + self.assertIn("TLS/HTTPS egress OK", out) + self.assertIn("Cannot reach Arch mirrors", out) + self.assertNotIn("DNS resolution failed", out) + + def test_mirror_only_failure_summary_names_mirror(self): + r = self.run_diag(FAKE_MIRROR_FAIL="1") + out = r.stdout + r.stderr + self.assertIn("geo.mirror.pkgbuild.com", out) + + # --- All checks run: multiple failures are all reported ------------ + + def test_multiple_failures_all_reported(self): + r = self.run_diag(FAKE_DNS_FAIL="1", FAKE_AUR_FAIL="1") + out = r.stdout + r.stderr + self.assertIn("found 2 failure", out) + self.assertIn("getent hosts archlinux.org", out) + self.assertIn("aur.archlinux.org", out) + + # --- Raw outputs saved to the results dir -------------------------- + + def test_raw_facts_saved_to_results_dir(self): + results = os.path.join(self.tmp, "results") + os.makedirs(results) + self.run_diag(results_dir=results) + for slug, needle in ( + ("ip-addr", "10.0.2.15/24"), + ("ip-route", "default via 10.0.2.2"), + ("resolv-conf", "nameserver 10.0.2.3"), + ): + path = os.path.join(results, "netdiag-%s.txt" % slug) + self.assertTrue(os.path.exists(path), "missing " + path) + with open(path) as f: + self.assertIn(needle, f.read()) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/nvidia-preflight/test_nvidia_preflight.py b/tests/nvidia-preflight/test_nvidia_preflight.py new file mode 100644 index 0000000..bdacfd5 --- /dev/null +++ b/tests/nvidia-preflight/test_nvidia_preflight.py @@ -0,0 +1,162 @@ +"""Tests for the nvidia_preflight_report helper in the archsetup installer. + +nvidia_preflight_report is the pure core of the NVIDIA/Wayland preflight +check: it scans DRM (then PCI display-class) modalias files for the NVIDIA +vendor id, and when one matches it prints the Wayland warning + required +environment variables and checks the repo's candidate nvidia-utils major +version. Return codes: 0 = no NVIDIA GPU, 10 = NVIDIA and the driver +requirement (535+) is met, 11 = NVIDIA and the requirement is not met +(driver too old or unknown). The interactive continue/abort prompt lives in +preflight_checks, not here, so this core is unit testable. + +These tests exercise the REAL function body, extracted from the `archsetup` +script at run time (not a copy), against temp modalias trees and a fake +pacman on PATH. + +Run from repo root: + python3 -m unittest tests.nvidia-preflight.test_nvidia_preflight +""" + +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_MODALIAS = "pci:v000010DEd00002684sv00001043sd000088E2bc03sc00i00" +NVIDIA_MODALIAS_LOWER = "pci:v000010ded00002684sv00001043sd000088e2bc03sc00i00" +AMD_MODALIAS = "pci:v00001002d0000164Esv00001462sd00007D78bc03sc80i00" +NON_DISPLAY_NVIDIA = "pci:v000010DEd00002684sv00001043sd000088E2bc0Csc03i30" + + +class NvidiaPreflightHarness(unittest.TestCase): + """Source nvidia_preflight_report out of the real archsetup script.""" + + def setUp(self): + self.tmp = tempfile.mkdtemp(prefix="nvidia-preflight-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.fakebin = os.path.join(self.tmp, "bin") + os.makedirs(self.fakebin) + 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_report() {/,/^}/p' \"$ARCHSETUP\")\n" + "nvidia_preflight_report\n" + ) + os.chmod(self.wrapper, 0o755) + + def tearDown(self): + shutil.rmtree(self.tmp, ignore_errors=True) + + def fake_pacman(self, version=None, fail=False): + """A pacman stub answering `pacman -Si nvidia-utils`.""" + path = os.path.join(self.fakebin, "pacman") + with open(path, "w") as f: + if fail: + f.write("#!/bin/sh\nexit 1\n") + else: + f.write( + "#!/bin/sh\n" + "printf 'Repository : extra\\n'\n" + "printf 'Name : nvidia-utils\\n'\n" + "printf 'Version : %s\\n'\n" % version + ) + os.chmod(path, 0o755) + + def add_modalias(self, root, subdir, content): + d = os.path.join(root, subdir) + os.makedirs(d, exist_ok=True) + with open(os.path.join(d, "modalias"), "w") as f: + f.write(content + "\n") + + def run_check(self): + env = dict(os.environ) + env["PATH"] = self.fakebin + os.pathsep + env["PATH"] + env["NVIDIA_DRM_GLOB"] = os.path.join(self.drm, "card*", "modalias") + env["NVIDIA_PCI_GLOB"] = os.path.join(self.pci, "*", "modalias") + return subprocess.run( + ["bash", self.wrapper, ARCHSETUP], + capture_output=True, text=True, env=env, + ) + + # ---------------------------------------------------------- normal ---- + def test_no_gpu_files_returns_zero_and_silent(self): + self.fake_pacman(version="575.51.02-1") + r = self.run_check() + self.assertEqual(r.returncode, 0) + self.assertNotIn("NVIDIA", r.stdout) + + def test_amd_only_returns_zero(self): + self.fake_pacman(version="575.51.02-1") + self.add_modalias(self.drm, "card0", AMD_MODALIAS) + r = self.run_check() + self.assertEqual(r.returncode, 0) + self.assertNotIn("NVIDIA", r.stdout) + + def test_nvidia_with_modern_driver_returns_ten_with_guidance(self): + self.fake_pacman(version="575.51.02-1") + self.add_modalias(self.drm, "card0", NVIDIA_MODALIAS) + r = self.run_check() + self.assertEqual(r.returncode, 10) + self.assertIn("NVIDIA GPU detected", r.stdout) + self.assertIn("LIBVA_DRIVER_NAME=nvidia", r.stdout) + self.assertIn("GBM_BACKEND=nvidia-drm", r.stdout) + self.assertIn("__GLX_VENDOR_LIBRARY_NAME=nvidia", r.stdout) + self.assertIn("575.51.02-1", r.stdout) + + # -------------------------------------------------------- boundary ---- + def test_lowercase_vendor_id_detected(self): + self.fake_pacman(version="575.51.02-1") + self.add_modalias(self.drm, "card0", NVIDIA_MODALIAS_LOWER) + r = self.run_check() + self.assertEqual(r.returncode, 10) + + def test_exactly_535_meets_requirement(self): + self.fake_pacman(version="535.216.01-1") + self.add_modalias(self.drm, "card0", NVIDIA_MODALIAS) + r = self.run_check() + self.assertEqual(r.returncode, 10) + + def test_pci_fallback_display_class_only(self): + # No DRM entries; PCI holds a display-class NVIDIA device -> detected. + self.fake_pacman(version="575.51.02-1") + self.add_modalias(self.pci, "0000:01:00.0", NVIDIA_MODALIAS) + r = self.run_check() + self.assertEqual(r.returncode, 10) + + def test_pci_non_display_nvidia_ignored(self): + # An NVIDIA audio/usb function (bc0C) must not trigger the check. + self.fake_pacman(version="575.51.02-1") + self.add_modalias(self.pci, "0000:01:00.1", NON_DISPLAY_NVIDIA) + r = self.run_check() + self.assertEqual(r.returncode, 0) + + # ----------------------------------------------------------- error ---- + def test_old_driver_returns_eleven_with_error(self): + self.fake_pacman(version="470.256.02-1") + self.add_modalias(self.drm, "card0", NVIDIA_MODALIAS) + r = self.run_check() + self.assertEqual(r.returncode, 11) + self.assertIn("535", r.stdout) + self.assertIn("470.256.02-1", r.stdout) + + def test_pacman_failure_returns_eleven_unknown(self): + self.fake_pacman(fail=True) + self.add_modalias(self.drm, "card0", NVIDIA_MODALIAS) + r = self.run_check() + self.assertEqual(r.returncode, 11) + self.assertIn("unknown", r.stdout) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/run-task/test_run_task.py b/tests/run-task/test_run_task.py new file mode 100644 index 0000000..35036dd --- /dev/null +++ b/tests/run-task/test_run_task.py @@ -0,0 +1,172 @@ +"""Tests for the run_task / enable_service helpers in the archsetup installer. + +run_task is the installer's describe-run-warn primitive. It replaces the +hand-written idiom that recurs ~100 times across the script: + + action="enabling rngd service" && display "task" "$action" + systemctl enable rngd >> "$logfile" 2>&1 || error_warn "$action" "$?" + +as a single call: + + run_task "enabling rngd service" systemctl enable rngd + +It announces the task via display, runs the command with stdout+stderr +appended to $logfile, and on failure calls error_warn with the command's +real exit code (non-fatal). enable_service is a thin wrapper that enables +one or more systemd units with the conventional "enabling <unit> service" +wording. + +These tests exercise the REAL function bodies, extracted from the +`archsetup` script at run time (not a copy), with recording stubs standing +in for display, error_warn, and systemctl. The command run by run_task is +genuinely executed. + +Run from repo root: + python3 -m unittest tests.run-task.test_run_task +""" + +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") + +# A bash harness that sources the real run_task + enable_service out of the +# installer, with recording stubs for their dependencies. Each stub appends a +# tab-separated record to a file named by an env var, so the Python side can +# assert what was called. The real command passed to run_task still runs. +WRAPPER = r"""#!/bin/bash +ARCHSETUP="$1"; shift +logfile="$LOGFILE" + +display() { printf '%s\t%s\n' "$1" "$2" >> "$DISPLAY_LOG"; } +error_warn() { printf '%s\t%s\n' "$1" "$2" >> "$ERRWARN_LOG"; return 1; } +systemctl() { printf 'systemctl %s\n' "$*"; } + +source <(sed -n '/^run_task() {/,/^}/p' "$ARCHSETUP") +source <(sed -n '/^enable_service() {/,/^}/p' "$ARCHSETUP") + +"$@" +""" + + +class RunTaskHarness(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.mkdtemp(prefix="run-task-test-") + self.wrapper = os.path.join(self.tmp, "run.sh") + with open(self.wrapper, "w") as f: + f.write(WRAPPER) + os.chmod(self.wrapper, 0o755) + self.logfile = os.path.join(self.tmp, "install.log") + self.display_log = os.path.join(self.tmp, "display.log") + self.errwarn_log = os.path.join(self.tmp, "errwarn.log") + + def tearDown(self): + shutil.rmtree(self.tmp, ignore_errors=True) + + def call(self, *args): + env = dict(os.environ) + env["LOGFILE"] = self.logfile + env["DISPLAY_LOG"] = self.display_log + env["ERRWARN_LOG"] = self.errwarn_log + return subprocess.run( + ["bash", self.wrapper, ARCHSETUP, *args], + capture_output=True, text=True, timeout=10, env=env, + ) + + def read(self, path): + if not os.path.exists(path): + return "" + with open(path) as f: + return f.read() + + # --- Normal cases ----------------------------------------------------- + + def test_run_task_success_announces_and_runs(self): + result = self.call("run_task", "doing a thing", "true") + self.assertEqual(result.returncode, 0, result.stderr) + # Announced as a "task" with the exact description. + self.assertEqual(self.read(self.display_log), "task\tdoing a thing\n") + # No warning on success. + self.assertEqual(self.read(self.errwarn_log), "") + + def test_run_task_captures_command_output_to_logfile(self): + result = self.call("run_task", "echo something", "echo", "hello-from-cmd") + self.assertEqual(result.returncode, 0, result.stderr) + self.assertIn("hello-from-cmd", self.read(self.logfile)) + # Command output is logged, not printed to the terminal. + self.assertNotIn("hello-from-cmd", result.stdout) + + def test_run_task_captures_stderr_to_logfile(self): + # `ls` of a missing path writes to stderr; it must land in the logfile. + missing = os.path.join(self.tmp, "no-such-path") + self.call("run_task", "listing", "ls", missing) + self.assertIn("no-such-path", self.read(self.logfile)) + + def test_run_task_preserves_multiple_arguments(self): + self.call("run_task", "multi-arg", "printf", "%s|%s|%s", "a", "b", "c") + self.assertIn("a|b|c", self.read(self.logfile)) + + def test_run_task_preserves_arguments_with_spaces(self): + self.call("run_task", "spacey", "printf", "[%s]", "two words") + self.assertIn("[two words]", self.read(self.logfile)) + + # --- enable_service --------------------------------------------------- + + def test_enable_service_single_unit(self): + self.call("enable_service", "rngd") + self.assertEqual(self.read(self.display_log), "task\tenabling rngd service\n") + self.assertIn("systemctl enable rngd", self.read(self.logfile)) + + def test_enable_service_multiple_units(self): + self.call("enable_service", "foo", "bar", "baz") + disp = self.read(self.display_log) + self.assertIn("task\tenabling foo service\n", disp) + self.assertIn("task\tenabling bar service\n", disp) + self.assertIn("task\tenabling baz service\n", disp) + log = self.read(self.logfile) + self.assertIn("systemctl enable foo", log) + self.assertIn("systemctl enable bar", log) + self.assertIn("systemctl enable baz", log) + + # --- Error cases ------------------------------------------------------ + + def test_run_task_failure_warns_with_description(self): + result = self.call("run_task", "failing thing", "false") + self.assertNotEqual(result.returncode, 0) + self.assertEqual(self.read(self.errwarn_log), "failing thing\t1\n") + + def test_run_task_failure_propagates_real_exit_code(self): + # `bash -c 'exit 42'` must surface 42 to error_warn, not a clobbered 0. + self.call("run_task", "exit-42", "bash", "-c", "exit 42") + self.assertEqual(self.read(self.errwarn_log), "exit-42\t42\n") + + def test_enable_service_failure_warns_per_unit(self): + # Override systemctl to fail; each unit should produce a warning. + env = dict(os.environ) + env["LOGFILE"] = self.logfile + env["DISPLAY_LOG"] = self.display_log + env["ERRWARN_LOG"] = self.errwarn_log + # Re-create wrapper with a failing systemctl stub for this case. + failing = os.path.join(self.tmp, "run-fail.sh") + with open(failing, "w") as f: + f.write(WRAPPER.replace( + "systemctl() { printf 'systemctl %s\\n' \"$*\"; }", + "systemctl() { printf 'systemctl %s\\n' \"$*\"; return 1; }", + )) + os.chmod(failing, 0o755) + subprocess.run( + ["bash", failing, ARCHSETUP, "enable_service", "alpha", "beta"], + capture_output=True, text=True, timeout=10, env=env, + ) + warns = self.read(self.errwarn_log) + self.assertIn("enabling alpha service\t1\n", warns) + self.assertIn("enabling beta service\t1\n", warns) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/zfs-pre-snapshot/fake-zfs b/tests/zfs-pre-snapshot/fake-zfs new file mode 100755 index 0000000..508c0f3 --- /dev/null +++ b/tests/zfs-pre-snapshot/fake-zfs @@ -0,0 +1,14 @@ +#!/bin/sh +# Fake zfs for the zfs-pre-snapshot unit test. `snapshot` and `destroy` are +# logged (FAKE_ZFS_LOG); `list` prints a fixture snapshot set (FAKE_ZFS_SNAPSHOTS). +# Set FAKE_ZFS_SNAPSHOT_FAIL to make snapshot creation fail. +case "$1" in + snapshot) + [ -n "$FAKE_ZFS_SNAPSHOT_FAIL" ] && exit 1 + echo "snapshot $2" >> "$FAKE_ZFS_LOG"; exit 0 ;; + destroy) + echo "destroy $2" >> "$FAKE_ZFS_LOG"; exit 0 ;; + list) + cat "$FAKE_ZFS_SNAPSHOTS" 2>/dev/null; exit 0 ;; +esac +exit 0 diff --git a/tests/zfs-pre-snapshot/test_zfs_pre_snapshot.py b/tests/zfs-pre-snapshot/test_zfs_pre_snapshot.py new file mode 100644 index 0000000..ed7731b --- /dev/null +++ b/tests/zfs-pre-snapshot/test_zfs_pre_snapshot.py @@ -0,0 +1,116 @@ +"""Unit tests for scripts/zfs-pre-snapshot. + +The script snapshots the root dataset before a pacman transaction and prunes to +the most recent KEEP pre-pacman snapshots. These tests drive the real script +with a fake zfs on PATH (snapshot/destroy logged, list returns a fixture set) +and env-rooted state, so nothing touches a real pool. +""" + +import os +import shutil +import subprocess +import tempfile +import time +import unittest + +REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")) +SCRIPT = os.path.join(REPO_ROOT, "scripts/zfs-pre-snapshot") +FAKE_ZFS = os.path.join(os.path.dirname(__file__), "fake-zfs") + +DATASET = "tank/test" +# Five pre-pacman snapshots oldest->newest (zfs list -s creation is ascending), +# plus one autosnap that the grep filter must ignore. +SNAPSHOTS = "\n".join([ + f"{DATASET}@autosnap_2026-01-01", + f"{DATASET}@pre-pacman_2026-06-01", + f"{DATASET}@pre-pacman_2026-06-02", + f"{DATASET}@pre-pacman_2026-06-03", + f"{DATASET}@pre-pacman_2026-06-04", + f"{DATASET}@pre-pacman_2026-06-05", +]) + "\n" + + +class Harness(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.mkdtemp(prefix="zfs-pre-snap-") + self.bin = os.path.join(self.tmp, "bin") + os.makedirs(self.bin) + shutil.copy(FAKE_ZFS, os.path.join(self.bin, "zfs")) + self.log = os.path.join(self.tmp, "zfs.log") + self.snaps = os.path.join(self.tmp, "snaps") + with open(self.snaps, "w") as f: + f.write(SNAPSHOTS) + self.lock = os.path.join(self.tmp, "lock") + + def tearDown(self): + shutil.rmtree(self.tmp, ignore_errors=True) + + def run_script(self, keep="3", fail=False, snaps=None): + env = os.environ.copy() + env["PATH"] = self.bin + os.pathsep + env["PATH"] + env["ZFS_PRE_DATASET"] = DATASET + env["ZFS_PRE_LOCKFILE"] = self.lock + env["ZFS_PRE_KEEP"] = keep + env["FAKE_ZFS_LOG"] = self.log + env["FAKE_ZFS_SNAPSHOTS"] = snaps if snaps is not None else self.snaps + if fail: + env["FAKE_ZFS_SNAPSHOT_FAIL"] = "1" + return subprocess.run([SCRIPT], env=env, capture_output=True, text=True, + timeout=15) + + def log_lines(self): + try: + with open(self.log) as f: + return [ln for ln in f.read().splitlines() if ln.strip()] + except FileNotFoundError: + return [] + + +class TestSnapshot(Harness): + def test_creates_a_pre_pacman_snapshot(self): + self.run_script() + snaps = [ln for ln in self.log_lines() if ln.startswith("snapshot ")] + self.assertEqual(len(snaps), 1) + self.assertIn(f"snapshot {DATASET}@pre-pacman_", snaps[0]) + + def test_skips_when_lockfile_is_fresh(self): + # A lockfile newer than MIN_INTERVAL → no snapshot this run. + open(self.lock, "w").close() + os.utime(self.lock, (time.time(), time.time())) + self.run_script() + self.assertEqual([ln for ln in self.log_lines() + if ln.startswith("snapshot ")], []) + + +class TestPrune(Harness): + def test_prunes_oldest_beyond_keep(self): + # 5 pre-pacman snapshots, KEEP=3 → the two oldest are destroyed. + self.run_script(keep="3") + destroyed = [ln.split(" ", 1)[1] for ln in self.log_lines() + if ln.startswith("destroy ")] + self.assertEqual(destroyed, + [f"{DATASET}@pre-pacman_2026-06-01", + f"{DATASET}@pre-pacman_2026-06-02"]) + + def test_never_destroys_non_pre_pacman_snapshots(self): + self.run_script(keep="1") + destroyed = [ln for ln in self.log_lines() if ln.startswith("destroy ")] + self.assertFalse(any("autosnap" in ln for ln in destroyed)) + + def test_no_prune_when_at_or_under_keep(self): + # KEEP=5 with exactly 5 pre-pacman snapshots → nothing destroyed. + self.run_script(keep="5") + self.assertEqual([ln for ln in self.log_lines() + if ln.startswith("destroy ")], []) + + +class TestError(Harness): + def test_snapshot_failure_skips_prune_and_warns(self): + r = self.run_script(fail=True) + self.assertIn("Failed to create snapshot", r.stderr) + self.assertEqual([ln for ln in self.log_lines() + if ln.startswith("destroy ")], []) + + +if __name__ == "__main__": + unittest.main() |