path: root/tests/maint-scenarios/test_scenario_plan.py

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