1 files changed, 273 insertions, 0 deletions
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()