"""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()