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
|
"""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",
],
"user_customizations": [
"clone_user_repos", "stow_dotfiles", "prune_waybar_battery",
"refresh_desktop_caches", "configure_dconf_defaults",
"finalize_dotfiles", "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(), [])
if __name__ == "__main__":
unittest.main()
|
