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
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
|
"""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()
# Helper -> the function that must call it. Every one of these helpers has its
# own suite proving it behaves correctly; none of those suites proves it is
# wired in. Deleting the call left every test green in five separate suites,
# including the one guarding a `pacman -Rdd` of twelve firmware packages.
#
# This is a static check on purpose. The behavioural harness above stubs only
# the names it is given and runs the rest of the body for real, which is safe
# for an orchestrator whose body is nothing but calls, and emphatically not
# safe for a leaf like trim_firmware. Cheaper and machine-independent too: the
# alternative depends on what /proc/cpuinfo says on the box running the tests.
CALL_SITES = {
"trim_firmware": ["detect_gpu_vendors", "is_framework_intel_laptop"],
"finalize_dotfiles": ["mark_volatile_configs"],
"preflight_checks": ["select_locale", "nvidia_preflight", "check_disk_space"],
"configure_encrypted_autologin": ["configure_autologin"],
"configure_initramfs_hook": ["switch_udev_hook_to_systemd"],
"add_nvme_early_module": ["ensure_nvme_early_module"],
"configure_snapshots": ["configure_zfs_snapshots", "configure_btrfs_snapshots"],
"configure_zfs_snapshots": ["zfs_scrub_timer_units"],
"install_maintenance_config": ["install_maintenance_thresholds"],
"hyprland": ["configure_hyprlock_pam"],
"update_grub_cmdline": ["detect_gpu_vendors"],
}
def function_body(name):
"""The body of `name` from archsetup, comments and blank lines stripped, so
a match is a real invocation rather than a mention in a comment."""
with open(ARCHSETUP) as f:
src = f.read()
start = src.index(f"\n{name}() {{\n")
end = src.index("\n}\n", start)
lines = src[start:end].splitlines()[1:]
keep = []
for line in lines:
stripped = line.strip()
if not stripped or stripped.startswith("#"):
continue
keep.append(line.split(" #", 1)[0])
return "\n".join(keep)
class CallSiteWiring(unittest.TestCase):
def test_every_guarded_helper_is_actually_called(self):
for caller, callees in CALL_SITES.items():
body = function_body(caller)
for callee in callees:
with self.subTest(caller=caller, callee=callee):
self.assertRegex(
body, rf"(^|[\s;&|(]){callee}\b",
f"{caller} no longer calls {callee}; its own suite "
f"cannot see this")
def test_the_check_would_notice_a_missing_call(self):
# The check is only worth having if it fails when the call goes away.
body = function_body("trim_firmware").replace("detect_gpu_vendors", "")
with self.assertRaises(AssertionError):
self.assertRegex(body, r"(^|[\s;&|(])detect_gpu_vendors\b")
|
