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