1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
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()
|
