diff options
Diffstat (limited to 'tests')
| -rw-r--r-- | tests/hypr-live-update-guard/test_hypr_live_update_guard.py | 72 | ||||
| -rw-r--r-- | tests/installer-steps/test_orchestrators.py | 33 | ||||
| -rw-r--r-- | tests/installer-steps/test_pacman_install.py | 95 | ||||
| -rw-r--r-- | tests/maint-scenarios/test_scenario_plan.py | 273 |
4 files changed, 471 insertions, 2 deletions
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 index 5ec5ce8..a6c6f68 100644 --- a/tests/hypr-live-update-guard/test_hypr_live_update_guard.py +++ b/tests/hypr-live-update-guard/test_hypr_live_update_guard.py @@ -12,6 +12,10 @@ 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 @@ -26,8 +30,18 @@ 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): + +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: @@ -35,6 +49,7 @@ def run_guard(stdin="mesa\n", running="1", allow=None, sentinel=None): # 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, @@ -75,6 +90,42 @@ class HyprLiveUpdateGuard(unittest.TestCase): 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): @@ -86,6 +137,25 @@ class HyprLiveUpdateGuard(unittest.TestCase): 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) diff --git a/tests/installer-steps/test_orchestrators.py b/tests/installer-steps/test_orchestrators.py index 48b7508..2a771ba 100644 --- a/tests/installer-steps/test_orchestrators.py +++ b/tests/installer-steps/test_orchestrators.py @@ -51,7 +51,8 @@ ORCHESTRATORS = { "user_customizations": [ "clone_user_repos", "stow_dotfiles", "prune_waybar_battery", "refresh_desktop_caches", "configure_dconf_defaults", - "finalize_dotfiles", "create_user_directories", + "finalize_dotfiles", "install_maintenance_config", + "create_user_directories", ], } @@ -114,5 +115,35 @@ class SnapshotDispatch(unittest.TestCase): 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() |