diff options
Diffstat (limited to 'tests')
19 files changed, 3569 insertions, 6 deletions
diff --git a/tests/bluetooth-resume/test_bluetooth_resume.py b/tests/bluetooth-resume/test_bluetooth_resume.py new file mode 100644 index 0000000..62fd24f --- /dev/null +++ b/tests/bluetooth-resume/test_bluetooth_resume.py @@ -0,0 +1,141 @@ +"""Tests for scripts/zz-bluetooth-resume. + +Two things break bluetooth across a sleep cycle on a TLP laptop, and nothing +else on the machine fixes either. + +The rfkill soft-block is not cleared. systemd-rfkill would do it, but it is +masked deliberately -- it fights TLP's radio handling, so TLP owns radios +instead. TLP's own sleep hook runs `tlp resume`, but that only puts bluetooth +back where it was at suspend and applies no policy: its setting is +DEVICES_TO_ENABLE_ON_STARTUP, startup not resume, and there is no ON_RESUME in +TLP's vocabulary. A block present at suspend stays, and so does one that a +controller re-probed after that replay comes up with. WiFi survives only +because NetworkManager unblocks itself; bluetooth has no equivalent. + +The controller also comes back wedged from a hibernate. It reports powered and +unblocked while scanning finds nothing whatever -- zero devices where the same +room gave seventeen a minute later. bluetoothd logs "Failed to set mode" and +"Failed to add device <mac>" at the instant of resume. Reloading btusb clears +it. + +Both observed on velox 2026-08-21, on its first suspend-then-hibernate cycle +after hibernate was switched back on. + +The hook re-asserts TLP's own declared intent rather than inventing a policy, +so a machine that deliberately blocks bluetooth keeps it blocked. + +Run from repo root: + python3 -m unittest tests.bluetooth-resume.test_bluetooth_resume +""" + +import os +import stat +import subprocess +import tempfile +import unittest + +REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")) +HOOK = os.path.join(REPO_ROOT, "scripts", "zz-bluetooth-resume") + +TLP_WANTS_BT = 'DEVICES_TO_ENABLE_ON_STARTUP="bluetooth wifi"\n' +TLP_WIFI_ONLY = 'DEVICES_TO_ENABLE_ON_STARTUP="wifi"\n' + + +def run(phase="post", kind="suspend-then-hibernate", tlp_conf=TLP_WANTS_BT, + conf_present=True): + """Drive the hook with rfkill and modprobe faked, and read back the calls.""" + with tempfile.TemporaryDirectory() as d: + calls = os.path.join(d, "calls.log") + bindir = os.path.join(d, "bin") + os.makedirs(bindir) + for tool in ("rfkill", "modprobe"): + p = os.path.join(bindir, tool) + with open(p, "w") as fh: + fh.write(f'#!/bin/sh\necho "{tool} $*" >> "{calls}"\nexit 0\n') + os.chmod(p, 0o755) + conf = os.path.join(d, "tlp.conf") + if conf_present: + with open(conf, "w") as fh: + fh.write(tlp_conf) + env = dict(os.environ) + env.update({ + "BTR_RFKILL": os.path.join(bindir, "rfkill"), + "BTR_MODPROBE": os.path.join(bindir, "modprobe"), + "BTR_TLP_CONF": conf, + "BTR_TLP_CONF_DIR": os.path.join(d, "tlp.d"), + "BTR_SETTLE": "0", + }) + r = subprocess.run(["sh", HOOK, phase, kind], env=env, + capture_output=True, text=True, timeout=20) + log = "" + if os.path.exists(calls): + with open(calls) as fh: + log = fh.read() + return r, log + + +class BluetoothResume(unittest.TestCase): + # --- Normal --------------------------------------------------------- + def test_hibernate_reloads_the_driver_and_unblocks(self): + _, log = run(kind="suspend-then-hibernate") + self.assertIn("modprobe -r btusb", log) + self.assertIn("modprobe btusb", log) + self.assertIn("rfkill unblock bluetooth", log) + + def test_the_unblock_comes_after_the_reload(self): + # A freshly loaded btusb can come up soft-blocked, so unblocking first + # would be undone by the reload that follows it. + _, log = run() + self.assertLess(log.index("modprobe btusb"), + log.index("rfkill unblock")) + + def test_plain_suspend_unblocks_without_reloading(self): + # The wedge was seen coming out of hibernate, which reinitialises the + # controller from a saved image. A plain suspend restores USB intact, + # so reloading there would cost a working adapter for nothing. + _, log = run(kind="suspend") + self.assertIn("rfkill unblock bluetooth", log) + self.assertNotIn("btusb", log) + + # --- Boundary ------------------------------------------------------- + def test_the_pre_phase_does_nothing(self): + _, log = run(phase="pre") + self.assertEqual(log, "") + + def test_a_tlp_policy_without_bluetooth_is_left_alone(self): + # The hook re-asserts TLP's stated intent. It must not invent one, or + # a machine that deliberately keeps bluetooth off gets it turned on at + # every wakeup. + _, log = run(tlp_conf=TLP_WIFI_ONLY) + self.assertEqual(log, "") + + def test_a_commented_out_policy_does_not_count(self): + _, log = run(tlp_conf='#DEVICES_TO_ENABLE_ON_STARTUP="bluetooth"\n') + self.assertEqual(log, "") + + def test_hibernate_proper_also_reloads(self): + _, log = run(kind="hibernate") + self.assertIn("modprobe -r btusb", log) + + # --- Error ---------------------------------------------------------- + def test_a_missing_tlp_config_is_left_alone(self): + # No declared policy means no intent to re-assert. Failing safe here + # means doing nothing, not guessing. + _, log = run(conf_present=False) + self.assertEqual(log, "") + + def test_the_hook_always_exits_zero(self): + # systemd-sleep logs a failing hook and the noise outlives the cause. + # Nothing here is worth delaying or alarming a resume over. + for kind in ("suspend", "hibernate", "suspend-then-hibernate"): + with self.subTest(kind=kind): + r, _ = run(kind=kind) + self.assertEqual(r.returncode, 0, r.stderr) + + def test_it_is_executable(self): + self.assertTrue(os.stat(HOOK).st_mode & stat.S_IXUSR, + "systemd-sleep only runs executables") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/installer-steps/test_clone_user_repos.py b/tests/installer-steps/test_clone_user_repos.py new file mode 100644 index 0000000..51d8434 --- /dev/null +++ b/tests/installer-steps/test_clone_user_repos.py @@ -0,0 +1,155 @@ +"""Test clone_user_repos: the two user repos are cloned with full history. + +archsetup and dotfiles are not build directories. They are the two repos I +actively develop in on every machine this installer builds, so a shallow clone +is wrong for both. Velox came back from its 2026-08-13 rebuild with 7 commits +of history in each instead of 851, and nothing about the tree said so. + +The quiet failure is what makes this worth a test rather than a one-line fix. +`git log -- <path>` against a shallow clone does not error; it answers "no +commits". So a credential-history check run on that machine reported five +sensitive files absent from history and exited clean, when the real answer was +that the clone could not see the history they live in. A security question came +back falsely reassuring. Everything else it breaks — blame, bisect, any +archaeology past the graft point — is merely annoying by comparison. + +The AUR build clones are a different case and stay shallow: they are throwaway +build trees, cloned to run `make install` and then discarded, where history has +no value and the download cost is real. So this suite asserts both halves — +full history for the two user repos, and depth still pinned on the AUR path — +because a fix applied with too broad a brush would regress the build clones +without failing any test that only looked at the user repos. + +Method: sed-extract clone_user_repos from the real `archsetup`, fake git / +mkdir / chown / display / error_warn / error_fatal, and read back the git +command lines the function issued. + +Run from repo root: + python3 -m unittest tests.installer-steps.test_clone_user_repos +""" + +import os +import re +import subprocess +import tempfile +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(clone_fails=False, make_git_dir=True): + """Drive clone_user_repos with every side effect faked. + + dotfiles_dir is pre-created with a .git so the function's "is this a real + checkout?" guard passes on the happy path; make_git_dir=False exercises the + guard itself. + """ + with tempfile.TemporaryDirectory() as d: + dotfiles_dir = os.path.join(d, "dotfiles") + os.makedirs(dotfiles_dir) + if make_git_dir: + os.makedirs(os.path.join(dotfiles_dir, ".git")) + clone_rc = 1 if clone_fails else 0 + script = textwrap.dedent(f"""\ + logfile=/dev/null + action="" + username=testuser + archsetup_repo="https://example.invalid/archsetup.git" + dotfiles_repo="https://example.invalid/dotfiles.git" + dotfiles_branch=main + dotfiles_dir="{dotfiles_dir}" + display() {{ :; }} + mkdir() {{ echo "MKDIR: $*" >> "{d}/calls.log"; return 0; }} + chown() {{ echo "CHOWN: $*" >> "{d}/calls.log"; return 0; }} + git() {{ + echo "GIT: $*" >> "{d}/calls.log" + case "$1" in + clone) return {clone_rc} ;; + *) return 0 ;; + esac + }} + error_warn() {{ echo "WARN: $1" >> "{d}/calls.log"; return 1; }} + error_fatal() {{ echo "FATAL: $1" >> "{d}/calls.log"; exit 1; }} + source <(sed -n '/^clone_user_repos() {{/,/^}}/p' "{ARCHSETUP}") + clone_user_repos + echo "RC=$?" >> "{d}/calls.log" + exit 0 + """) + subprocess.run( + ["bash", "-c", script], capture_output=True, text=True, timeout=10, + ) + with open(os.path.join(d, "calls.log")) as fh: + return fh.read() + + +def clone_lines(log): + return [ln for ln in log.splitlines() if ln.startswith("GIT: clone")] + + +class CloneUserRepos(unittest.TestCase): + # ------------------------------------------------------------ normal ---- + def test_both_user_repos_are_cloned(self): + lines = clone_lines(run()) + self.assertEqual(len(lines), 2, + f"expected an archsetup clone and a dotfiles clone, got: {lines}") + self.assertTrue(any("archsetup.git" in ln for ln in lines)) + self.assertTrue(any("dotfiles.git" in ln for ln in lines)) + + def test_archsetup_clone_carries_full_history(self): + """A shallow archsetup clone answers history questions wrongly.""" + line = next(ln for ln in clone_lines(run()) if "archsetup.git" in ln) + self.assertNotIn("--depth", line, + "archsetup is a working repo, not a build tree — a shallow " + "clone makes `git log -- <path>` answer 'no commits' instead " + "of failing, which is how a credential-history check came " + "back falsely clean on velox") + + def test_dotfiles_clone_carries_full_history(self): + line = next(ln for ln in clone_lines(run()) if "dotfiles.git" in ln) + self.assertNotIn("--depth", line, + "dotfiles is a working repo, not a build tree") + + def test_dotfiles_clone_still_pins_the_branch(self): + """Dropping --depth must not disturb the --branch argument beside it.""" + line = next(ln for ln in clone_lines(run()) if "dotfiles.git" in ln) + self.assertIn("--branch main", line) + + # ---------------------------------------------------------- boundary ---- + def test_no_user_repo_clone_is_shallow_by_any_spelling(self): + """--depth, --depth=N and -depth are all shallow; catch the lot.""" + for line in clone_lines(run()): + self.assertNotRegex(line, r"(^|\s)-{1,2}depth(\s|=)", + f"user-repo clone must be full: {line}") + + def test_aur_build_clones_stay_shallow(self): + """The fix must not over-apply — build trees are throwaway. + + Read against the real file rather than the extracted function, because + these clones live in a different function entirely and the risk being + guarded is a careless repo-wide sed. + """ + with open(ARCHSETUP) as fh: + source = fh.read() + build_clones = re.findall(r"^.*git clone.*build_dir.*$", source, re.M) + self.assertTrue(build_clones, "expected AUR build clones to exist") + for line in build_clones: + self.assertIn("--depth 1", line, + f"AUR build clone should stay shallow: {line.strip()}") + + # ------------------------------------------------------------- error ---- + def test_clone_failure_is_reported_not_swallowed(self): + log = run(clone_fails=True) + self.assertIn("WARN:", log, + "a failed clone must surface through error_warn") + + def test_dotfiles_clone_producing_no_checkout_is_fatal(self): + """The stow/restore steps downstream need a real checkout.""" + log = run(make_git_dir=False) + self.assertIn("FATAL:", log) + self.assertNotIn("RC=", log, "error_fatal must halt, not fall through") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/installer-steps/test_configure_backlight_access.py b/tests/installer-steps/test_configure_backlight_access.py new file mode 100644 index 0000000..6fd171a --- /dev/null +++ b/tests/installer-steps/test_configure_backlight_access.py @@ -0,0 +1,139 @@ +"""Test configure_backlight_access — udev rule for brightness writes. + +Arch's brightnessctl ships no udev rules: it relies on logind, which grants +brightness writes only to the *active seat session*. Anything outside that +session — a script, a remote shell, a panel launched into a different +session — gets EPERM against root-owned sysfs. Found on velox 2026-08-13: +on a fresh install the desktop settings panel's screen and keyboard +brightness sliders were both inert. + +The step drops a rule making the brightness attributes group-writable by +video (the group create_user already adds the user to). leds are granted to +video as well rather than input, so nobody needs input-group membership — +and the keylogging surface that carries — just to dim a keyboard. + +Method: sed-extract configure_backlight_access from the real `archsetup`, +point it at a temp rules dir, and assert on the file it writes. + + python3 -m unittest tests.installer-steps.test_configure_backlight_access +""" + +import os +import re +import stat +import subprocess +import tempfile +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(ruledir, marker=None): + # The step sends udevadm's own output to $logfile, so the stub records + # its argv to a marker file instead of stdout — writing straight to the + # file is unaffected by the caller's redirect. + marker = marker or os.devnull + script = textwrap.dedent(f"""\ + logfile=/dev/null + action="" + display() {{ :; }} + error_warn() {{ echo "WARN: $1"; return 1; }} + udevadm() {{ echo "UDEVADM: $*" >> "{marker}"; }} + source <(sed -n '/^configure_backlight_access() {{/,/^}}/p' "{ARCHSETUP}") + configure_backlight_access "{ruledir}" + echo "RC=$?" + exit 0 + """) + return subprocess.run( + ["bash", "-c", script], capture_output=True, text=True, timeout=10, + ) + + +def rc_of(r): + m = re.search(r"^RC=(\d+)$", r.stdout, re.M) + assert m, "no RC line in output: %r / %r" % (r.stdout, r.stderr) + return int(m.group(1)) + + +class ConfigureBacklightAccess(unittest.TestCase): + # ------------------------------------------------------------ normal ---- + def test_writes_a_rule_covering_backlight_and_keyboard_leds(self): + with tempfile.TemporaryDirectory() as d: + r = run(d) + self.assertEqual(rc_of(r), 0) + rule = os.path.join(d, "90-backlight.rules") + self.assertTrue(os.path.exists(rule)) + body = open(rule).read() + self.assertIn('SUBSYSTEM=="backlight"', body) + self.assertIn('SUBSYSTEM=="leds"', body) + self.assertIn('KERNEL=="*kbd_backlight"', body) + + def test_grants_the_video_group_write_access(self): + with tempfile.TemporaryDirectory() as d: + run(d) + body = open(os.path.join(d, "90-backlight.rules")).read() + self.assertIn("chgrp video", body) + self.assertIn("chmod g+w", body) + + def test_does_not_use_the_input_group(self): + # Granting leds to input would require input-group membership, + # which also confers read access to every input device. + with tempfile.TemporaryDirectory() as d: + run(d) + body = open(os.path.join(d, "90-backlight.rules")).read() + self.assertNotIn("chgrp input", body) + + def test_rule_is_world_readable_not_writable(self): + with tempfile.TemporaryDirectory() as d: + run(d) + mode = stat.S_IMODE(os.stat(os.path.join(d, "90-backlight.rules")).st_mode) + self.assertEqual(mode, 0o644) + + def test_reloads_udev_so_the_rule_applies_without_a_reboot(self): + with tempfile.TemporaryDirectory() as d: + marker = os.path.join(d, "udevadm.calls") + run(d, marker=marker) + calls = open(marker).read() + self.assertIn("UDEVADM: control --reload", calls) + self.assertRegex(calls, r"UDEVADM: trigger .*backlight") + self.assertRegex(calls, r"UDEVADM: trigger .*leds") + + # ---------------------------------------------------------- boundary ---- + def test_running_twice_leaves_one_correct_rule(self): + # Compare whole bodies rather than counting lines: a count encodes + # today's line total, so it breaks on a correct edit and passes an + # append regression that happens to hit the same number. + with tempfile.TemporaryDirectory() as d: + run(d) + first = open(os.path.join(d, "90-backlight.rules")).read() + r = run(d) + self.assertEqual(rc_of(r), 0) + second = open(os.path.join(d, "90-backlight.rules")).read() + self.assertEqual(first, second) + + def test_absent_rules_directory_is_created(self): + with tempfile.TemporaryDirectory() as d: + nested = os.path.join(d, "etc", "udev", "rules.d") + r = run(nested) + self.assertEqual(rc_of(r), 0) + self.assertTrue(os.path.exists(os.path.join(nested, "90-backlight.rules"))) + + # ------------------------------------------------------------- error ---- + @unittest.skipUnless(os.geteuid() != 0, "root ignores directory write bits") + def test_unwritable_directory_warns_and_does_not_crash(self): + with tempfile.TemporaryDirectory() as d: + ruledir = os.path.join(d, "ro") + os.mkdir(ruledir) + os.chmod(ruledir, 0o500) + try: + r = run(ruledir) + self.assertIn("WARN:", r.stdout) + self.assertNotEqual(rc_of(r), 0) + finally: + os.chmod(ruledir, 0o700) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/installer-steps/test_configure_framework_amd_internal_mic.py b/tests/installer-steps/test_configure_framework_amd_internal_mic.py new file mode 100644 index 0000000..7a45fb1 --- /dev/null +++ b/tests/installer-steps/test_configure_framework_amd_internal_mic.py @@ -0,0 +1,277 @@ +"""Test configure_framework_amd_internal_mic and its DMI gate. + +The Framework Laptop 13 on AMD Ryzen AI 300 (Strix Point and Krackan Point +alike) has a known, still-open kernel bug in the ACP PDM digital-mic path: the +DMIC card (acp-pdm-mach, hw:acppdmmach) returns a constant full-scale rail +(rms 0.99999) instead of audio. With UCM on, alsa-ucm-conf links that card +into the ALC285 codec card, so the only "internal mic" PipeWire offers IS the +broken DMIC, and the built-in microphone is dead in every app. Found on velox +2026-10-06 after the 08-13 reinstall; Framework community threads 81866 and +75748 document it across several HX 370 owners. + +The workaround is a WirePlumber drop-in that turns UCM off (the ALC285 falls +back to its analog-stereo profile, whose Internal Microphone pin carries real +audio) and disables the standalone PDM card so nothing can pick it. The step +installs the shipped copy of that drop-in, and only on this hardware: the +gate reads DMI, not the hostname, because the fix belongs to the board, not +to velox. + +The gate has to be exact in both directions. The same bug does not exist on +the Intel Framework 13 (SOF stack, not ACP), and the Framework Desktop (ratio, +"AMD Ryzen AI Max 300") has no internal mic at all, so a vendor-only or +"AMD"-only match would drop a UCM-off file onto machines whose audio is fine +and silently flatten their codec profiles. + +Method: sed-extract both functions from the real `archsetup`, point the step +at a temp conf.d dir and a fixture DMI dir, and fake display / error_warn. +The shipped drop-in is read from the real configs/ tree so the test pins the +content the installer actually delivers. + +Run from repo root: + python3 -m unittest tests.installer-steps.test_configure_framework_amd_internal_mic +""" + +import os +import re +import stat +import subprocess +import tempfile +import textwrap +import unittest + +REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")) +ARCHSETUP = os.path.join(REPO_ROOT, "archsetup") +DROPIN_SRC = os.path.join(REPO_ROOT, "configs", "wireplumber", "50-disable-ucm.conf") +DROPIN_NAME = "50-disable-ucm.conf" +# The broken DMIC card as velox names it, and the codec card that must stay. +VELOX_PDM_CARD = "alsa_card.pci-0000_c1_00.5-platform-acp-pdm-mach" +VELOX_CODEC_CARD = "alsa_card.pci-0000_c1_00.6" + +# DMI fixtures. The product strings are what the real machines report. +VELOX = dict(sys_vendor="Framework", product_name="Laptop 13 (AMD Ryzen AI 300 Series)") +RATIO = dict(sys_vendor="Framework", product_name="Desktop (AMD Ryzen AI Max 300 Series)") +INTEL_13 = dict(sys_vendor="Framework", product_name="Laptop (13th Gen Intel Core)") +AMD_7040_13 = dict(sys_vendor="Framework", product_name="Laptop 13 (AMD Ryzen 7040Series)") +AMD_16 = dict(sys_vendor="Framework", product_name="Laptop 16 (AMD Ryzen 7040 Series)") +OTHER_VENDOR = dict(sys_vendor="LENOVO", product_name="Laptop 13 (AMD Ryzen AI 300 Series)") + + +def write_dmi(dmi_dir, fields): + os.makedirs(dmi_dir, exist_ok=True) + for name, value in fields.items(): + with open(os.path.join(dmi_dir, name), "w") as f: + f.write(value + "\n") + + +def gate(dmi_dir): + script = textwrap.dedent(f"""\ + source <(sed -n '/^is_framework_13_ryzen_ai_300() {{/,/^}}/p' "{ARCHSETUP}") + if is_framework_13_ryzen_ai_300 "{dmi_dir}"; then echo "GATE=yes"; else echo "GATE=no"; fi + exit 0 + """) + r = subprocess.run(["bash", "-c", script], capture_output=True, text=True, timeout=10) + m = re.search(r"^GATE=(yes|no)$", r.stdout, re.M) + assert m, "no GATE line in output: %r / %r" % (r.stdout, r.stderr) + return m.group(1) == "yes" + + +def run(confd, dmi_dir, src_root=REPO_ROOT): + # The step sends install's output to $logfile; the fakes report on stdout + # so the test can see the warning and the function's own return code. + script = textwrap.dedent(f"""\ + logfile=/dev/null + action="" + username=nobody + display() {{ :; }} + error_warn() {{ echo "WARN: $1"; return 1; }} + user_archsetup_dir="{src_root}" + source <(sed -n '/^is_framework_13_ryzen_ai_300() {{/,/^}}/p' "{ARCHSETUP}") + source <(sed -n '/^configure_framework_amd_internal_mic() {{/,/^}}/p' "{ARCHSETUP}") + configure_framework_amd_internal_mic "{confd}" "{dmi_dir}" + echo "RC=$?" + exit 0 + """) + return subprocess.run( + ["bash", "-c", script], capture_output=True, text=True, timeout=10, + ) + + +def rc_of(r): + m = re.search(r"^RC=(\d+)$", r.stdout, re.M) + assert m, "no RC line in output: %r / %r" % (r.stdout, r.stderr) + return int(m.group(1)) + + +class Framework13RyzenAi300Gate(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.dmi = os.path.join(self.tmp.name, "dmi") + + def tearDown(self): + self.tmp.cleanup() + + # ------------------------------------------------------------ normal ---- + def test_matches_the_framework_13_on_ryzen_ai_300(self): + write_dmi(self.dmi, VELOX) + self.assertTrue(gate(self.dmi)) + + # ---------------------------------------------------------- boundary ---- + def test_rejects_the_framework_desktop_on_ryzen_ai_max(self): + """ratio: same vendor, "AMD Ryzen AI" in the name, no internal mic. + + "AMD Ryzen AI Max 300" must not satisfy an "AMD Ryzen AI 300" match. + """ + write_dmi(self.dmi, RATIO) + self.assertFalse(gate(self.dmi)) + + def test_rejects_the_framework_13_on_intel(self): + """Intel's SOF stack has no ACP PDM path; the bug does not exist there.""" + write_dmi(self.dmi, INTEL_13) + self.assertFalse(gate(self.dmi)) + + def test_rejects_the_framework_13_on_ryzen_7040(self): + """Phoenix is not Ryzen AI 300; the workaround is unverified there.""" + write_dmi(self.dmi, AMD_7040_13) + self.assertFalse(gate(self.dmi)) + + def test_rejects_the_framework_16(self): + write_dmi(self.dmi, AMD_16) + self.assertFalse(gate(self.dmi)) + + def test_rejects_another_vendor_with_the_same_model_string(self): + write_dmi(self.dmi, OTHER_VENDOR) + self.assertFalse(gate(self.dmi)) + + def test_rejects_a_missing_dmi_directory(self): + self.assertFalse(gate(os.path.join(self.tmp.name, "nope"))) + + +class ConfigureFrameworkAmdInternalMic(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.dmi = os.path.join(self.tmp.name, "dmi") + self.confd = os.path.join(self.tmp.name, "wireplumber.conf.d") + self.dropin = os.path.join(self.confd, DROPIN_NAME) + + def tearDown(self): + self.tmp.cleanup() + + # ------------------------------------------------------------ normal ---- + def test_framework_13_ryzen_ai_300_gets_the_shipped_dropin(self): + write_dmi(self.dmi, VELOX) + r = run(self.confd, self.dmi) + self.assertEqual(rc_of(r), 0, r.stdout) + self.assertTrue(os.path.exists(self.dropin), "drop-in not installed") + with open(self.dropin) as f, open(DROPIN_SRC) as src: + self.assertEqual(f.read(), src.read(), + "the installed file must be the shipped file, byte for byte") + + def test_dropin_mode_is_0644(self): + """Owner-writable, world-readable, like every file under /etc.""" + write_dmi(self.dmi, VELOX) + run(self.confd, self.dmi) + mode = stat.S_IMODE(os.stat(self.dropin).st_mode) + self.assertEqual(mode, 0o644) + + def test_shipped_dropin_turns_ucm_off(self): + """Pin the shipped content, not just the copy: use-ucm must be an + active setting, not a commented example.""" + with open(DROPIN_SRC) as f: + body = f.read() + active = [ln for ln in body.splitlines() + if "alsa.use-ucm" in ln and not ln.lstrip().startswith("#")] + self.assertEqual(len(active), 1, body) + self.assertRegex(active[0], r"alsa\.use-ucm\s*=\s*false") + + def test_shipped_dropin_disables_the_standalone_pdm_card(self): + """Without UCM the broken DMIC still shows up as its own card; the + rule has to catch it and disable it, or an app or the default + selection can still land on the full-scale rail. + + The gate fires on every Framework 13 on Ryzen AI 300, so the rule + must match the card by name pattern (WirePlumber's ~regex form), not + by the one PCI path velox happens to enumerate: a renumbered bus on + the same board would otherwise get UCM turned off with the broken + card left selectable. WirePlumber's regex match is unanchored, so + re.search mirrors it. + """ + with open(DROPIN_SRC) as f: + body = f.read() + m = re.search(r'device\.name\s*=\s*"~([^"]+)"', body) + self.assertIsNotNone(m, "the rule must match device.name with a ~regex, " + "not a literal PCI path") + assert m is not None # narrows the type; the assertion above is the test + pattern = m.group(1) + self.assertTrue(re.search(pattern, VELOX_PDM_CARD), + f"{pattern!r} must match velox's card {VELOX_PDM_CARD!r}") + self.assertTrue(re.search(pattern, "alsa_card.pci-0000_c3_00.5-platform-acp-pdm-mach"), + "the same card on another bus number must match too") + self.assertFalse(re.search(pattern, VELOX_CODEC_CARD), + "the ALC285 codec card carries the working mic and must not be disabled") + self.assertRegex(body, r"device\.disabled\s*=\s*true") + + # ---------------------------------------------------------- boundary ---- + def test_other_hardware_writes_nothing(self): + for fixture in (RATIO, INTEL_13, AMD_7040_13, AMD_16, OTHER_VENDOR): + with self.subTest(product=fixture["product_name"], vendor=fixture["sys_vendor"]): + write_dmi(self.dmi, fixture) + r = run(self.confd, self.dmi) + self.assertEqual(rc_of(r), 0, r.stdout) + self.assertFalse(os.path.exists(self.confd), + "no conf.d may be created on hardware the step does not apply to") + self.assertNotIn("WARN:", r.stdout) + + def test_missing_dmi_directory_is_a_quiet_no_op(self): + """A VM or a stripped sysfs reports no DMI; that is not an error.""" + r = run(self.confd, os.path.join(self.tmp.name, "nope")) + self.assertEqual(rc_of(r), 0, r.stdout) + self.assertFalse(os.path.exists(self.confd)) + self.assertNotIn("WARN:", r.stdout) + + def test_running_twice_leaves_one_identical_dropin(self): + write_dmi(self.dmi, VELOX) + run(self.confd, self.dmi) + with open(self.dropin) as f: + first = f.read() + r = run(self.confd, self.dmi) + self.assertEqual(rc_of(r), 0, r.stdout) + with open(self.dropin) as f: + second = f.read() + self.assertEqual(first, second) + self.assertEqual(os.listdir(self.confd), [DROPIN_NAME]) + + def test_absent_confd_is_created(self): + """wireplumber's package ships /etc/wireplumber/ but not conf.d/; the + step must create the nested path rather than fail on it.""" + write_dmi(self.dmi, VELOX) + nested = os.path.join(self.tmp.name, "etc", "wireplumber", "wireplumber.conf.d") + r = run(nested, self.dmi) + self.assertEqual(rc_of(r), 0, r.stdout) + self.assertTrue(os.path.exists(os.path.join(nested, DROPIN_NAME))) + + # ------------------------------------------------------------- error ---- + def test_missing_shipped_file_warns_and_does_not_crash(self): + """A partial checkout has no configs/ tree; surface it, keep going.""" + write_dmi(self.dmi, VELOX) + empty_root = os.path.join(self.tmp.name, "empty-checkout") + os.makedirs(empty_root) + r = run(self.confd, self.dmi, src_root=empty_root) + self.assertIn("WARN:", r.stdout) + self.assertNotEqual(rc_of(r), 0) + self.assertFalse(os.path.exists(self.dropin)) + + @unittest.skipUnless(os.geteuid() != 0, "root ignores directory write bits") + def test_unwritable_confd_warns_and_does_not_crash(self): + write_dmi(self.dmi, VELOX) + os.makedirs(self.confd) + os.chmod(self.confd, 0o500) + try: + r = run(self.confd, self.dmi) + self.assertIn("WARN:", r.stdout) + self.assertNotEqual(rc_of(r), 0) + finally: + os.chmod(self.confd, 0o700) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/installer-steps/test_configure_hibernate_delay.py b/tests/installer-steps/test_configure_hibernate_delay.py new file mode 100644 index 0000000..95f2dcb --- /dev/null +++ b/tests/installer-steps/test_configure_hibernate_delay.py @@ -0,0 +1,131 @@ +"""Test configure_hibernate_delay — the suspend-to-hibernate handoff. + +The desktop panel can upgrade its idle suspend stage to +suspend-then-hibernate, but the delay between the two is systemd's +HibernateDelaySec, which lives in a root-owned drop-in and so cannot be a +panel setting. Left unset the delay depends on the hardware: a machine +with a battery hibernates on a low-battery alarm at an hour nobody chose, +and one without falls back to systemd's 2h default. + +90 minutes is the seeded value: long enough that stepping out of a meeting +costs a screen unlock rather than a passphrase and a resume, short enough +that a laptop in a bag is not holding its encryption keys in RAM all +afternoon (a suspended machine is; a hibernated one is not). + +Method: sed-extract configure_hibernate_delay from the real `archsetup`, +point it at a temp config dir, and assert on the file it writes. + + python3 -m unittest tests.installer-steps.test_configure_hibernate_delay +""" + +import os +import re +import stat +import subprocess +import tempfile +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(confdir): + script = textwrap.dedent(f"""\ + logfile=/dev/null + action="" + display() {{ :; }} + error_warn() {{ echo "WARN: $1"; return 1; }} + source <(sed -n '/^configure_hibernate_delay() {{/,/^}}/p' "{ARCHSETUP}") + configure_hibernate_delay "{confdir}" + echo "RC=$?" + exit 0 + """) + return subprocess.run( + ["bash", "-c", script], capture_output=True, text=True, timeout=10, + ) + + +def rc_of(r): + m = re.search(r"^RC=(\d+)$", r.stdout, re.M) + assert m, "no RC line in output: %r / %r" % (r.stdout, r.stderr) + return int(m.group(1)) + + +def body(confdir): + with open(os.path.join(confdir, "60-hibernate-delay.conf")) as f: + return f.read() + + +class ConfigureHibernateDelay(unittest.TestCase): + # ------------------------------------------------------------ normal ---- + def test_writes_a_sleep_drop_in_with_the_delay(self): + with tempfile.TemporaryDirectory() as d: + r = run(d) + self.assertEqual(rc_of(r), 0) + text = body(d) + self.assertIn("[Sleep]", text) + self.assertIn("HibernateDelaySec=90min", text) + + def test_drop_in_is_world_readable_not_writable(self): + with tempfile.TemporaryDirectory() as d: + run(d) + mode = stat.S_IMODE( + os.stat(os.path.join(d, "60-hibernate-delay.conf")).st_mode) + self.assertEqual(mode, 0o644) + + def test_explains_why_the_value_is_not_left_to_systemd(self): + # The number is a judgement call, so the file has to carry its + # reasoning: whoever changes it next needs the tradeoff, not just + # a number to overwrite. Assert on the tradeoff's own terms — + # "contains a hash" would pass on any comment at all. + with tempfile.TemporaryDirectory() as d: + run(d) + text = body(d).lower() + self.assertIn("2h default", text) # what unset would give + self.assertIn("encryption keys", text) # why not longer + self.assertIn("passphrase", text) # why not shorter + + def test_drop_in_sorts_above_vendor_drop_ins(self): + # systemd reserves 10-40 for /usr and 60-90 for /etc. A 10- file + # in /etc sorts below a vendor 20- file and silently loses to it. + with tempfile.TemporaryDirectory() as d: + run(d) + name = os.listdir(d)[0] + prefix = int(name.split("-")[0]) + self.assertGreaterEqual(prefix, 60) + self.assertLessEqual(prefix, 90) + + # ---------------------------------------------------------- boundary ---- + def test_running_twice_leaves_one_correct_drop_in(self): + with tempfile.TemporaryDirectory() as d: + run(d) + first = body(d) + r = run(d) + self.assertEqual(rc_of(r), 0) + self.assertEqual(first, body(d)) + + def test_absent_directory_is_created(self): + with tempfile.TemporaryDirectory() as d: + nested = os.path.join(d, "etc", "systemd", "sleep.conf.d") + r = run(nested) + self.assertEqual(rc_of(r), 0) + self.assertIn("HibernateDelaySec=90min", body(nested)) + + # ------------------------------------------------------------- error ---- + @unittest.skipUnless(os.geteuid() != 0, "root ignores directory write bits") + def test_unwritable_directory_warns_and_does_not_crash(self): + with tempfile.TemporaryDirectory() as d: + confdir = os.path.join(d, "ro") + os.mkdir(confdir) + os.chmod(confdir, 0o500) + try: + r = run(confdir) + self.assertIn("WARN:", r.stdout) + self.assertNotEqual(rc_of(r), 0) + finally: + os.chmod(confdir, 0o700) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/installer-steps/test_configure_openai_chatgpt_repo.py b/tests/installer-steps/test_configure_openai_chatgpt_repo.py new file mode 100644 index 0000000..cb0e84b --- /dev/null +++ b/tests/installer-steps/test_configure_openai_chatgpt_repo.py @@ -0,0 +1,324 @@ +"""Test configure_openai_chatgpt_repo — OpenAI's signed pacman repo for ChatGPT. + +OpenAI ships the ChatGPT (Codex) desktop app for Arch as chatgpt-bin from +its own signed pacman repository. Its install-arch.sh pins the repository +signing-key fingerprint, locally signs the key into the pacman keyring, +writes /etc/pacman.d/openai-chatgpt.conf and adds an Include line to +pacman.conf; ratio and velox were set up that way on 2026-10-11. The step +reproduces exactly that configuration so a re-run on those machines is a +no-op and the app rides normal pacman upgrades under OpenAI's signature. +The AUR repackage of the same binary (chatgpt-desktop) would conflict with +chatgpt-bin on any machine already using the repo, so it is not used. + +A half-configured repository is worse than none: with the Include in place +and the key untrusted, SigLevel TrustedOnly makes every later pacman call +fail on the invalid database. So an untrusted key (fingerprint mismatch or +failed download) must import nothing and leave pacman.conf untouched. + +The pin is on the primary key's fingerprint alone. pacman-key --lsign-key +resolves a subkey fingerprint to its primary, so a bundle whose subkey or +second primary carries the pinned fingerprint must be rejected too. + +Method: sed-extract configure_openai_chatgpt_repo from the real `archsetup`, +point it at a temp pacman.conf and repo drop-in, and replace curl, gpg, +pacman-key and pacman with recorders driven by environment switches. + + python3 -m unittest tests.installer-steps.test_configure_openai_chatgpt_repo +""" + +import os +import re +import stat +import subprocess +import tempfile +import textwrap +import unittest + +REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")) +ARCHSETUP = os.path.join(REPO_ROOT, "archsetup") + +FPR = "3BFA0E4AE8B8CC16A2D9BA684A3B4A566C4660E4" +KEY_URL = "https://persistent.oaistatic.com/codex-app-prod/linux/repository-signing-key.gpg" +SERVER = "https://persistent.oaistatic.com/codex-app-prod/linux/arch/$arch" +BASE_CONF = "[core]\nInclude = /etc/pacman.d/mirrorlist\n" + +# Recorders. Every fake appends its argv to $MARKER; the pacman fake also +# records whether the Include line is already in pacman.conf at call time, +# which is how the tests see that -Sy ran after the repo was wired in. +STUBS = r""" +logfile=/dev/null +action="" +display() { :; } +backup_system_file() { :; } +error_warn() { echo "WARN: $1"; return 1; } +pacman-key() { + echo "PACMAN_KEY: $*" >> "$MARKER" + case "$1" in + --list-keys) [ "$KEY_PRESENT" = 1 ] ;; + *) return 0 ;; + esac +} +curl() { + echo "CURL: $*" >> "$MARKER" + [ "$CURL_FAIL" = 1 ] && return 22 + local out="" + while [ $# -gt 0 ]; do + case "$1" in -o) out="$2"; shift ;; esac + shift + done + [ -n "$out" ] && printf 'FAKE KEY\n' > "$out" + return 0 +} +gpg() { + echo "GPG: $*" >> "$MARKER" + case " $* " in + *" --show-keys "*) + printf 'pub:u:4096:1:4A3B4A566C4660E4:1754362069:::u:::scSC::::::23::0:\n' + printf 'fpr:::::::::%s:\n' "$SERVED_FPR" + printf 'uid:u::::1754362069::ABC::Codex Linux Repository::::::::::0:\n' + if [ -n "$SERVED_SUBKEY_FPR" ]; then + printf 'sub:u:4096:1:0123456789ABCDEF:1754362069::::::e::::::23:\n' + printf 'fpr:::::::::%s:\n' "$SERVED_SUBKEY_FPR" + fi + if [ -n "$SERVED_SECOND_PUB_FPR" ]; then + printf 'pub:u:4096:1:FEDCBA9876543210:1754362069:::u:::scSC::::::23::0:\n' + printf 'fpr:::::::::%s:\n' "$SERVED_SECOND_PUB_FPR" + printf 'uid:u::::1754362069::DEF::Somebody Else::::::::::0:\n' + fi + ;; + esac + return 0 +} +pacman() { + local n + n=$(grep -cxF "Include = $REPO_CONF" "$PACMAN_CONF" 2>/dev/null) + echo "PACMAN: $* include=$n" >> "$MARKER" + return 0 +} +""" + + +def paths(d): + return (os.path.join(d, "pacman.conf"), + os.path.join(d, "pacman.d", "openai-chatgpt.conf"), + os.path.join(d, "calls")) + + +def run(d, *, key_present=False, served_fpr=FPR, served_subkey_fpr="", + served_second_pub_fpr="", curl_fail=False, pacman_conf_body=BASE_CONF): + pacman_conf, repo_conf, marker = paths(d) + if pacman_conf_body is not None and not os.path.exists(pacman_conf): + with open(pacman_conf, "w") as f: + f.write(pacman_conf_body) + script = STUBS + textwrap.dedent(f"""\ + source <(sed -n '/^configure_openai_chatgpt_repo() {{/,/^}}/p' "{ARCHSETUP}") + configure_openai_chatgpt_repo "$PACMAN_CONF" "$REPO_CONF" + echo "RC=$?" + exit 0 + """) + env = dict(os.environ, MARKER=marker, PACMAN_CONF=pacman_conf, + REPO_CONF=repo_conf, KEY_PRESENT="1" if key_present else "0", + SERVED_FPR=served_fpr, SERVED_SUBKEY_FPR=served_subkey_fpr, + SERVED_SECOND_PUB_FPR=served_second_pub_fpr, + CURL_FAIL="1" if curl_fail else "0") + return subprocess.run(["bash", "-c", script], capture_output=True, + text=True, timeout=10, env=env) + + +def rc_of(r): + m = re.search(r"^RC=(\d+)$", r.stdout, re.M) + assert m, "no RC line in output: %r / %r" % (r.stdout, r.stderr) + return int(m.group(1)) + + +def read(path): + with open(path) as f: + return f.read() + + +def calls(d): + marker = paths(d)[2] + return read(marker) if os.path.exists(marker) else "" + + +class ConfigureOpenaiChatgptRepo(unittest.TestCase): + # ------------------------------------------------------------ normal ---- + def test_writes_the_published_repo_definition(self): + with tempfile.TemporaryDirectory() as d: + r = run(d) + self.assertEqual(rc_of(r), 0) + _, repo_conf, _ = paths(d) + self.assertTrue(os.path.exists(repo_conf)) + body = read(repo_conf) + self.assertIn("[openai-chatgpt]\n", body) + self.assertIn("SigLevel = Required DatabaseRequired TrustedOnly\n", body) + # $arch stays literal: pacman expands it, the shell must not. + self.assertIn(f"Server = {SERVER}\n", body) + + def test_repo_definition_is_world_readable_not_writable(self): + with tempfile.TemporaryDirectory() as d: + run(d) + mode = stat.S_IMODE(os.stat(paths(d)[1]).st_mode) + self.assertEqual(mode, 0o644) + + def test_adds_one_include_line_to_pacman_conf(self): + with tempfile.TemporaryDirectory() as d: + run(d) + pacman_conf, repo_conf, _ = paths(d) + body = read(pacman_conf) + self.assertTrue(body.startswith(BASE_CONF), body) + self.assertEqual(body.count(f"\nInclude = {repo_conf}\n"), 1) + + def test_imports_and_locally_signs_the_pinned_key(self): + with tempfile.TemporaryDirectory() as d: + run(d) + c = calls(d) + self.assertIn(f" {KEY_URL}", c) + self.assertIn("GPG: ", c) + self.assertIn("--show-keys", c) + self.assertIn("PACMAN_KEY: --add ", c) + self.assertIn(f"PACMAN_KEY: --lsign-key {FPR}", c) + + def test_refreshes_databases_after_the_include_is_in_place(self): + with tempfile.TemporaryDirectory() as d: + run(d) + self.assertIn("PACMAN: -Sy include=1", calls(d)) + + # ---------------------------------------------------------- boundary ---- + def test_running_twice_leaves_one_include_and_the_same_files(self): + with tempfile.TemporaryDirectory() as d: + pacman_conf, repo_conf, _ = paths(d) + self.assertEqual(rc_of(run(d)), 0) + first_conf, first_repo = read(pacman_conf), read(repo_conf) + self.assertEqual(rc_of(run(d)), 0) + self.assertEqual(read(pacman_conf), first_conf) + self.assertEqual(read(repo_conf), first_repo) + self.assertEqual(first_conf.count("Include = " + repo_conf), 1) + + def test_already_trusted_key_is_not_downloaded_or_added_again(self): + with tempfile.TemporaryDirectory() as d: + r = run(d, key_present=True) + self.assertEqual(rc_of(r), 0) + c = calls(d) + self.assertNotIn("CURL:", c) + self.assertNotIn("PACMAN_KEY: --add", c) + self.assertTrue(os.path.exists(paths(d)[1])) + + def test_pacman_conf_without_trailing_newline_gets_its_own_line(self): + with tempfile.TemporaryDirectory() as d: + r = run(d, pacman_conf_body=BASE_CONF.rstrip("\n")) + self.assertEqual(rc_of(r), 0) + pacman_conf, repo_conf, _ = paths(d) + body = read(pacman_conf) + self.assertIn("Include = /etc/pacman.d/mirrorlist\n", body) + self.assertRegex(body, re.compile(rf"^Include = {re.escape(repo_conf)}$", re.M)) + + def test_existing_differing_repo_definition_is_kept_and_flagged(self): + with tempfile.TemporaryDirectory() as d: + pacman_conf, repo_conf, _ = paths(d) + os.makedirs(os.path.dirname(repo_conf)) + custom = "[openai-chatgpt]\nSigLevel = Required\nServer = file:///srv/mirror\n" + with open(repo_conf, "w") as f: + f.write(custom) + r = run(d) + self.assertIn("WARN:", r.stdout) + self.assertEqual(read(repo_conf), custom) + # The repository is still configured, just by the user's file. + self.assertIn(f"\nInclude = {repo_conf}\n", read(pacman_conf)) + self.assertEqual(rc_of(r), 0) + + # ------------------------------------------------------------- error ---- + def assert_nothing_configured(self, d, r): + pacman_conf, repo_conf, _ = paths(d) + c = calls(d) + self.assertIn("WARN:", r.stdout) + self.assertNotIn("PACMAN_KEY: --add", c) + self.assertNotIn("PACMAN_KEY: --lsign-key", c) + self.assertNotIn("PACMAN: -Sy", c) + self.assertFalse(os.path.exists(repo_conf)) + self.assertEqual(read(pacman_conf), BASE_CONF) + self.assertNotEqual(rc_of(r), 0) + + def test_fingerprint_mismatch_imports_nothing_and_touches_nothing(self): + with tempfile.TemporaryDirectory() as d: + r = run(d, served_fpr="DEADBEEF" * 5) + self.assert_nothing_configured(d, r) + + def test_key_download_failure_imports_nothing_and_touches_nothing(self): + with tempfile.TemporaryDirectory() as d: + r = run(d, curl_fail=True) + self.assert_nothing_configured(d, r) + + # The pin is on the PRIMARY key's fingerprint and nothing else. pacman-key + # --lsign-key resolves a subkey fingerprint to the primary that carries it, + # so a pin that a subkey or a second primary could satisfy would trust + # whatever key the bundle smuggled in beside the expected one. + def test_subkey_carrying_the_pinned_fingerprint_is_rejected(self): + with tempfile.TemporaryDirectory() as d: + r = run(d, served_fpr="DEADBEEF" * 5, served_subkey_fpr=FPR) + self.assert_nothing_configured(d, r) + + def test_second_primary_carrying_the_pinned_fingerprint_is_rejected(self): + with tempfile.TemporaryDirectory() as d: + r = run(d, served_fpr="DEADBEEF" * 5, served_second_pub_fpr=FPR) + self.assert_nothing_configured(d, r) + + def test_rogue_second_primary_beside_the_pinned_key_is_rejected(self): + with tempfile.TemporaryDirectory() as d: + r = run(d, served_fpr=FPR, served_second_pub_fpr="DEADBEEF" * 5) + self.assert_nothing_configured(d, r) + + @unittest.skipUnless(os.geteuid() != 0, "root ignores file write bits") + def test_unwritable_pacman_conf_warns_and_reports_failure(self): + with tempfile.TemporaryDirectory() as d: + pacman_conf, _, _ = paths(d) + with open(pacman_conf, "w") as f: + f.write(BASE_CONF) + os.chmod(pacman_conf, 0o444) + try: + r = run(d) + self.assertIn("WARN:", r.stdout) + self.assertNotEqual(rc_of(r), 0) + self.assertNotIn("PACMAN: -Sy", calls(d)) + finally: + os.chmod(pacman_conf, 0o644) + + +class SupplementalSoftwareWiring(unittest.TestCase): + """chatgpt-bin is installed only once its repository is trusted.""" + + def run_step(self, repo_rc): + script = textwrap.dedent(f"""\ + desktop_env=hyprland + display() {{ :; }} + run_task() {{ :; }} + mask_fwupd_passim() {{ :; }} + pacman_install() {{ echo "PACMAN_INSTALL: $1"; }} + aur_install() {{ echo "AUR_INSTALL: $1"; }} + configure_openai_chatgpt_repo() {{ echo "CONFIGURE_REPO"; return {repo_rc}; }} + source <(sed -n '/^supplemental_software() {{/,/^}}/p' "{ARCHSETUP}") + supplemental_software + """) + return subprocess.run(["bash", "-c", script], capture_output=True, + text=True, timeout=10) + + def test_installs_chatgpt_bin_after_configuring_the_repo(self): + out = self.run_step(0).stdout + self.assertIn("CONFIGURE_REPO\n", out) + self.assertLess(out.index("CONFIGURE_REPO"), + out.index("PACMAN_INSTALL: chatgpt-bin")) + + def test_skips_chatgpt_bin_when_the_repo_cannot_be_trusted(self): + out = self.run_step(1).stdout + self.assertNotIn("PACMAN_INSTALL: chatgpt-bin", out) + # The failure is local to the app; the rest of the step still runs. + self.assertIn("AUR_INSTALL: claude-desktop", out) + self.assertIn("AUR_INSTALL: freetube", out) + + def test_does_not_also_pull_the_conflicting_aur_repackage(self): + out = self.run_step(0).stdout + self.assertNotIn("chatgpt-desktop", out) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/installer-steps/test_configure_service_discovery.py b/tests/installer-steps/test_configure_service_discovery.py new file mode 100644 index 0000000..90ce041 --- /dev/null +++ b/tests/installer-steps/test_configure_service_discovery.py @@ -0,0 +1,85 @@ +"""Pin configure_service_discovery's source: the WS-Discovery host daemon +is never enabled. + +wsdd.service advertises this machine as a Samba host to Windows clients. +Nothing the installer sets up runs Samba, so enabled it advertised a share +server that doesn't exist while listening on every interface, VPN and +tailscale links included (2026-09-12). Browsing Windows shares is the other +direction: gvfs-wsdd spawns its own wsdd in discovery mode and needs only +the package. + +Method: the step writes straight to /etc (geoclue.conf, the dbus-broker +drop-in) with no directory parameter, so unlike the other step tests this +one can't run the function in a temp dir. It sed-extracts the source and +asserts on the calls it contains. + + python3 -m unittest tests.installer-steps.test_configure_service_discovery +""" + +import os +import re +import unittest + +REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")) +ARCHSETUP = os.path.join(REPO_ROOT, "archsetup") + + +def function_source(name): + with open(ARCHSETUP) as f: + text = f.read() + m = re.search(r"^%s\(\) \{\n.*?^\}\n" % re.escape(name), text, re.S | re.M) + assert m, "function %s not found in archsetup" % name + return m.group(0) + + +ENABLE_WSDD = r"(systemctl\s+enable|enable_service)\s+(--now\s+)?wsdd\b" + + +def calls(src): + """Non-comment lines — what the function actually runs.""" + return [l for l in src.splitlines() if l.strip() and not l.strip().startswith("#")] + + +class ConfigureServiceDiscovery(unittest.TestCase): + # ------------------------------------------------------------ normal ---- + def test_does_not_enable_the_wsdd_host_daemon(self): + # Both spellings the script uses: a raw systemctl call and the + # enable_service helper (which takes the bare unit name). + body = "\n".join(calls(function_source("configure_service_discovery"))) + self.assertNotRegex(body, ENABLE_WSDD) + + def test_turns_off_a_previously_enabled_wsdd_on_rerun(self): + # Machines installed before this change still have the unit + # enabled; a re-run converges them. The guard keeps a fresh install + # (no unit yet) quiet. + body = "\n".join(calls(function_source("configure_service_discovery"))) + self.assertRegex(body, r"systemctl\s+is-enabled\s+(--quiet\s+)?wsdd\.service") + self.assertRegex(body, r"systemctl\s+disable\s+--now\s+wsdd\.service") + + def test_still_enables_the_discovery_it_does_want(self): + # Characterization: removing wsdd must not have taken avahi (mDNS) + # or geoclue with it. + body = "\n".join(calls(function_source("configure_service_discovery"))) + self.assertIn("systemctl enable avahi-daemon.service", body) + self.assertIn("systemctl enable geoclue.service", body) + + # ---------------------------------------------------------- boundary ---- + def test_wsdd_package_still_installed_for_gvfs(self): + # The package stays: gvfs-wsdd depends on it and spawns its own + # discovery-mode instance. Only the host service is gone. + body = "\n".join(calls(function_source("supplemental_software"))) + self.assertRegex(body, r"pacman_install\s+wsdd\b") + self.assertRegex(body, r"pacman_install\s+gvfs-wsdd\b") + + # ------------------------------------------------------------- error ---- + def test_no_other_step_enables_wsdd_either(self): + # A regression that re-enables it from a different step is the same + # bug; scan the whole script, not just the one function. + with open(ARCHSETUP) as f: + lines = [l for l in f if l.strip() and not l.strip().startswith("#")] + offenders = [l.rstrip() for l in lines if re.search(ENABLE_WSDD, l)] + self.assertEqual(offenders, []) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/installer-steps/test_configure_tlp_power.py b/tests/installer-steps/test_configure_tlp_power.py new file mode 100644 index 0000000..dc46c93 --- /dev/null +++ b/tests/installer-steps/test_configure_tlp_power.py @@ -0,0 +1,239 @@ +"""Test configure_tlp_power's radio-enable line, daemon masking, and laptop gating. + +systemd-rfkill is masked on laptops because it fights TLP's radio handling — +which means nothing restores radio state at boot unless TLP is told to. The +velox 2026-04-10 setup found wifi and bluetooth soft-blocked on first boot for +exactly this reason. The conf written here must carry +DEVICES_TO_ENABLE_ON_STARTUP so a fresh install comes up with radios on. + +power-profiles-daemon is masked and stopped on laptops for the same class of +reason. power-profiles-daemon.service declares "Conflicts=tuned.service +tlp.service auto-cpufreq.service ..." — the line is in ppd's unit, not tlp's — +so systemd TERMs TLP the instant ppd starts. Leaving ppd merely disabled does +not prevent that: ppd ships D-Bus activation files, and the desktop-settings +panel's own powerprofilesctl call activates it on demand. Velox ran that way +from its 2026-08-13 rebuild until 2026-08-16, with TLP failing at every boot and +none of its battery policy applied, while the machine looked correctly +configured. Masking blocks D-Bus activation too, which both keeps TLP alive and +makes the panel's power control read as unavailable, the behavior the +package-install site in `archsetup` already documents as intended. The stop is +what makes a repair re-run take effect on a booted machine, where a mask alone +would leave a running ppd running. + +The bluetooth resume hook installed here must land in +/usr/lib/systemd/system-sleep/. That is the only directory systemd-sleep scans +(systemd 262's binary and its man page both name just the one), so a hook under +/etc/systemd/system-sleep/ is silently dead. Velox ran about fifteen hibernate +cycles between 2026-08-21 and 2026-10-05 with the hook installed there and it +never fired once; bluetooth came back rfkilled every time, and TLP's own hook +(which lives in /usr/lib and does run) carried the block across every later +cycle. + +Method: sed-extract configure_tlp_power from the real `archsetup`, point it at +a temp tlp.d dir and a temp power-supply dir, and fake pacman_install / +run_task / display / error_warn / systemctl. + +Run from repo root: + python3 -m unittest tests.installer-steps.test_configure_tlp_power +""" + +import os +import stat +import subprocess +import tempfile +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(battery=True, bat_name="BAT0", unwritable_tlpd=False, systemctl_fails=False, + install_fails=False): + with tempfile.TemporaryDirectory() as d: + psdir = os.path.join(d, "power_supply") + os.makedirs(psdir) + if battery: + open(os.path.join(psdir, bat_name), "w").close() + tlpd = os.path.join(d, "tlp.d") + if unwritable_tlpd: + os.makedirs(tlpd) + os.chmod(tlpd, stat.S_IRUSR | stat.S_IXUSR) + # The real mask call redirects stdout into $logfile, so the fake + # systemctl records to a side file the test reads back instead. + sysrc = 1 if systemctl_fails else 0 + # The real install_executable warns on its own; the fake only reports + # the return code so the test can see whether the step keeps going. + instrc = 1 if install_fails else 0 + script = textwrap.dedent(f"""\ + logfile=/dev/null + action="" + display() {{ :; }} + pacman_install() {{ echo "INSTALL: $1"; }} + run_task() {{ echo "TASK: $1"; }} + systemctl() {{ echo "SYSTEMCTL: $*" >> "{d}/systemctl.log"; return {sysrc}; }} + error_warn() {{ echo "WARN: $1"; return 1; }} + user_archsetup_dir=/src + install_executable() {{ echo "INSTALL_EXEC: $1 -> $2"; return {instrc}; }} + source <(sed -n '/^configure_tlp_power() {{/,/^}}/p' "{ARCHSETUP}") + configure_tlp_power "{tlpd}" "{psdir}" + echo "RC=$?" + echo "CONF:[$(cat "{tlpd}/01-custom.conf" 2>/dev/null)]" + [ -f "{d}/systemctl.log" ] && cat "{d}/systemctl.log" + exit 0 + """) + r = subprocess.run( + ["bash", "-c", script], capture_output=True, text=True, timeout=10, + ) + if unwritable_tlpd: + os.chmod(tlpd, stat.S_IRWXU) + return r + + +class ConfigureTlpPower(unittest.TestCase): + # ------------------------------------------------------------ normal ---- + def test_laptop_gets_tlp_with_radio_enable_line(self): + r = run(battery=True) + self.assertIn("INSTALL: tlp", r.stdout) + conf = r.stdout.split("CONF:[")[1] + self.assertIn('DEVICES_TO_ENABLE_ON_STARTUP="bluetooth wifi"', conf, + "radios must be re-enabled at boot: systemd-rfkill is " + "masked, so TLP is the only thing that can do it") + self.assertIn("CPU_ENERGY_PERF_POLICY_ON_AC", conf) + self.assertIn("SYSTEMCTL: mask systemd-rfkill.service systemd-rfkill.socket", + r.stdout) + self.assertIn("TASK: enabling TLP service", r.stdout) + + def test_laptop_masks_power_profiles_daemon(self): + r = run(battery=True) + self.assertIn("SYSTEMCTL: mask power-profiles-daemon.service", r.stdout, + "ppd's unit declares Conflicts=...tlp.service..., so ppd " + "must be masked or it TERMs TLP whenever it is activated") + + def test_laptop_stops_running_power_profiles_daemon(self): + """Masking alone leaves an already-running ppd running. + + The installer runs on a booted system, so a repair re-run would + otherwise mask ppd, leave it live, and let it keep TLP dead until the + next reboot with nothing reporting it. + """ + r = run(battery=True) + self.assertIn("SYSTEMCTL: stop power-profiles-daemon.service", r.stdout) + + def test_ppd_is_masked_before_it_is_stopped(self): + """Order matters: stopping first leaves a window to re-activate in.""" + calls = [line for line in run(battery=True).stdout.splitlines() + if line.startswith("SYSTEMCTL:") and "power-profiles-daemon" in line] + verbs = [line.split()[1] for line in calls] + self.assertEqual(verbs, ["mask", "stop"]) + + def test_power_profiles_daemon_is_masked_not_merely_disabled(self): + """Disabling ppd is not enough — D-Bus activation ignores it. + + This is the whole point of the mask, so assert the verb directly. A + `disable` here would pass a naive "ppd is handled" check while leaving + the panel's powerprofilesctl call free to start ppd and kill TLP. + """ + r = run(battery=True) + ppd_calls = [line for line in r.stdout.splitlines() + if line.startswith("SYSTEMCTL:") and "power-profiles-daemon" in line] + self.assertTrue(ppd_calls, "configure_tlp_power must act on ppd at all") + for line in ppd_calls: + self.assertNotIn(" disable ", line, + "disable leaves D-Bus activation live; only mask blocks it") + + def test_radio_line_is_active_not_commented(self): + r = run(battery=True) + conf = r.stdout.split("CONF:[")[1].split("]")[0] + for line in conf.splitlines(): + if "DEVICES_TO_ENABLE_ON_STARTUP" in line: + self.assertFalse(line.lstrip().startswith("#"), + "the radio-enable line must not be commented out") + break + else: + self.fail("DEVICES_TO_ENABLE_ON_STARTUP line missing from conf") + + def test_laptop_installs_the_resume_hook_where_systemd_sleep_scans(self): + """The hook goes to /usr/lib/systemd/system-sleep/, the only scanned dir. + + Assert the full install list, not a substring: a second copy under + /etc/ would pass an assertIn and still be the dead hook this test + exists to catch. + """ + r = run(battery=True) + installs = [line for line in r.stdout.splitlines() + if line.startswith("INSTALL_EXEC:")] + self.assertEqual( + installs, + ["INSTALL_EXEC: /src/scripts/zz-bluetooth-resume -> " + "/usr/lib/systemd/system-sleep/zz-bluetooth-resume"], + "systemd-sleep never scans /etc/systemd/system-sleep/; a hook " + "there installs cleanly and never runs") + + # ---------------------------------------------------------- boundary ---- + def test_desktop_installs_no_resume_hook(self): + """No TLP on a desktop means no masked rfkill and no gap to fill.""" + r = run(battery=False) + self.assertNotIn("INSTALL_EXEC:", r.stdout) + + def test_desktop_without_battery_is_a_no_op(self): + r = run(battery=False) + self.assertNotIn("INSTALL:", r.stdout) + self.assertNotIn("SYSTEMCTL:", r.stdout) + self.assertIn("CONF:[]", r.stdout, "no conf may be written on a desktop") + + def test_second_battery_index_still_counts_as_laptop(self): + r = run(battery=True, bat_name="BAT1") + self.assertIn("INSTALL: tlp", r.stdout) + self.assertIn('DEVICES_TO_ENABLE_ON_STARTUP', r.stdout) + + def test_desktop_keeps_power_profiles_daemon(self): + """A batteryless machine must NOT get ppd masked. + + There is no TLP on a desktop to conflict with it, and the package-install + site enables ppd precisely so the settings panel's three-way power + control works there. Masking it here would break that control for no gain. + """ + r = run(battery=False) + self.assertNotIn("power-profiles-daemon", r.stdout) + + # ------------------------------------------------------------- error ---- + def test_failed_ppd_mask_warns_and_does_not_crash(self): + """A masking failure must surface, not pass silently. + + Silence is the exact failure mode being fixed: velox looked configured + while TLP was dead. If the mask cannot be applied, say so. + + Assert on the harness's own RC= line, not on r.returncode. The harness + script ends in a literal `exit 0`, so r.returncode is 0 no matter what + configure_tlp_power does — asserting it can never fail, which would make + this test the same silent no-op it exists to catch. + """ + r = run(battery=True, systemctl_fails=True) + self.assertIn("WARN: masking power-profiles-daemon for TLP", r.stdout) + self.assertIn("WARN: stopping power-profiles-daemon for TLP", r.stdout) + self.assertIn("RC=", r.stdout, + "the function must return so the install continues, " + "not exit and take the script down with it") + + def test_failed_hook_install_does_not_stop_the_ppd_mask(self): + """install_executable warns for itself, so the step must carry on. + + The mask that follows is what keeps TLP alive; a hook copy that fails + (source missing on a partial checkout) must not take it with it. + """ + r = run(battery=True, install_fails=True) + self.assertIn("INSTALL_EXEC:", r.stdout) + self.assertIn("SYSTEMCTL: mask power-profiles-daemon.service", r.stdout) + self.assertIn("RC=", r.stdout) + + @unittest.skipUnless(os.geteuid() != 0, "root ignores directory write bits") + def test_unwritable_tlpd_warns_and_does_not_crash(self): + r = run(battery=True, unwritable_tlpd=True) + self.assertIn("WARN:", r.stdout, + "a failed conf write must surface through error_warn") + self.assertEqual(r.returncode, 0) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/installer-steps/test_configure_tunnel_dns_over_tls.py b/tests/installer-steps/test_configure_tunnel_dns_over_tls.py new file mode 100644 index 0000000..89677d8 --- /dev/null +++ b/tests/installer-steps/test_configure_tunnel_dns_over_tls.py @@ -0,0 +1,147 @@ +"""Test configure_tunnel_dns_over_tls — per-link DoT off for VPN tunnels. + +The resolved drop-in pins DNSOverTLS=yes globally. Proton VPN (proton0) and +the static Proton WireGuard profiles (wgpvpn) push an in-tunnel resolver, +10.2.0.1, that answers plain port 53 and never completes TLS on 853, so +every lookup through the tunnel hangs. The Proton client recreates its NM +profile on each connect, so a per-profile setting can't stick; a +NetworkManager [connection-*] default matched on the interface names is +what NM pushes to resolved on every activation. Diagnosed 2026-09-09/10, +verified live on ratio and velox. + +Method: sed-extract configure_tunnel_dns_over_tls from the real +`archsetup`, point it at a temp conf.d, and assert on the file it writes. + + python3 -m unittest tests.installer-steps.test_configure_tunnel_dns_over_tls +""" + +import os +import re +import stat +import subprocess +import tempfile +import textwrap +import unittest + +REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")) +ARCHSETUP = os.path.join(REPO_ROOT, "archsetup") +FILENAME = "tunnel-dns-over-tls.conf" + + +def run(confdir): + script = textwrap.dedent(f"""\ + logfile=/dev/null + action="" + display() {{ :; }} + error_warn() {{ echo "WARN: $1"; return 1; }} + source <(sed -n '/^configure_tunnel_dns_over_tls() {{/,/^}}/p' "{ARCHSETUP}") + configure_tunnel_dns_over_tls "{confdir}" + echo "RC=$?" + exit 0 + """) + return subprocess.run( + ["bash", "-c", script], capture_output=True, text=True, timeout=10, + ) + + +def rc_of(r): + m = re.search(r"^RC=(\d+)$", r.stdout, re.M) + assert m, "no RC line in output: %r / %r" % (r.stdout, r.stderr) + return int(m.group(1)) + + +def body(confdir): + with open(os.path.join(confdir, FILENAME)) as f: + return f.read() + + +def directives(text): + """The non-comment, non-blank lines — what NM actually parses.""" + return [l.strip() for l in text.splitlines() + if l.strip() and not l.lstrip().startswith("#")] + + +class ConfigureTunnelDnsOverTls(unittest.TestCase): + # ------------------------------------------------------------ normal ---- + def test_writes_a_connection_default_that_turns_dot_off(self): + with tempfile.TemporaryDirectory() as d: + r = run(d) + self.assertEqual(rc_of(r), 0) + lines = directives(body(d)) + self.assertIn("[connection-tunnel-dot]", lines) + self.assertIn("connection.dns-over-tls=0", lines) + + def test_matches_both_proton_interfaces_and_nothing_else(self): + # proton0 is the Proton client; wgpvpn is the static WireGuard + # profiles. A match-device line without both leaves one tunnel + # broken; one with a wildcard would turn DoT off for wifi too. + with tempfile.TemporaryDirectory() as d: + run(d) + match = [l for l in directives(body(d)) if l.startswith("match-device=")] + self.assertEqual(len(match), 1) + devices = match[0].split("=", 1)[1].split(",") + self.assertEqual(sorted(devices), + ["interface-name:proton0", "interface-name:wgpvpn"]) + + def test_explains_why_the_global_setting_is_overridden_per_link(self): + # The file contradicts the resolved drop-in next to it, so it has + # to carry the reason: the in-tunnel resolver that never completes + # TLS, and why the Proton client can't hold a per-profile setting. + with tempfile.TemporaryDirectory() as d: + run(d) + text = body(d).lower() + self.assertIn("10.2.0.1", text) + self.assertIn("853", text) + self.assertIn("recreates", text) + + def test_drop_in_is_world_readable_not_writable(self): + with tempfile.TemporaryDirectory() as d: + run(d) + mode = stat.S_IMODE(os.stat(os.path.join(d, FILENAME)).st_mode) + self.assertEqual(mode, 0o644) + + # ---------------------------------------------------------- boundary ---- + def test_running_twice_leaves_one_identical_file(self): + with tempfile.TemporaryDirectory() as d: + run(d) + first = body(d) + r = run(d) + self.assertEqual(rc_of(r), 0) + self.assertEqual(first, body(d)) + self.assertEqual(os.listdir(d), [FILENAME]) + + def test_absent_directory_is_created(self): + with tempfile.TemporaryDirectory() as d: + nested = os.path.join(d, "etc", "NetworkManager", "conf.d") + r = run(nested) + self.assertEqual(rc_of(r), 0) + self.assertIn("connection.dns-over-tls=0", directives(body(nested))) + + def test_leaves_sibling_drop_ins_alone(self): + # dns.conf, wifi-privacy.conf and wifi-powersave-off.conf share the + # directory; this step must add a file, never rewrite the dir. + with tempfile.TemporaryDirectory() as d: + sibling = os.path.join(d, "dns.conf") + with open(sibling, "w") as f: + f.write("[main]\ndns=systemd-resolved\n") + run(d) + with open(sibling) as f: + self.assertEqual(f.read(), "[main]\ndns=systemd-resolved\n") + + # ------------------------------------------------------------- error ---- + @unittest.skipUnless(os.geteuid() != 0, "root ignores directory write bits") + def test_unwritable_directory_warns_and_does_not_crash(self): + with tempfile.TemporaryDirectory() as d: + confdir = os.path.join(d, "ro") + os.mkdir(confdir) + os.chmod(confdir, 0o500) + try: + r = run(confdir) + self.assertIn("WARN:", r.stdout) + self.assertNotEqual(rc_of(r), 0) + finally: + os.chmod(confdir, 0o700) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/installer-steps/test_dotfiles_dependency_packages.py b/tests/installer-steps/test_dotfiles_dependency_packages.py new file mode 100644 index 0000000..58b7275 --- /dev/null +++ b/tests/installer-steps/test_dotfiles_dependency_packages.py @@ -0,0 +1,119 @@ +"""Pin the packages the dotfiles assume are present. + +Three packages are not optional extras: something the dotfiles ship depends +on each one, and when the package is missing the dependent silently does the +wrong thing rather than failing. + +- libreoffice-fresh :: common/.config/mimeapps.list maps presentations, + documents and spreadsheets to libreoffice-impress/-writer/-calc. With the + package absent those .desktop files don't exist, so xdg-mime falls through + to the next application claiming the type — on a machine with the winvm + dotfiles that is powerpoint.desktop, which boots a Windows VM to open a + deck (2026-09-18). +- imv :: gui-open --image execs imv, and without it every agent-side image + render fails with "required application is unavailable" (2026-09-18). +- git-lfs :: a repo tracking globs in LFS fails every checkout and merge + with "smudge filter lfs failed" (2026-09-20). + +All three were installed before the 2026-08-13 rebuild and absent after it, +which is the regression this pins: they are dependencies of shipped defaults, +so the installer has to declare them rather than leave them to whatever a +machine happens to carry. + +Method mirrors test_required_software: sed-extract supplemental_software from +the real `archsetup`, stub pacman_install as a recorder, run it, and assert +against what it actually invoked. Running the function beats matching its +source text, because the property under test is "every machine installs this" +and only a run resolves the conditionals that could make that false. The +function is run once per desktop environment for the same reason. + +Run from repo root: + python3 -m unittest tests.installer-steps.test_dotfiles_dependency_packages +""" + +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") + +DEPENDENCY_PACKAGES = ("libreoffice-fresh", "imv", "git-lfs") + +# Every desktop environment the installer branches on inside this function. +DESKTOP_ENVS = ("dwm", "hyprland") + + +def declared_packages(desktop_env): + """Return (exit_code, [pacman package, ...]) for one desktop environment. + + Everything the function calls besides pacman_install is stubbed to a no-op, + so the run records package declarations and nothing else. aur_install is + deliberately separate: these three are pacman packages, and folding the two + recorders together would let an AUR declaration satisfy the pin. + """ + script = textwrap.dedent(f"""\ + desktop_env={desktop_env} + display() {{ :; }} + aur_install() {{ :; }} + mask_fwupd_passim() {{ :; }} + run_task() {{ :; }} + error_warn() {{ :; }} + pacman_install() {{ echo "$1"; }} + source <(sed -n '/^supplemental_software() {{/,/^}}/p' "{ARCHSETUP}") + supplemental_software + """) + result = subprocess.run( + ["bash", "-c", script], capture_output=True, text=True, timeout=30, + ) + return result.returncode, result.stdout.split() + + +class DotfilesDependencyPackages(unittest.TestCase): + # ------------------------------------------------------------ normal ---- + def test_each_dependency_package_is_installed(self): + rc, pkgs = declared_packages("hyprland") + self.assertEqual(rc, 0) + for package in DEPENDENCY_PACKAGES: + with self.subTest(package=package): + self.assertIn( + package, pkgs, + f"{package} is a dependency of a shipped dotfiles default " + "and must be declared", + ) + + # ---------------------------------------------------------- boundary ---- + def test_each_is_declared_exactly_once(self): + # A second declaration is dead weight and drifts out of sync with the + # first when one of them is edited. + rc, pkgs = declared_packages("hyprland") + self.assertEqual(rc, 0) + for package in DEPENDENCY_PACKAGES: + with self.subTest(package=package): + self.assertEqual(pkgs.count(package), 1) + + # ------------------------------------------------------------- error ---- + def test_none_is_gated_behind_a_desktop_environment(self): + # The dotfiles defaults that need these apply on every DE, so a + # declaration reachable under only one of them would leave the same + # hole on the other. + for env in DESKTOP_ENVS: + rc, pkgs = declared_packages(env) + self.assertEqual(rc, 0) + for package in DEPENDENCY_PACKAGES: + with self.subTest(desktop_env=env, package=package): + self.assertIn(package, pkgs) + + def test_harness_observes_desktop_environment_gating(self): + # The control for the test above: ranger IS gated to dwm on purpose, so + # if this run can't see that, the DE-independence assertion is vacuous + # and would pass against a gated package too. + _, dwm = declared_packages("dwm") + _, hyprland = declared_packages("hyprland") + self.assertIn("ranger", dwm) + self.assertNotIn("ranger", hyprland) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/installer-steps/test_idempotency_cluster.py b/tests/installer-steps/test_idempotency_cluster.py index ecb279d..0cca750 100644 --- a/tests/installer-steps/test_idempotency_cluster.py +++ b/tests/installer-steps/test_idempotency_cluster.py @@ -114,6 +114,33 @@ class EnableUserService(unittest.TestCase): out = run("enable_user_service", one).stdout self.assertIn("RC=0", out) + def test_socket_unit_lands_in_sockets_target_wants(self): + # A socket unit's [Install] is WantedBy=sockets.target, so enabling it + # via default.target.wants would never socket-activate. The optional + # fifth arg names the wants target. + with tempfile.TemporaryDirectory() as home: + body = ( + f'enable_user_service {ME!r} podman.socket ' + f'/usr/lib/systemd/user/podman.socket {home!r} sockets.target\n' + f'link="{home}/.config/systemd/user/sockets.target.wants/podman.socket"\n' + f'[ -L "$link" ] && echo "LINK=yes" || echo "LINK=no"\n' + f'echo "TARGET=$(readlink "$link")"' + ) + out = run("enable_user_service", body).stdout + self.assertIn("LINK=yes", out) + self.assertIn("TARGET=/usr/lib/systemd/user/podman.socket", out) + + def test_omitted_target_still_defaults_to_default_target(self): + with tempfile.TemporaryDirectory() as home: + body = ( + f'enable_user_service {ME!r} gamemoded.service ' + f'/usr/lib/systemd/user/gamemoded.service {home!r}\n' + f'[ -L "{home}/.config/systemd/user/default.target.wants/gamemoded.service" ] ' + f'&& echo "DEFAULT=yes" || echo "DEFAULT=no"' + ) + out = run("enable_user_service", body).stdout + self.assertIn("DEFAULT=yes", out) + if __name__ == "__main__": unittest.main() diff --git a/tests/installer-steps/test_install_camera_passthrough_rules.py b/tests/installer-steps/test_install_camera_passthrough_rules.py new file mode 100644 index 0000000..ab28575 --- /dev/null +++ b/tests/installer-steps/test_install_camera_passthrough_rules.py @@ -0,0 +1,118 @@ +"""Test install_camera_passthrough_rules — the VM camera udev grant. + +usbredirect must open a camera's raw USB node read-write to claim it for the +Windows VM; the node defaults to root-owned with no group write, so the +attach fails with a bare "Failed to open device!". The rule grants +GROUP="video", MODE="0660" (the verified fix) and keeps TAG+="uaccess". + +The file NUMBER is load-bearing (winvm correction, 2026-08-08): the uaccess +ACL is applied by 73-seat-late.rules, so a 99- file adds the tag after that +already ran. The shipped filename must sort below 73. + +Method: sed-extract install_camera_passthrough_rules from the real +`archsetup`, point it at a temp rules path, fake display / error_warn. + +Run from repo root: + python3 -m unittest tests.installer-steps.test_install_camera_passthrough_rules +""" + +import os +import re +import stat +import subprocess +import tempfile +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(rules_path, pre=""): + script = textwrap.dedent(f"""\ + logfile=/dev/null + action="" + display() {{ :; }} + error_warn() {{ echo "WARN: $1"; return 1; }} + source <(sed -n '/^install_camera_passthrough_rules() {{/,/^}}/p' "{ARCHSETUP}") + {pre} + install_camera_passthrough_rules "{rules_path}" + echo "RC=$?" + echo "RULES:[$(cat "{rules_path}" 2>/dev/null)]" + exit 0 + """) + return subprocess.run( + ["bash", "-c", script], capture_output=True, text=True, timeout=10, + ) + + +def rules_body(r): + return r.stdout.split("RULES:[")[1].split("]")[0] + + +class InstallCameraPassthroughRules(unittest.TestCase): + # ------------------------------------------------------------ normal ---- + def test_writes_both_camera_rules_with_group_mode_and_tag(self): + with tempfile.TemporaryDirectory() as d: + r = run(os.path.join(d, "72-test.rules")) + body = rules_body(r) + for vendor, product in (("3564", "ff02"), ("046d", "085e")): + line = next((ln for ln in body.splitlines() + if f'ATTR{{idVendor}}=="{vendor}"' in ln), None) + assert line is not None, f"no rule line for {vendor}:{product}" + self.assertIn(f'ATTR{{idProduct}}=="{product}"', line) + self.assertIn('GROUP="video"', line) + self.assertIn('MODE="0660"', line) + self.assertIn('TAG+="uaccess"', line) + self.assertIn("RC=0", r.stdout) + + def test_default_filename_sorts_below_seat_late(self): + # The rule file's default install path must sort before + # 73-seat-late.rules or the uaccess tag lands too late to be ACLed. + with open(ARCHSETUP) as f: + src = f.read() + func = re.search( + r'^install_camera_passthrough_rules\(\)\s*{.*?^}', src, re.S | re.M) + assert func is not None, "function not found in archsetup" + m = re.search(r'\$\{1:-(/etc/udev/rules\.d/[^}]+)\}', func.group(0)) + assert m is not None, "default rules path not found in the function" + basename = os.path.basename(m.group(1)) + self.assertLess(basename, "73-seat-late.rules", + "the rules file must sort below 73-seat-late.rules") + + # ---------------------------------------------------------- boundary ---- + def test_rerun_is_idempotent(self): + with tempfile.TemporaryDirectory() as d: + path = os.path.join(d, "72-test.rules") + first = run(path) + body_one = rules_body(first) + second = run(path) + self.assertEqual(body_one, rules_body(second)) + self.assertIn("RC=0", second.stdout) + + def test_overwrites_a_stale_existing_file(self): + with tempfile.TemporaryDirectory() as d: + path = os.path.join(d, "72-test.rules") + with open(path, "w") as f: + f.write("# stale content that must not survive\n") + r = run(path) + self.assertNotIn("stale content", rules_body(r)) + self.assertIn('GROUP="video"', rules_body(r)) + + # ------------------------------------------------------------- error ---- + @unittest.skipUnless(os.geteuid() != 0, "root ignores directory write bits") + def test_unwritable_dir_warns_and_does_not_crash(self): + with tempfile.TemporaryDirectory() as d: + locked = os.path.join(d, "locked") + os.makedirs(locked) + os.chmod(locked, stat.S_IRUSR | stat.S_IXUSR) + r = run(os.path.join(locked, "72-test.rules")) + os.chmod(locked, stat.S_IRWXU) + self.assertIn("WARN:", r.stdout) + self.assertIn("RC=", r.stdout, + "the harness must reach its RC line — the function " + "returned rather than aborting the script") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/installer-steps/test_install_cpu_microcode.py b/tests/installer-steps/test_install_cpu_microcode.py new file mode 100644 index 0000000..20f1820 --- /dev/null +++ b/tests/installer-steps/test_install_cpu_microcode.py @@ -0,0 +1,110 @@ +"""Test install_cpu_microcode — vendor-detected microcode package install. + +archsetup never installed intel-ucode/amd-ucode (found on velox 2026-04-10: +CPU running old microcode, hand-fixed). The step reads vendor_id from +/proc/cpuinfo and installs the matching package. It must run before +configure_grub in boot_ux: grub-mkconfig detects /boot/<vendor>-ucode.img for +its initrd lines, and mkinitcpio's microcode hook embeds it, so the package +has to exist before either generates. + +Method: sed-extract install_cpu_microcode from the real `archsetup`, point it +at a fixture cpuinfo, fake pacman_install / display / error_warn. + +Run from repo root: + python3 -m unittest tests.installer-steps.test_install_cpu_microcode +""" + +import os +import re +import subprocess +import tempfile +import textwrap +import unittest + +REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")) +ARCHSETUP = os.path.join(REPO_ROOT, "archsetup") + +INTEL = "vendor_id\t: GenuineIntel\n" +AMD = "vendor_id\t: AuthenticAMD\n" + + +def run(cpuinfo_body, missing=False): + with tempfile.TemporaryDirectory() as d: + cpuinfo = os.path.join(d, "cpuinfo") + if not missing: + with open(cpuinfo, "w") as f: + f.write(cpuinfo_body) + script = textwrap.dedent(f"""\ + logfile=/dev/null + action="" + display() {{ :; }} + pacman_install() {{ echo "INSTALL: $1"; }} + error_warn() {{ echo "WARN: $1"; return 1; }} + source <(sed -n '/^install_cpu_microcode() {{/,/^}}/p' "{ARCHSETUP}") + install_cpu_microcode "{cpuinfo}" + echo "RC=$?" + exit 0 + """) + return subprocess.run( + ["bash", "-c", script], capture_output=True, text=True, timeout=10, + ) + + +def rc_of(r): + m = re.search(r"^RC=(\d+)$", r.stdout, re.M) + assert m, "no RC line in output: %r / %r" % (r.stdout, r.stderr) + return int(m.group(1)) + + +class InstallCpuMicrocode(unittest.TestCase): + # ------------------------------------------------------------ normal ---- + def test_intel_vendor_installs_intel_ucode(self): + r = run(f"processor\t: 0\n{INTEL}model name\t: whatever\n") + self.assertIn("INSTALL: intel-ucode", r.stdout) + self.assertNotIn("amd-ucode", r.stdout) + self.assertEqual(rc_of(r), 0) + + def test_amd_vendor_installs_amd_ucode(self): + r = run(f"processor\t: 0\n{AMD}") + self.assertIn("INSTALL: amd-ucode", r.stdout) + self.assertNotIn("intel-ucode", r.stdout) + self.assertEqual(rc_of(r), 0) + + # ---------------------------------------------------------- boundary ---- + def test_multicore_cpuinfo_installs_exactly_once(self): + # /proc/cpuinfo repeats vendor_id per logical CPU. + body = "".join(f"processor\t: {i}\n{AMD}\n" for i in range(16)) + r = run(body) + self.assertEqual(r.stdout.count("INSTALL:"), 1, + "one package install, not one per core") + + def test_space_separated_vendor_line_parses(self): + # Some kernels/arches pad with spaces rather than a tab. + r = run("vendor_id : GenuineIntel\n") + self.assertIn("INSTALL: intel-ucode", r.stdout) + + def test_vendor_id_substring_elsewhere_does_not_confuse(self): + # A flags line or model name mentioning a vendor string must not win + # over the real vendor_id line. + body = "model name\t: AuthenticAMD emulator\n" + INTEL + r = run(body) + self.assertIn("INSTALL: intel-ucode", r.stdout) + self.assertNotIn("amd-ucode", r.stdout) + + # ------------------------------------------------------------- error ---- + def test_unknown_vendor_warns_and_installs_nothing(self): + r = run("vendor_id\t: CentaurHauls\n") + self.assertNotIn("INSTALL:", r.stdout) + self.assertIn("WARN:", r.stdout) + self.assertEqual(rc_of(r), 1) + + def test_missing_cpuinfo_warns_and_installs_nothing(self): + r = run("", missing=True) + self.assertNotIn("INSTALL:", r.stdout) + self.assertIn("WARN:", r.stdout) + self.assertEqual(rc_of(r), 1) + self.assertEqual(r.returncode, 0) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/installer-steps/test_install_gnome_keyring_fix.py b/tests/installer-steps/test_install_gnome_keyring_fix.py new file mode 100644 index 0000000..1a4115a --- /dev/null +++ b/tests/installer-steps/test_install_gnome_keyring_fix.py @@ -0,0 +1,209 @@ +"""Tests for install_gnome_keyring_fix in the archsetup installer. + +gnome-keyring 51.1 (and 48 through 50) aborts when a Secret Service client's +first OpenSession is dispatched before the daemon has created its per-client +record. Every abort loses the unlocked state, so the next request pops the +keyring password dialog; on ratio this fired three times a day once the +ChatGPT desktop app's codex app-server started bursting keyring reads. The +fix is upstream merge request !112 plus an alias-lifetime correction, carried +as a patched rebuild of Arch's own package under packages/gnome-keyring in +this repo (pkgrel 1.1). install_gnome_keyring_fix builds and installs it with +makepkg as the user when the stock 1:51.1-1 is what is installed. + +Versioning is the whole contract. The rebuild sorts above stock 1:51.1-1 and +below any future 1:51.1-2 or 51.2, so a normal pacman upgrade replaces it the +moment Arch or upstream ships a fix, and the step must never pin or hold the +package. When something newer than the rebuild is already installed, the fix +is assumed upstream and the step skips with a warning so the assumption is +visible rather than silent. When the rebuild itself is installed, the step is +a no-op. + +These tests exercise the REAL function body, extracted from the `archsetup` +script at run time. pacman, vercmp, makepkg and sudo are replaced by +recorders driven by environment switches; the package directory is a temp +copy of the real packages/gnome-keyring tree. + +Run from repo root: + python3 -m unittest tests.installer-steps.test_install_gnome_keyring_fix +""" + +import os +import re +import shutil +import subprocess +import tempfile +import unittest + +REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")) +ARCHSETUP = os.path.join(REPO_ROOT, "archsetup") +PACKAGE_DIR = os.path.join(REPO_ROOT, "packages", "gnome-keyring") + +STUBS = r""" +logfile=/dev/null +action="" +username=tester +display() { echo "DISPLAY: $*"; } +pacman_install() { echo "PACMAN_INSTALL: $1" >> "$MARKER"; } +error_warn() { echo "WARN:$1"; return 1; } +# pacman -Q prints the installed version, or fails when INSTALLED is empty. +pacman() { + echo "PACMAN: $*" >> "$MARKER" + case "$1" in + -Q) [ -n "$INSTALLED" ] && echo "gnome-keyring $INSTALLED" ;; + *) return 0 ;; + esac +} +# The real vercmp is used so the ordering under test is pacman's own. +makepkg() { + echo "MAKEPKG: $* (pwd=$PWD)" >> "$MARKER" + [ "$MAKEPKG_FAILS" = 1 ] && return 2 + return 0 +} +# sudo -u user cmd... runs cmd in place; the recorder notes the user. +sudo() { + echo "SUDO: $*" >> "$MARKER" + shift 2 + "$@" +} +safe_rm_rf() { rm -rf "$1"; } +# chown needs root; the tests only care that the copy was staged. +chown() { echo "CHOWN: $*" >> "$MARKER"; } +""" + + +def run(pkg_dir, installed, makepkg_fails=False, source_dir=None): + marker = tempfile.NamedTemporaryFile(prefix="keyring-fix-", delete=False) + marker.close() + source_dir = source_dir or tempfile.mkdtemp(prefix="keyring-src-") + script = ( + STUBS + + "source <(sed -n '/^install_gnome_keyring_fix() {/,/^}/p' \"%s\")\n" + + 'user_archsetup_dir="%s"\n' + + 'source_dir="%s"\n' + + "install_gnome_keyring_fix\n" + + 'echo "RC=$?"\n' + ) % (ARCHSETUP, pkg_dir, source_dir) + env = dict(os.environ, MARKER=marker.name, INSTALLED=installed, + MAKEPKG_FAILS="1" if makepkg_fails else "0") + result = subprocess.run(["bash", "-c", script], capture_output=True, + text=True, timeout=30, env=env) + with open(marker.name) as f: + calls = f.read().splitlines() + os.remove(marker.name) + return result, calls, source_dir + + +class InstallGnomeKeyringFix(unittest.TestCase): + def setUp(self): + # A fake archsetup checkout holding a copy of the real package tree. + self.checkout = tempfile.mkdtemp(prefix="archsetup-checkout-") + self.addCleanup(shutil.rmtree, self.checkout, ignore_errors=True) + shutil.copytree(PACKAGE_DIR, os.path.join(self.checkout, "packages", "gnome-keyring")) + self.source_dir = tempfile.mkdtemp(prefix="keyring-src-") + self.addCleanup(shutil.rmtree, self.source_dir, ignore_errors=True) + + def _run(self, installed, **kw): + return run(self.checkout, installed, source_dir=self.source_dir, **kw) + + @staticmethod + def _makepkg_calls(calls): + return [c for c in calls if c.startswith("MAKEPKG:")] + + # ---------------------------------------------------------- normal ---- + def test_stock_51_1_gets_the_rebuild(self): + result, calls, _ = self._run("1:51.1-1") + self.assertEqual(len(self._makepkg_calls(calls)), 1, calls) + self.assertIn("RC=0", result.stdout) + self.assertNotIn("WARN:", result.stdout) + + def test_builds_as_the_user_with_syncdeps_and_install(self): + _, calls, _ = self._run("1:51.1-1") + self.assertTrue(any(c.startswith("SUDO: -u tester makepkg") for c in calls), calls) + makepkg = self._makepkg_calls(calls)[0] + self.assertIn("--noconfirm", makepkg) + # -s (sync makedepends) and -i (install), spelled separately or combined. + short = "".join(t[1:] for t in makepkg.split() if re.fullmatch(r"-[a-zA-Z]+", t)) + self.assertIn("s", short, makepkg) + self.assertIn("i", short, makepkg) + + def test_builds_from_a_user_owned_copy_not_the_checkout(self): + # makepkg writes into its directory; the checkout must stay clean. + _, calls, _ = self._run("1:51.1-1") + makepkg = self._makepkg_calls(calls)[0] + self.assertIn("pwd=" + self.source_dir, makepkg) + self.assertNotIn("pwd=" + self.checkout, makepkg) + self.assertTrue(os.path.exists(os.path.join(self.source_dir, "gnome-keyring", "PKGBUILD"))) + + def test_older_stock_releases_also_get_the_rebuild(self): + # 50.0 carries the same bug; the rebuild is 51.1 so it is an upgrade. + _, calls, _ = self._run("1:50.0-1") + self.assertEqual(len(self._makepkg_calls(calls)), 1) + + # -------------------------------------------------------- boundary ---- + def test_rebuild_already_installed_is_a_noop(self): + result, calls, _ = self._run("1:51.1-1.1") + self.assertEqual(self._makepkg_calls(calls), []) + self.assertIn("RC=0", result.stdout) + self.assertNotIn("WARN:", result.stdout) + + def test_newer_package_skips_visibly_without_counting_an_error(self): + for newer in ("1:51.1-2", "1:51.2-1", "1:52.0-1"): + with self.subTest(installed=newer): + result, calls, _ = self._run(newer) + self.assertEqual(self._makepkg_calls(calls), [], newer) + self.assertIn("newer than the rebuild", result.stdout, + "a newer package means the fix is assumed upstream; say so") + self.assertNotIn("WARN:", result.stdout, + "a deliberate skip must not count as an installer error") + self.assertIn("RC=0", result.stdout) + + def test_rerun_after_success_does_not_rebuild(self): + _, calls, _ = self._run("1:51.1-1") + self.assertEqual(len(self._makepkg_calls(calls)), 1) + _, calls, _ = self._run("1:51.1-1.1") + self.assertEqual(self._makepkg_calls(calls), []) + + def test_missing_package_installs_stock_first_then_builds(self): + # With no gnome-keyring present, stock is installed as the baseline + # and the rebuild then upgrades it; both happen in that order. + result, calls, _ = self._run("") + self.assertEqual(len(self._makepkg_calls(calls)), 1) + stock = [i for i, c in enumerate(calls) if c.startswith("PACMAN_INSTALL: gnome-keyring")] + build = [i for i, c in enumerate(calls) if c.startswith("MAKEPKG:")] + self.assertEqual(len(stock), 1, calls) + self.assertLess(stock[0], build[0], "stock install must precede the build") + self.assertIn("RC=0", result.stdout) + + def test_rerun_with_rebuild_installed_never_touches_stock(self): + # pacman's --needed only skips an equal version, so a stock install + # ahead of the rebuild would downgrade it on every re-run. The step + # must not call the stock install at all once anything is installed. + for installed in ("1:51.1-1.1", "1:51.1-1", "1:51.2-1"): + with self.subTest(installed=installed): + _, calls, _ = self._run(installed) + self.assertFalse(any(c.startswith("PACMAN_INSTALL:") for c in calls), calls) + + # ----------------------------------------------------------- error ---- + def test_failed_build_warns_and_leaves_stock_in_place(self): + result, calls, _ = self._run("1:51.1-1", makepkg_fails=True) + self.assertIn("WARN:", result.stdout) + self.assertNotIn("RC=0", result.stdout) + self.assertFalse(any(c.startswith("PACMAN: -U") for c in calls), + "nothing may be installed by hand after a failed build") + + def test_missing_recipe_warns_instead_of_building(self): + shutil.rmtree(os.path.join(self.checkout, "packages", "gnome-keyring")) + result, calls, _ = self._run("1:51.1-1") + self.assertEqual(self._makepkg_calls(calls), []) + self.assertIn("WARN:", result.stdout) + + def test_never_pins_or_holds_the_package(self): + # No IgnorePkg edit and no pacman -D hold: the next upstream fix must + # replace the rebuild on a normal upgrade. + _, calls, _ = self._run("1:51.1-1") + self.assertFalse(any("IgnorePkg" in c for c in calls)) + self.assertFalse(any(c.startswith("PACMAN: -D") for c in calls)) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/installer-steps/test_mask_fwupd_passim.py b/tests/installer-steps/test_mask_fwupd_passim.py new file mode 100644 index 0000000..8977504 --- /dev/null +++ b/tests/installer-steps/test_mask_fwupd_passim.py @@ -0,0 +1,79 @@ +"""Test mask_fwupd_passim — keep fwupd's LAN metadata daemon off. + +fwupd pulls in passim, a daemon that shares firmware metadata with other +machines on the LAN by listening publicly on 0.0.0.0:27500. Any fwupdmgr +run D-Bus-activates it, and because the unit is static (no [Install] +section) `systemctl disable` is a no-op: it came back on velox the next +time fwupdmgr ran (2026-09-12). Masking is what holds, and it is how +ratio has carried it since 2026-07-21. + +Method: sed-extract mask_fwupd_passim from the real `archsetup`; fake +run_task (minus error_warn, so the failure test sees the function's own +return path) and systemctl, and assert on the call. + + python3 -m unittest tests.installer-steps.test_mask_fwupd_passim +""" + +import os +import re +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(systemctl_body='echo "SYSTEMCTL: $*";'): + script = textwrap.dedent(f"""\ + logfile=/dev/null + action="" + display() {{ :; }} + run_task() {{ echo "TASK: $1"; shift; "$@"; }} + systemctl() {{ {systemctl_body} }} + error_warn() {{ echo "WARN: $1"; return 1; }} + source <(sed -n '/^mask_fwupd_passim() {{/,/^}}/p' "{ARCHSETUP}") + mask_fwupd_passim + echo "RC=$?" + exit 0 + """) + return subprocess.run( + ["bash", "-c", script], capture_output=True, text=True, timeout=10, + ) + + +def rc_of(r): + m = re.search(r"^RC=(\d+)$", r.stdout, re.M) + assert m, "no RC line in output: %r / %r" % (r.stdout, r.stderr) + return int(m.group(1)) + + +class MaskFwupdPassim(unittest.TestCase): + # ------------------------------------------------------------ normal ---- + def test_masks_the_passim_unit(self): + r = run() + self.assertIn("SYSTEMCTL: mask passim.service", r.stdout) + self.assertEqual(rc_of(r), 0) + + def test_masks_rather_than_disables(self): + # disable is a no-op on a static unit and is the mistake this step + # exists to avoid; a mask is the only call that sticks. + r = run() + calls = [l for l in r.stdout.splitlines() if l.startswith("SYSTEMCTL:")] + self.assertEqual(calls, ["SYSTEMCTL: mask passim.service"]) + + # ---------------------------------------------------------- boundary ---- + # Re-running is idempotent because `systemctl mask` on an already-masked + # unit exits 0 and changes nothing; that property lives in systemctl, so + # there is no stateless test here that could tell it apart from a pass. + + # ------------------------------------------------------------- error ---- + def test_failed_mask_reports_nonzero(self): + # The fake run_task returns the command's status without the real + # error_warn wiring, so this checks the function's own return path. + r = run(systemctl_body='echo "SYSTEMCTL: $*"; return 1;') + self.assertNotEqual(rc_of(r), 0) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/installer-steps/test_mask_tmp_mount_for_zfs.py b/tests/installer-steps/test_mask_tmp_mount_for_zfs.py new file mode 100644 index 0000000..f03086f --- /dev/null +++ b/tests/installer-steps/test_mask_tmp_mount_for_zfs.py @@ -0,0 +1,109 @@ +"""Test mask_tmp_mount_for_zfs — the tmpfs-over-ZFS /tmp guard. + +When the pool carries a /tmp dataset, systemd's tmp.mount (tmpfs) races it at +boot; when tmpfs wins, the dataset is shadowed and systemd-tmpfiles-clean +fails repeatedly with "Protocol driver not attached" (velox, 2026-04-10). The +fix is masking tmp.mount so the dataset owns /tmp — but only when such a +dataset actually exists, and never on a machine without zfs at all. + +Method: sed-extract mask_tmp_mount_for_zfs from the real `archsetup`; fake +zfs / run_task / systemctl. The zfs-binary-absent case runs with a stripped +PATH containing only the tools the function itself needs. + +Run from repo root: + python3 -m unittest tests.installer-steps.test_mask_tmp_mount_for_zfs +""" + +import os +import re +import subprocess +import tempfile +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(zfs_body=None, strip_zfs_from_path=False): + """zfs_body: bash body for a fake zfs function, or None for no fake. + strip_zfs_from_path: run with a minimal PATH that has grep/sed but no zfs. + """ + with tempfile.TemporaryDirectory() as d: + path_setup = "" + if strip_zfs_from_path: + fakebin = os.path.join(d, "bin") + os.makedirs(fakebin) + for tool in ("grep", "sed", "cat"): + src = subprocess.run(["bash", "-lc", f"command -v {tool}"], + capture_output=True, text=True).stdout.strip() + if src: + os.symlink(src, os.path.join(fakebin, tool)) + path_setup = f'PATH="{fakebin}"' + zfs_fake = f"zfs() {{ {zfs_body} }}" if zfs_body is not None else "" + script = textwrap.dedent(f"""\ + logfile=/dev/null + action="" + display() {{ :; }} + run_task() {{ echo "TASK: $1"; shift; "$@"; }} + systemctl() {{ echo "SYSTEMCTL: $*"; }} + error_warn() {{ echo "WARN: $1"; return 1; }} + {zfs_fake} + source <(sed -n '/^mask_tmp_mount_for_zfs() {{/,/^}}/p' "{ARCHSETUP}") + {path_setup} + mask_tmp_mount_for_zfs + echo "RC=$?" + exit 0 + """) + return subprocess.run( + ["bash", "-c", script], capture_output=True, text=True, timeout=10, + ) + + +def rc_of(r): + m = re.search(r"^RC=(\d+)$", r.stdout, re.M) + assert m, "no RC line in output: %r / %r" % (r.stdout, r.stderr) + return int(m.group(1)) + + +class MaskTmpMountForZfs(unittest.TestCase): + # ------------------------------------------------------------ normal ---- + def test_tmp_dataset_present_masks_tmp_mount(self): + r = run(zfs_body='printf "/\\n/home\\n/tmp\\n";') + self.assertIn("SYSTEMCTL: mask tmp.mount", r.stdout) + self.assertEqual(rc_of(r), 0) + + # ---------------------------------------------------------- boundary ---- + def test_no_tmp_dataset_is_a_no_op(self): + r = run(zfs_body='printf "/\\n/home\\n/var\\n";') + self.assertNotIn("SYSTEMCTL:", r.stdout) + self.assertEqual(rc_of(r), 0) + + def test_tmp_prefix_dataset_does_not_match(self): + # /tmp/scratch or /tmpfoo must not trigger the mask — exact match only. + r = run(zfs_body='printf "/\\n/tmp/scratch\\n/tmpfoo\\n";') + self.assertNotIn("SYSTEMCTL:", r.stdout) + self.assertEqual(rc_of(r), 0) + + def test_legacy_and_none_mountpoints_do_not_match(self): + r = run(zfs_body='printf "legacy\\nnone\\n-\\n";') + self.assertNotIn("SYSTEMCTL:", r.stdout) + self.assertEqual(rc_of(r), 0) + + # ------------------------------------------------------------- error ---- + def test_zfs_binary_absent_is_a_silent_no_op(self): + r = run(zfs_body=None, strip_zfs_from_path=True) + self.assertNotIn("SYSTEMCTL:", r.stdout) + self.assertNotIn("WARN:", r.stdout) + self.assertEqual(rc_of(r), 0) + + def test_zfs_list_failure_is_a_no_op_not_a_crash(self): + r = run(zfs_body='echo "no pools available" >&2; return 1;') + self.assertNotIn("SYSTEMCTL:", r.stdout) + self.assertEqual(rc_of(r), 0) + self.assertNotIn("no pools available", r.stdout, + "zfs stderr noise must not leak into output") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/installer-steps/test_orchestrators.py b/tests/installer-steps/test_orchestrators.py index a2b235b..c2eb710 100644 --- a/tests/installer-steps/test_orchestrators.py +++ b/tests/installer-steps/test_orchestrators.py @@ -28,6 +28,7 @@ ARCHSETUP = os.path.join(REPO_ROOT, "archsetup") ORCHESTRATORS = { "essential_services": [ "configure_randomness", "configure_networking", "configure_power", + "configure_hibernate_delay", "configure_backlight_access", "configure_ssh_server", "configure_fail2ban", "configure_firewall", "configure_service_discovery", "configure_job_scheduling", "configure_package_cache", "configure_snapshots", @@ -43,10 +44,10 @@ ORCHESTRATORS = { "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", + "install_cpu_microcode", "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", @@ -88,7 +89,8 @@ class OrchestratorSequence(unittest.TestCase): class SnapshotDispatch(unittest.TestCase): """configure_snapshots branches on filesystem; pin each branch.""" - SUBS = ["configure_zfs_snapshots", "configure_btrfs_snapshots"] + SUBS = ["configure_zfs_snapshots", "configure_btrfs_snapshots", + "mask_tmp_mount_for_zfs"] def test_zfs_root_runs_zfs_snapshots(self): result = run_orchestrator( @@ -96,7 +98,8 @@ class SnapshotDispatch(unittest.TestCase): 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"]) + self.assertEqual(result.stdout.split(), + ["configure_zfs_snapshots", "mask_tmp_mount_for_zfs"]) def test_btrfs_root_runs_btrfs_snapshots(self): result = run_orchestrator( diff --git a/tests/net-scenarios/test_run_net_scenarios.py b/tests/net-scenarios/test_run_net_scenarios.py index 1d92185..a9cd275 100644 --- a/tests/net-scenarios/test_run_net_scenarios.py +++ b/tests/net-scenarios/test_run_net_scenarios.py @@ -63,6 +63,12 @@ class RunNetScenarios(unittest.TestCase): return subprocess.run( ["bash", SCRIPT, "--target", "root@fake-vm"], capture_output=True, text=True, timeout=20, env=env, + # The stubbed ssh is `cat >/dev/null`, which drains stdin to EOF. + # Without this the stub inherits whatever stdin the test runner + # had, so `make test-unit` passed when stdin was redirected and + # hung on all five tests when it was a terminal or a live pipe -- + # which is how it gets run by hand. + stdin=subprocess.DEVNULL, ) def test_all_checks_pass_exits_zero(self): diff --git a/tests/post-rebuild-check/test_post_rebuild_check.py b/tests/post-rebuild-check/test_post_rebuild_check.py new file mode 100644 index 0000000..1145f34 --- /dev/null +++ b/tests/post-rebuild-check/test_post_rebuild_check.py @@ -0,0 +1,1145 @@ +"""Tests for the post-rebuild-check script. + +A rebuilt machine looks finished and isn't: on velox 2026-08-13 five gaps +surfaced within two days, three of which LOOKED fine (a stowed unit file, an +enabled timer, a present git clone). The script runs the checks from the +post-rebuild task and turns each silent no-op into a visible line: + + 1. failed systemd units (user and system scope) + 2. user unit files present but not enabled (linked-and-inert timers) + 3. tracked *.example files whose real sibling is missing + 4. gitignore-mode projects missing tooling paths their own .gitignore names + 5. signal-cli holds no registered account + 6. every NTP source named by hostname (a wrong clock takes DNS with it) + 7. hypridle installed but not running (nothing triggers idle suspend) + 8. a working repo cloned read-only (push returns 403) + +Exit 0 with every check clean, 1 when any check found something. + +Test seams (env vars the production script honors; for each, SET-BUT-EMPTY +means "the real probe ran and found nothing", UNSET means "run the real +probe"): + PRC_FAILED_UNITS newline list of "scope:unit" (scope user|system) + PRC_UNIT_STATES newline list of "unit-file state" for the user unit dir + PRC_LOCAL_SCAN_ROOTS newline-separated roots to scan for *.example orphans + PRC_PROJECT_ROOTS newline-separated project dirs for the tooling check + PRC_SIGNAL_ACCOUNTS signal-cli listAccounts output ("" = no accounts); + PRC_NTP_SOURCES newline list of configured NTP server addresses + ("MISSING" = no NTP daemon active) + the special value MISSING means the binary is absent + PRC_IDLE_DAEMON pgrep output for hypridle ("" = installed but not + running; "MISSING" = not installed on this machine) + PRC_REPO_REMOTES newline list of "path<space>origin-url" for the + push-capability check ("" = no repos to check) + PRC_UNITS_EXPECTED_DISABLED + newline list of units whose not-enabled state is + deliberate on this machine ("" = no exemptions) + +Run from repo root: + python3 -m unittest tests.post-rebuild-check.test_post_rebuild_check +""" + +import os +import shutil +import subprocess +import tempfile +import time +import unittest + + +REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")) +CHECK = os.path.join(REPO_ROOT, "scripts", "post-rebuild-check") + + +def run_check(failed_units="", unit_states="", local_roots="", + project_roots="", signal_accounts="+15045551234", + ntp_sources="162.159.200.1\npool.ntp.org", + idle_daemon="4242", repo_remotes="", + units_expected_disabled=""): + """Run the script with every probe stubbed; defaults are all-clean. + + Roots are newline-separated. Empty means "the seam is set and names no + roots" -- the script tests with ${VAR+set}, so an empty value is still + set and never falls through to the real probe. + """ + env = dict(os.environ) + env["PRC_FAILED_UNITS"] = failed_units + env["PRC_UNIT_STATES"] = unit_states + env["PRC_LOCAL_SCAN_ROOTS"] = local_roots + env["PRC_PROJECT_ROOTS"] = project_roots + env["PRC_SIGNAL_ACCOUNTS"] = signal_accounts + env["PRC_NTP_SOURCES"] = ntp_sources + env["PRC_IDLE_DAEMON"] = idle_daemon + env["PRC_REPO_REMOTES"] = repo_remotes + env["PRC_UNITS_EXPECTED_DISABLED"] = units_expected_disabled + return subprocess.run( + ["sh", CHECK], capture_output=True, text=True, timeout=30, env=env, + ) + + +class NtpBootstrap(unittest.TestCase): + """Check 6 — the clock/DNS bootstrap deadlock. + + A wrong clock fails the DoT certificate and DNSSEC signature checks this + machine's DNS runs on, so nothing resolves; and an NTP daemon whose every + source is a hostname then cannot resolve the servers that would correct + the clock. One source addressed by IP is what makes the machine able to + recover on its own. + """ + + # --- Normal cases --------------------------------------------------- + + def test_an_ip_addressed_source_is_clean(self): + r = run_check(ntp_sources="162.159.200.1\npool.ntp.org") + self.assertIn("check 6/8: NTP bootstrap — ok", r.stdout) + self.assertEqual(r.returncode, 0, r.stdout) + + def test_all_hostname_sources_is_a_finding(self): + # The velox 2026-08-19 shape exactly: stock Arch chrony.conf, whose + # only source is a pool hostname. + r = run_check(ntp_sources="2.arch.pool.ntp.org") + self.assertIn("every NTP source is named by hostname", r.stdout) + self.assertEqual(r.returncode, 1) + + def test_an_ipv6_addressed_source_counts(self): + r = run_check(ntp_sources="2606:4700:f1::1") + self.assertIn("check 6/8: NTP bootstrap — ok", r.stdout) + + # --- Boundary cases ------------------------------------------------- + + def test_the_literal_may_sit_anywhere_in_the_list(self): + # Order must not matter; the property is "at least one", and the + # drop-in that carries it is read after the main config. + r = run_check(ntp_sources="a.pool.ntp.org\nb.pool.ntp.org\n162.159.200.1") + self.assertIn("check 6/8: NTP bootstrap — ok", r.stdout) + + def test_blank_lines_between_sources_are_ignored(self): + r = run_check(ntp_sources="\n\n162.159.200.1\n\n") + self.assertIn("check 6/8: NTP bootstrap — ok", r.stdout) + + def test_a_hostname_containing_digits_and_dots_is_not_an_address(self): + # The trap in any naive "looks like an IP" test: these resolve through + # DNS like any other name, so counting one as an address would hand a + # deadlocked machine a clean bill. + for host in ("0.arch.pool.ntp.org", "3.us.pool.ntp.org", "time1.google.com"): + with self.subTest(host=host): + r = run_check(ntp_sources=host) + self.assertIn("every NTP source is named by hostname", r.stdout) + + # --- Error cases ---------------------------------------------------- + + def test_no_ntp_daemon_is_a_finding(self): + r = run_check(ntp_sources="MISSING") + self.assertIn("no NTP implementation is active", r.stdout) + self.assertEqual(r.returncode, 1) + + def test_no_sources_configured_is_a_finding(self): + # Fails closed: an empty list proves nothing about the machine, and + # reporting ok would be a false pass on a box with no time sync at all. + r = run_check(ntp_sources="") + self.assertIn("no NTP sources are configured", r.stdout) + self.assertEqual(r.returncode, 1) + + # --- the confdir false pass ------------------------------------------ + # + # A drop-in is inert unless chrony.conf names its directory, and Arch's + # stock chrony.conf names none. Reading the drop-in without checking for + # confdir would find the IP-addressed source, call the machine healthy, and + # be describing a file chrony never opens — a false pass on exactly the + # misconfiguration this check exists to catch. + + def _chrony_fixture(self, main_lines, dropin_lines=None): + """Write a chrony.conf (plus an adjacent drop-in dir) and return its path.""" + d = tempfile.mkdtemp(prefix="prc-chrony-") + self.addCleanup(shutil.rmtree, d, True) + dropin_dir = os.path.join(d, "chrony.d") + os.makedirs(dropin_dir) + if dropin_lines is not None: + with open(os.path.join(dropin_dir, "10-bootstrap-ip-ntp.conf"), "w") as f: + f.write(dropin_lines) + conf = os.path.join(d, "chrony.conf") + with open(conf, "w") as f: + f.write(main_lines.replace("@DROPIN@", dropin_dir)) + return conf + + def _run_real_probe(self, chrony_conf): + """Run with PRC_NTP_SOURCES unset so the real chrony reader runs.""" + env = dict(os.environ) + env.update({"PRC_FAILED_UNITS": "", "PRC_UNIT_STATES": "", + "PRC_LOCAL_SCAN_ROOTS": "", "PRC_PROJECT_ROOTS": "", + "PRC_SIGNAL_ACCOUNTS": "+15045551234", + "PRC_IDLE_DAEMON": "4242", + "PRC_REPO_REMOTES": "", + "PRC_UNITS_EXPECTED_DISABLED": "", + "PRC_CHRONY_CONF": chrony_conf}) + env.pop("PRC_NTP_SOURCES", None) + return subprocess.run(["sh", CHECK], capture_output=True, text=True, + timeout=30, env=env) + + def test_dropin_without_confdir_does_not_count(self): + # The regression. The IP-addressed source is present on disk but + # chrony.conf never points at it, so the machine is still deadlock-prone + # and the check has to say so. + conf = self._chrony_fixture("pool 2.arch.pool.ntp.org iburst\n", + "server 162.159.200.1 iburst\n") + r = self._run_real_probe(conf) + if "no NTP implementation is active" in r.stdout: + self.skipTest("no chronyd on this host — the reader branch can't run") + self.assertIn("every NTP source is named by hostname", r.stdout) + + def test_dropin_with_confdir_counts(self): + # The same two files, with chrony.conf actually naming the directory. + conf = self._chrony_fixture( + "pool 2.arch.pool.ntp.org iburst\nconfdir @DROPIN@\n", + "server 162.159.200.1 iburst\n") + r = self._run_real_probe(conf) + if "no NTP implementation is active" in r.stdout: + self.skipTest("no chronyd on this host — the reader branch can't run") + self.assertIn("check 6/8: NTP bootstrap — ok", r.stdout) + + def test_confdir_naming_an_empty_directory_is_not_a_pass(self): + # confdir present, nothing behind it: the sources are the hostname-only + # main file, so the finding stands. + conf = self._chrony_fixture( + "pool 2.arch.pool.ntp.org iburst\nconfdir @DROPIN@\n", None) + r = self._run_real_probe(conf) + if "no NTP implementation is active" in r.stdout: + self.skipTest("no chronyd on this host — the reader branch can't run") + self.assertIn("every NTP source is named by hostname", r.stdout) + + def test_unset_seam_falls_through_to_the_real_probe(self): + # Same contract as every other seam: unset means "really look", so a + # caller who forgets the variable cannot silently skip the check. + env = dict(os.environ) + env.update({"PRC_FAILED_UNITS": "", "PRC_UNIT_STATES": "", + "PRC_LOCAL_SCAN_ROOTS": "", "PRC_PROJECT_ROOTS": "", + "PRC_SIGNAL_ACCOUNTS": "+15045551234"}) + env.pop("PRC_NTP_SOURCES", None) + r = subprocess.run(["sh", CHECK], capture_output=True, text=True, + timeout=30, env=env) + self.assertIn("check 6/8: NTP bootstrap", r.stdout) + + +class IdleDaemon(unittest.TestCase): + """Check 7 — whether anything still triggers idle lock and suspend. + + A laptop that never sleeps has no symptom until the battery is gone, which + is why this needs a check rather than trusting the desktop to look right. + On velox 2026-08-19 hypridle started cleanly at 15:29:48 and `settings + restore` killed it six seconds later, replaying a caffeine stored in an + earlier boot. The machine then ran 11h40m fully awake on battery, died when + it flattened, and reset its RTC — which took DNS down with it, the very + deadlock check 6 exists for. Nothing looked wrong at any point. + + The check is behavioural: it asks whether the daemon is alive, not why it + might not be, so a stale caffeine, a crash and a bad config all surface the + same way. + """ + + # --- Normal cases --------------------------------------------------- + + def test_running_daemon_is_clean(self): + r = run_check(idle_daemon="4242") + self.assertEqual(r.returncode, 0, r.stdout + r.stderr) + self.assertNotIn("DEVIATION", r.stdout) + + def test_installed_but_not_running_flags(self): + r = run_check(idle_daemon="") + self.assertEqual(r.returncode, 1) + self.assertIn("hypridle", r.stdout) + self.assertIn("DEVIATION", r.stdout) + + def test_the_finding_names_the_consequence_not_just_the_process(self): + # "hypridle is not running" reads as a detail. The reason it matters is + # that the machine stays awake until the battery is gone, and that is + # what has to be in the line someone skims at 1am. + r = run_check(idle_daemon="") + self.assertIn("awake", r.stdout.lower()) + + def test_the_finding_names_the_known_cause(self): + # Behavioural checks are cheap to write and expensive to act on. Naming + # the one cause already seen saves the reader the investigation this + # session had to do from scratch. + r = run_check(idle_daemon="") + self.assertIn("caffeine", r.stdout.lower()) + + # --- Boundary cases ------------------------------------------------- + + def test_not_installed_is_not_a_finding(self): + # A headless or dwm machine never installs hypridle — archsetup pulls + # it in only for Hyprland. Flagging its absence there would be noise on + # every run, and noise is how a real finding gets skimmed past. + r = run_check(idle_daemon="MISSING") + self.assertEqual(r.returncode, 0, r.stdout + r.stderr) + self.assertNotIn("DEVIATION", r.stdout) + + def test_several_pids_still_read_as_running(self): + # pgrep prints one pid per line. More than one is its own problem (five + # concurrent daemons wedged a velox session on 2026-07-22) but it is + # not *this* check's, and it must not read as "not running". + r = run_check(idle_daemon="4242\n4243") + self.assertEqual(r.returncode, 0, r.stdout + r.stderr) + + def test_the_check_always_prints_its_line(self): + for pids in ("4242", "", "MISSING"): + with self.subTest(pids=pids): + self.assertIn("idle daemon", + run_check(idle_daemon=pids).stdout.lower()) + + +class UnitsExpectedDisabled(unittest.TestCase): + """Check 2 — units nothing intends to enable on this machine. + + "Enabled" is the check's proxy for "will actually run", and the proxy is + wrong for a unit nobody means to enable here. velox carries four such + units, for four different reasons: geoclue-agent is redundant because + hyprland's exec-once starts the binary directly, emacs is started on demand + by emacsclient, obs-record-watchdog only matters while recording, and + obsbot-wb-guard needs an OBSBOT the machine doesn't have. + + Left unexempted they report at every run, which is the standing-findings + problem check 4's own comment already argues against: four permanent lines + in front of every real one teach you to skim the output. + + The exemption is machine-local rather than a marker in the shared unit + file, because obsbot-wb-guard is correctly ENABLED on ratio. Same unit, + different right answer per machine. + """ + + # --- Normal cases --------------------------------------------------- + + def test_an_exempt_unit_is_not_flagged(self): + r = run_check(unit_states="emacs.service linked", + units_expected_disabled="emacs.service") + self.assertEqual(r.returncode, 0, r.stdout) + self.assertNotIn("emacs.service", r.stdout) + + def test_a_non_exempt_unit_still_flags(self): + r = run_check(unit_states="roam-sync.timer linked", + units_expected_disabled="emacs.service") + self.assertEqual(r.returncode, 1) + self.assertIn("roam-sync.timer", r.stdout) + + def test_several_exemptions_all_apply(self): + r = run_check( + unit_states=("emacs.service linked\n" + "geoclue-agent.service linked\n" + "obsbot-wb-guard.service linked"), + units_expected_disabled=("emacs.service\n" + "geoclue-agent.service\n" + "obsbot-wb-guard.service")) + self.assertEqual(r.returncode, 0, r.stdout) + + # --- Boundary cases ------------------------------------------------- + + def test_an_exemption_that_is_actually_enabled_is_a_finding(self): + # A stale exemption must surface rather than sit there suppressing + # nothing. Otherwise the list rots into a place real findings go to + # die, which is worse than the noise it was added to remove. + r = run_check(unit_states="obsbot-wb-guard.service enabled", + units_expected_disabled="obsbot-wb-guard.service") + self.assertEqual(r.returncode, 1) + self.assertIn("obsbot-wb-guard.service", r.stdout) + + def test_comments_and_blank_lines_are_ignored(self): + # The reason a unit is exempt is the most useful thing about the + # entry, so the format has to hold a comment next to it. + r = run_check(unit_states="emacs.service linked", + units_expected_disabled=("# started on demand\n" + "\n" + "emacs.service # not by systemd\n")) + self.assertEqual(r.returncode, 0, r.stdout) + + def test_no_exemptions_flags_everything_as_before(self): + r = run_check(unit_states="emacs.service linked", + units_expected_disabled="") + self.assertEqual(r.returncode, 1) + self.assertIn("emacs.service", r.stdout) + + def test_an_exemption_does_not_suppress_a_dangling_link(self): + # A stowed unit pointing at a missing target is a different finding, + # decided on the filesystem. Exempting the name must not hide that. + d = tempfile.mkdtemp(prefix="prc-units-") + self.addCleanup(shutil.rmtree, d, True) + unit_dir = os.path.join(d, "systemd", "user") + os.makedirs(unit_dir) + link = os.path.join(unit_dir, "emacs.service") + os.symlink(os.path.join(d, "gone.service"), link) + env = dict(os.environ) + env.update({"PRC_FAILED_UNITS": "", "PRC_LOCAL_SCAN_ROOTS": "", + "PRC_PROJECT_ROOTS": "", + "PRC_SIGNAL_ACCOUNTS": "+15045551234", + "PRC_NTP_SOURCES": "162.159.200.1", + "PRC_IDLE_DAEMON": "4242", + "PRC_REPO_REMOTES": "", + "PRC_UNITS_EXPECTED_DISABLED": "emacs.service", + "XDG_CONFIG_HOME": d}) + env.pop("PRC_UNIT_STATES", None) + r = subprocess.run(["sh", CHECK], capture_output=True, text=True, + timeout=30, env=env) + self.assertIn("points at a missing target", r.stdout) + self.assertEqual(r.returncode, 1) + + +class RepoPushCapability(unittest.TestCase): + """Check 8 — a working repo cloned from the read-only endpoint. + + archsetup clones the user's own archsetup and dotfiles from + https://git.cjennings.net/..., the anonymous read-only endpoint. That is + the right default for a stranger installing archsetup, who has no key on + the server, and the wrong one for this machine, which has to push. The + override exists (ARCHSETUP_REPO / DOTFILES_REPO) but only applies where it + is configured — a curl|bash install, or a rebuild from a stock ISO, picks + the default straight back up. + + Nothing about the tree shows it. The clone is complete and ordinary, and + the machine finds out at the first push, with a 403. That is how velox's + dotfiles remote was found on 2026-08-17, four days after its rebuild. + + Only the read-only endpoint is flagged. An https remote to some other host + may well be pushable with a credential helper, and guessing about hosts + this machine does not own would put standing noise in front of the real + findings. + """ + + RO = "https://git.cjennings.net/dotfiles.git" + RW = "git@cjennings.net:dotfiles.git" + + # --- Normal cases --------------------------------------------------- + + def test_an_ssh_remote_is_clean(self): + r = run_check(repo_remotes=f"/home/x/.dotfiles {self.RW}") + self.assertEqual(r.returncode, 0, r.stdout) + self.assertNotIn("DEVIATION", r.stdout) + + def test_the_read_only_endpoint_flags(self): + r = run_check(repo_remotes=f"/home/x/.dotfiles {self.RO}") + self.assertEqual(r.returncode, 1) + self.assertIn("/home/x/.dotfiles", r.stdout) + + def test_the_finding_names_the_consequence(self): + # "the remote is https" is a detail. That pushing fails is the point. + r = run_check(repo_remotes=f"/home/x/.dotfiles {self.RO}") + self.assertIn("push", r.stdout.lower()) + + def test_every_offending_repo_is_named(self): + r = run_check(repo_remotes=(f"/home/x/.dotfiles {self.RO}\n" + f"/home/x/code/archsetup {self.RO}")) + self.assertIn("/home/x/.dotfiles", r.stdout) + self.assertIn("/home/x/code/archsetup", r.stdout) + + # --- Boundary cases ------------------------------------------------- + + def test_a_mixed_set_flags_only_the_read_only_one(self): + r = run_check(repo_remotes=(f"/home/x/.dotfiles {self.RW}\n" + f"/home/x/code/archsetup {self.RO}")) + self.assertEqual(r.returncode, 1) + self.assertIn("/home/x/code/archsetup", r.stdout) + self.assertNotIn("/home/x/.dotfiles", r.stdout) + + def test_an_https_remote_to_another_host_is_not_flagged(self): + # GitHub over https is pushable with a credential helper. Flagging it + # would be a guess about a host this machine does not own. + r = run_check(repo_remotes="/home/x/code/thing https://github.com/a/b.git") + self.assertEqual(r.returncode, 0, r.stdout) + + def test_no_repos_is_not_a_finding(self): + r = run_check(repo_remotes="") + self.assertEqual(r.returncode, 0, r.stdout) + + def test_blank_lines_are_ignored(self): + r = run_check(repo_remotes=f"\n\n/home/x/.dotfiles {self.RW}\n\n") + self.assertEqual(r.returncode, 0, r.stdout) + + def test_the_check_always_prints_its_line(self): + for remotes in ("", f"/home/x/.dotfiles {self.RW}", + f"/home/x/.dotfiles {self.RO}"): + with self.subTest(remotes=remotes): + self.assertIn("repo remotes", + run_check(repo_remotes=remotes).stdout.lower()) + + # --- Error cases ---------------------------------------------------- + + def test_a_repo_with_no_origin_is_a_finding(self): + # Fails closed. A repo whose origin could not be read was not checked, + # and reporting it clean is the false pass this script exists to avoid. + r = run_check(repo_remotes="/home/x/.dotfiles") + self.assertEqual(r.returncode, 1) + self.assertIn("/home/x/.dotfiles", r.stdout) + + +class AllClean(unittest.TestCase): + # --- Normal cases --------------------------------------------------- + + def test_all_clean_exits_zero(self): + r = run_check() + self.assertEqual(r.returncode, 0, r.stdout + r.stderr) + + def test_all_clean_prints_one_line_per_check(self): + # The visible line per check is the point of the script: a silent + # no-op is exactly what let the velox gaps sit unseen for two days. + r = run_check() + for label in ("failed units", "unit files", "local files", + "project tooling", "signal"): + self.assertIn(label, r.stdout.lower()) + + def test_all_clean_summary_says_clean(self): + r = run_check() + self.assertIn("all checks clean", r.stdout.lower()) + + def test_all_clean_no_deviation_lines(self): + r = run_check() + self.assertNotIn("DEVIATION", r.stdout) + + +class FailedUnits(unittest.TestCase): + # --- Normal cases --------------------------------------------------- + + def test_failed_user_unit_flags(self): + r = run_check(failed_units="user:calendar-sync.service") + self.assertEqual(r.returncode, 1) + self.assertIn("calendar-sync.service", r.stdout) + self.assertIn("DEVIATION", r.stdout) + + def test_failed_system_unit_flags(self): + r = run_check(failed_units="system:tlp.service") + self.assertEqual(r.returncode, 1) + self.assertIn("tlp.service", r.stdout) + + def test_multiple_failed_units_each_reported(self): + r = run_check( + failed_units="user:calendar-sync.service\nsystem:tlp.service") + self.assertIn("calendar-sync.service", r.stdout) + self.assertIn("tlp.service", r.stdout) + + # --- Boundary cases ------------------------------------------------- + + def test_blank_lines_in_seam_ignored(self): + r = run_check(failed_units="\n\nuser:a.service\n\n") + self.assertEqual(r.returncode, 1) + self.assertIn("a.service", r.stdout) + + +class UnitFilesNotEnabled(unittest.TestCase): + # --- Normal cases --------------------------------------------------- + + def test_disabled_timer_flags(self): + r = run_check(unit_states="roam-sync.timer disabled") + self.assertEqual(r.returncode, 1) + self.assertIn("roam-sync.timer", r.stdout) + + def test_linked_timer_flags(self): + # The exact velox case: a unit symlinked into the user dir by hand, + # never enabled — present, inert, and it LOOKS installed. + r = run_check(unit_states="signal-receive.timer linked") + self.assertEqual(r.returncode, 1) + self.assertIn("signal-receive.timer", r.stdout) + + def test_enabled_timer_passes(self): + r = run_check(unit_states="roam-sync.timer enabled") + self.assertEqual(r.returncode, 0, r.stdout) + + def test_static_service_passes(self): + # A service with no [Install] section is pulled in by its timer; + # "static" is its healthy state, not a gap. + r = run_check(unit_states="roam-sync.service static") + self.assertEqual(r.returncode, 0, r.stdout) + + def test_disabled_service_flags(self): + r = run_check(unit_states="obsbot-wb-guard.service disabled") + self.assertEqual(r.returncode, 1) + self.assertIn("obsbot-wb-guard.service", r.stdout) + + # --- Boundary cases ------------------------------------------------- + + def test_mixed_states_only_inert_reported(self): + r = run_check(unit_states="a.timer enabled\nb.timer disabled\n" + "c.service static\nd.service linked") + self.assertEqual(r.returncode, 1) + self.assertNotIn("a.timer", r.stdout) + self.assertIn("b.timer", r.stdout) + self.assertNotIn("c.service", r.stdout) + self.assertIn("d.service", r.stdout) + + def test_masked_unit_passes(self): + # Masking is a deliberate act (ppd on laptops), not rebuild rot. + r = run_check(unit_states="power-profiles-daemon.service masked") + self.assertEqual(r.returncode, 0, r.stdout) + + def test_service_whose_timer_is_enabled_passes(self): + # A timer-activated service is SUPPOSED to sit linked-not-enabled: + # the timer owns activation, and enabling the service too would run + # it at boot as well. Six of velox's units are this shape, and + # flagging them is the noise that gets a check ignored. + r = run_check(unit_states="roam-sync.service linked\n" + "roam-sync.timer enabled") + self.assertEqual(r.returncode, 0, r.stdout) + + def test_service_whose_timer_is_inert_flags_the_timer_only(self): + # When the timer itself never got enabled, the timer is the finding. + # Naming the service too would double-count one gap. + r = run_check(unit_states="obs-record-watchdog.service linked\n" + "obs-record-watchdog.timer linked") + self.assertEqual(r.returncode, 1) + self.assertIn("obs-record-watchdog.timer", r.stdout) + self.assertNotIn("obs-record-watchdog.service", r.stdout) + + def test_service_without_a_timer_still_flags(self): + # Nothing else can start it, so linked-not-enabled means dead. + r = run_check(unit_states="emacs.service linked") + self.assertEqual(r.returncode, 1) + self.assertIn("emacs.service", r.stdout) + + def test_a_runtime_enabled_timer_suppresses_its_service(self): + # enabled-runtime is a live activation path (enabled until reboot) + # and generated means something produced and installed it, so the + # service beneath either is being started and is not a finding. + for state in ("enabled-runtime", "generated"): + with self.subTest(timer=state): + r = run_check(unit_states=f"foo.service linked\n" + f"foo.timer {state}") + self.assertEqual(r.returncode, 0, + f"a {state} timer failed to suppress") + + def test_an_indirect_timer_does_not_suppress_its_service(self): + # "indirect" means the unit file itself is NOT enabled -- only that + # some Also= relative might be. Under this script's own fail-closed + # rule the uncertain case flags, so suppressing here would be the + # masked blind spot again in a narrower form. + r = run_check(unit_states="foo.service linked\nfoo.timer indirect") + self.assertEqual(r.returncode, 1, + "an indirect timer suppressed a service nothing starts") + self.assertIn("foo.service", r.stdout) + + def test_a_service_whose_timer_cannot_start_it_still_flags(self): + # Suppression is earned by a timer that can actually run the service. + # A masked, static, or absent timer starts nothing, so the service is + # as dead as one with no timer at all -- and suppressing on the mere + # presence of a timer line hides exactly that. + for state in ("masked", "static", "not-found"): + with self.subTest(timer=state): + r = run_check(unit_states=f"foo.service linked\n" + f"foo.timer {state}") + self.assertEqual(r.returncode, 1, + f"a {state} timer suppressed a dead service") + self.assertIn("foo.service", r.stdout) + + +class LocalExampleOrphans(unittest.TestCase): + # --- Normal cases --------------------------------------------------- + + def test_example_without_sibling_flags(self): + with tempfile.TemporaryDirectory() as root: + open(os.path.join(root, "auth.local.el.example"), "w").close() + r = run_check(local_roots=root) + self.assertEqual(r.returncode, 1) + self.assertIn("auth.local.el.example", r.stdout) + + def test_a_real_file_that_is_a_dangling_symlink_still_flags(self): + # A sibling that exists only as a broken link is not a config the + # machine can read, so it is the same gap as an absent one. + with tempfile.TemporaryDirectory() as root: + open(os.path.join(root, "auth.local.el.example"), "w").close() + os.symlink("/nonexistent/stow/target", + os.path.join(root, "auth.local.el")) + r = run_check(local_roots=root) + self.assertEqual(r.returncode, 1, + "a dangling sibling counted as present") + self.assertIn("auth.local.el.example", r.stdout) + + def test_example_with_sibling_passes(self): + with tempfile.TemporaryDirectory() as root: + open(os.path.join(root, "auth.local.el.example"), "w").close() + open(os.path.join(root, "auth.local.el"), "w").close() + r = run_check(local_roots=root) + self.assertEqual(r.returncode, 0, r.stdout) + + def test_nested_example_found(self): + with tempfile.TemporaryDirectory() as root: + sub = os.path.join(root, "modules") + os.makedirs(sub) + open(os.path.join(sub, "mail.local.el.example"), "w").close() + r = run_check(local_roots=root) + self.assertEqual(r.returncode, 1) + self.assertIn("mail.local.el.example", r.stdout) + + # --- Boundary cases ------------------------------------------------- + + def test_two_roots_both_scanned(self): + with tempfile.TemporaryDirectory() as a, \ + tempfile.TemporaryDirectory() as b: + open(os.path.join(a, "one.example"), "w").close() + open(os.path.join(b, "two.example"), "w").close() + r = run_check(local_roots=a + "\n" + b) + self.assertIn("one.example", r.stdout) + self.assertIn("two.example", r.stdout) + + def test_vendored_package_dirs_not_scanned(self): + # elpa/ and friends hold third-party packages that ship their own + # .example docs. Those are the package's business, not this machine's, + # and one of them (dirvish's) was the only finding check 3 produced on + # velox — a standing false positive in front of any real one. + with tempfile.TemporaryDirectory() as root: + for vendor in ("elpa", "node_modules", ".venv", "straight"): + d = os.path.join(root, vendor, "pkg-1.0", "docs") + os.makedirs(d) + open(os.path.join(d, "config.example"), "w").close() + r = run_check(local_roots=root) + self.assertEqual(r.returncode, 0, r.stdout) + + def test_an_unreadable_vendored_dir_is_not_a_finding(self): + # The vendored trees are excluded by design, so failing to descend + # into one is not a gap in what this check covers. Filtering find's + # output without pruning its descent turns a package directory + # nobody wanted read into a standing "could not fully scan". + with tempfile.TemporaryDirectory() as root: + locked = os.path.join(root, "elpa", "pkg-1.0") + os.makedirs(locked) + os.chmod(locked, 0o000) + try: + r = run_check(local_roots=root) + finally: + os.chmod(locked, 0o755) + self.assertEqual(r.returncode, 0, r.stdout) + + def test_git_dir_not_scanned(self): + # .git holds hooks' sample files; those are git's, not the tree's. + with tempfile.TemporaryDirectory() as root: + g = os.path.join(root, ".git", "hooks") + os.makedirs(g) + open(os.path.join(g, "pre-commit.example"), "w").close() + r = run_check(local_roots=root) + self.assertEqual(r.returncode, 0, r.stdout) + + # --- Error cases ---------------------------------------------------- + + def test_an_unreadable_subdirectory_is_a_finding_not_a_pass(self): + # find exits non-zero when it cannot descend somewhere, and prints + # what it did reach. Discarding that status hides every orphan under + # the unreadable directory behind a clean "ok" -- the same defect + # class as a probe that cannot run reading as a pass. + with tempfile.TemporaryDirectory() as root: + locked = os.path.join(root, "locked") + os.makedirs(locked) + open(os.path.join(locked, "auth.local.el.example"), "w").close() + os.chmod(locked, 0o000) + try: + r = run_check(local_roots=root) + finally: + os.chmod(locked, 0o755) + self.assertEqual(r.returncode, 1, + "an unreadable directory read as nothing to check") + self.assertIn("could not", r.stdout.lower()) + + def test_missing_root_is_its_own_finding(self): + # A scan root that's gone is a rebuild gap too, not a pass. + r = run_check(local_roots="/nonexistent/scan-root") + self.assertEqual(r.returncode, 1) + self.assertIn("/nonexistent/scan-root", r.stdout) + + def test_a_path_with_spaces_is_one_root_not_three(self): + # Roots arrive newline-separated for this reason: splitting on spaces + # turns one real directory into several imaginary missing ones. + with tempfile.TemporaryDirectory() as base: + root = os.path.join(base, "a dir with spaces") + os.makedirs(root) + open(os.path.join(root, "orphan.example"), "w").close() + r = run_check(local_roots=root) + self.assertEqual(r.stdout.count("DEVIATION"), 1, r.stdout) + self.assertIn("orphan.example", r.stdout) + + +class ProjectTooling(unittest.TestCase): + def project(self, root, gitignore_lines, present=()): + os.makedirs(os.path.join(root, ".git")) + with open(os.path.join(root, ".gitignore"), "w") as f: + f.write("\n".join(gitignore_lines) + "\n") + for p in present: + path = os.path.join(root, p) + if p.endswith("/"): + os.makedirs(path, exist_ok=True) + else: + open(path, "w").close() + + # --- Normal cases --------------------------------------------------- + + def test_ignored_but_absent_tooling_flags(self): + with tempfile.TemporaryDirectory() as root: + self.project(root, [".ai/", "todo.org"]) + r = run_check(project_roots=root) + self.assertEqual(r.returncode, 1) + for missing in (".ai", "todo.org"): + self.assertIn(missing, r.stdout) + + def test_claude_dir_absence_never_flags(self): + # Same shape as CLAUDE.md below, and proven the same way. The bootstrap + # and the gitignore sweep write `.claude/` into the ignore set of every + # gitignore-mode project whether or not one ever exists there, so the + # entry is aspirational rather than a promise. Three projects tripped + # this on velox, and ratio is missing the identical directory in the + # identical three, which is what proves it is the steady state and not + # reinstall drift. + # + # Nor does dropping it lose a real signal. A project that genuinely + # carries one (rules and hooks from a language bundle) has it re-synced + # by sync-language-bundle.sh at every session start, so a true absence + # heals itself before this check would ever run. + with tempfile.TemporaryDirectory() as root: + self.project(root, [".ai/", ".claude/"], present=(".ai/",)) + r = run_check(project_roots=root) + self.assertEqual(r.returncode, 0, r.stdout) + self.assertNotIn(".claude", r.stdout) + + def test_ignored_and_present_tooling_passes(self): + with tempfile.TemporaryDirectory() as root: + self.project(root, [".ai/", "CLAUDE.md"], + present=(".ai/", "CLAUDE.md")) + r = run_check(project_roots=root) + self.assertEqual(r.returncode, 0, r.stdout) + + def test_claude_md_absence_never_flags(self): + # CLAUDE.md is seed-only: install-lang writes it once and the project + # owns it afterward, so most projects legitimately never have one. + # Ratio shows the identical absences in the identical projects, which + # is what proves it is the steady state and not reinstall drift. + with tempfile.TemporaryDirectory() as root: + self.project(root, [".ai/", "CLAUDE.md"], present=(".ai/",)) + r = run_check(project_roots=root) + self.assertEqual(r.returncode, 0, r.stdout) + + def test_missing_ai_dir_still_flags(self): + # The one that carries real working state — 374 files in .emacs.d's + # case — and that nothing restores: not git, not stow, not bootstrap. + with tempfile.TemporaryDirectory() as root: + self.project(root, [".ai/", "CLAUDE.md"]) + r = run_check(project_roots=root) + self.assertEqual(r.returncode, 1) + self.assertIn(".ai", r.stdout) + self.assertNotIn("CLAUDE.md", r.stdout) + + def test_unignored_tooling_never_expected(self): + # A project that never gitignored todo.org never had one to lose; + # the project's own .gitignore is the record of what it should hold. + with tempfile.TemporaryDirectory() as root: + self.project(root, [".ai/"], present=(".ai/",)) + r = run_check(project_roots=root) + self.assertEqual(r.returncode, 0, r.stdout) + + # --- Boundary cases ------------------------------------------------- + + def test_anchored_ignore_style_recognized(self): + # Both /.ai/ (anchored) and .ai/ (unanchored) styles exist across + # the fleet; the sweep-gitignore audit hit exactly this split. + with tempfile.TemporaryDirectory() as root: + self.project(root, ["/.ai/", "/CLAUDE.md"]) + r = run_check(project_roots=root) + self.assertEqual(r.returncode, 1) + self.assertIn(".ai", r.stdout) + + def test_a_worktree_or_submodule_is_still_a_project(self): + # In a worktree or submodule, .git is a file pointing at the real + # gitdir rather than a directory, so a -d test skips the project + # silently. + with tempfile.TemporaryDirectory() as root: + with open(os.path.join(root, ".git"), "w") as f: + f.write("gitdir: /somewhere/else\n") + with open(os.path.join(root, ".gitignore"), "w") as f: + f.write(".ai/\n") + r = run_check(project_roots=root) + self.assertEqual(r.returncode, 1) + self.assertIn(".ai", r.stdout) + + def test_an_unreadable_gitignore_is_a_finding_not_a_pass(self): + # grep exits 2 on error and 1 on no-match, so treating both as + # "nothing named" lets an unreadable ignore file pass the project + # silently. + with tempfile.TemporaryDirectory() as root: + self.project(root, [".ai/"]) + os.chmod(os.path.join(root, ".gitignore"), 0o000) + try: + r = run_check(project_roots=root) + finally: + os.chmod(os.path.join(root, ".gitignore"), 0o644) + self.assertEqual(r.returncode, 1, + "an unreadable .gitignore read as nothing to check") + self.assertIn("could not", r.stdout.lower()) + + def test_non_git_dir_skipped(self): + with tempfile.TemporaryDirectory() as root: + with open(os.path.join(root, ".gitignore"), "w") as f: + f.write(".ai/\n") + r = run_check(project_roots=root) + self.assertEqual(r.returncode, 0, r.stdout) + + def test_project_without_gitignore_skipped(self): + with tempfile.TemporaryDirectory() as root: + os.makedirs(os.path.join(root, ".git")) + r = run_check(project_roots=root) + self.assertEqual(r.returncode, 0, r.stdout) + + def test_unrelated_ignore_lines_no_flags(self): + with tempfile.TemporaryDirectory() as root: + self.project(root, ["*.pyc", "node_modules/", "dist/"]) + r = run_check(project_roots=root) + self.assertEqual(r.returncode, 0, r.stdout) + + +class SignalAccount(unittest.TestCase): + # --- Normal cases --------------------------------------------------- + + def test_registered_account_passes(self): + r = run_check(signal_accounts="Number: +15045551234 ...") + self.assertEqual(r.returncode, 0, r.stdout) + + def test_no_account_flags(self): + # The velox case: a wiped registration silently breaks paging for + # the whole fleet, because agent-text relays into this machine. + r = run_check(signal_accounts="") + self.assertEqual(r.returncode, 1) + self.assertIn("signal", r.stdout.lower()) + + # --- Error cases ---------------------------------------------------- + + def test_missing_binary_flags(self): + r = run_check(signal_accounts="MISSING") + self.assertEqual(r.returncode, 1) + self.assertIn("signal-cli", r.stdout) + + def test_a_stray_signal_missing_in_the_environment_is_ignored(self): + # The script's own internal flag must not be settable from outside, + # or a caller's unrelated variable turns a registered account into a + # "signal-cli is not installed" finding. + env = dict(os.environ) + env.update({"PRC_FAILED_UNITS": "", "PRC_UNIT_STATES": "", + "PRC_LOCAL_SCAN_ROOTS": "", "PRC_PROJECT_ROOTS": "", + "PRC_NTP_SOURCES": "162.159.200.1", + "PRC_SIGNAL_ACCOUNTS": "+15045551234", + "PRC_IDLE_DAEMON": "4242", + "PRC_REPO_REMOTES": "", + "PRC_UNITS_EXPECTED_DISABLED": "", + "signal_missing": "1"}) + r = subprocess.run(["sh", CHECK], capture_output=True, text=True, + timeout=30, env=env) + self.assertEqual(r.returncode, 0, r.stdout) + + def test_a_stray_idle_absent_in_the_environment_is_ignored(self): + # Same class as the flag above, and the dangerous direction: an + # inherited idle_absent=1 would make check 7 skip silently and report + # ok on a machine that cannot sleep, which is the exact false pass the + # check exists to prevent. + env = dict(os.environ) + env.update({"PRC_FAILED_UNITS": "", "PRC_UNIT_STATES": "", + "PRC_LOCAL_SCAN_ROOTS": "", "PRC_PROJECT_ROOTS": "", + "PRC_NTP_SOURCES": "162.159.200.1", + "PRC_SIGNAL_ACCOUNTS": "+15045551234", + "PRC_IDLE_DAEMON": "", + "idle_absent": "1"}) + r = subprocess.run(["sh", CHECK], capture_output=True, text=True, + timeout=30, env=env) + self.assertEqual(r.returncode, 1, r.stdout) + self.assertIn("hypridle is installed but not running", r.stdout) + + +class ProbeFailure(unittest.TestCase): + """A probe that could not run must never read as a clean check. + + This is the defect the whole script exists to catch, so it would be the + worst possible place to have it. `systemctl --user` exits 1 with empty + output when there is no user bus -- over ssh, from cron, under sudo, or on + a TTY before the graphical session starts. Reading that as "no failed + units" reports a machine as healthy precisely when nothing can be checked. + """ + + def unset(self, *names): + """Run with the named seams unset, so the real probes execute.""" + env = dict(os.environ) + env.update({"PRC_FAILED_UNITS": "", "PRC_UNIT_STATES": "", + "PRC_LOCAL_SCAN_ROOTS": "", "PRC_PROJECT_ROOTS": "", + "PRC_NTP_SOURCES": "162.159.200.1", + "PRC_SIGNAL_ACCOUNTS": "+15045551234"}) + for n in names: + env.pop(n, None) + env["XDG_RUNTIME_DIR"] = "/nonexistent-runtime-dir" + return subprocess.run(["sh", CHECK], capture_output=True, text=True, + timeout=30, env=env) + + # --- Error cases ---------------------------------------------------- + + def test_unreachable_user_bus_is_a_finding_not_a_pass(self): + r = self.unset("PRC_FAILED_UNITS") + self.assertEqual(r.returncode, 1, + "a failed probe reported the machine as clean") + self.assertIn("could not", r.stdout.lower()) + + def test_unreachable_user_bus_fails_the_unit_state_check_too(self): + r = self.unset("PRC_UNIT_STATES") + self.assertEqual(r.returncode, 1, + "a failed probe reported the machine as clean") + + def test_an_unusable_tmpdir_is_a_finding_not_a_pass(self): + # Every check stages its input through a temp file. If that write + # fails, each loop reads nothing and every check comes back clean -- + # with real findings passed in. + env = dict(os.environ) + env.update({"PRC_FAILED_UNITS": "user:calendar-sync.service", + "PRC_UNIT_STATES": "", "PRC_LOCAL_SCAN_ROOTS": "", + "PRC_PROJECT_ROOTS": "", + "PRC_SIGNAL_ACCOUNTS": "+15045551234", + "PRC_IDLE_DAEMON": "4242", + "PRC_REPO_REMOTES": "", + "PRC_UNITS_EXPECTED_DISABLED": "", + "TMPDIR": "/nonexistent-tmp-dir"}) + r = subprocess.run(["sh", CHECK], capture_output=True, text=True, + timeout=30, env=env) + self.assertEqual(r.returncode, 1, + "an unwritable TMPDIR swallowed a real finding") + + +class RealUnitDirEnumeration(unittest.TestCase): + """Check 2's unseamed path, where the unit files are read off disk. + + The PRC_UNIT_STATES seam skips this enumeration entirely, so a defect in + it survives every seamed test. That is where the dangling-stow-link case + lives, and a dangling stow link is precisely the requirement's headline + example of a unit file that LOOKED fine. + """ + + def run_real(self, config_home): + env = dict(os.environ) + env.update({"PRC_FAILED_UNITS": "", "PRC_LOCAL_SCAN_ROOTS": "", + "PRC_PROJECT_ROOTS": "", + "PRC_SIGNAL_ACCOUNTS": "+15045551234", + "PRC_IDLE_DAEMON": "4242", + "PRC_REPO_REMOTES": "", + "PRC_UNITS_EXPECTED_DISABLED": "", + "XDG_CONFIG_HOME": config_home}) + env.pop("PRC_UNIT_STATES", None) + return subprocess.run(["sh", CHECK], capture_output=True, text=True, + timeout=30, env=env) + + # --- Boundary cases ------------------------------------------------- + + def test_a_dangling_stow_link_is_enumerated_not_skipped(self): + with tempfile.TemporaryDirectory() as home: + unit_dir = os.path.join(home, "systemd", "user") + os.makedirs(unit_dir) + os.symlink("/nonexistent/stow/roam-sync.timer", + os.path.join(unit_dir, "roam-sync.timer")) + r = self.run_real(home) + self.assertEqual(r.returncode, 1, + "a dangling stow link read as nothing to check") + self.assertIn("roam-sync.timer", r.stdout) + self.assertIn("missing target", r.stdout) + + # --- Error cases ---------------------------------------------------- + + def test_a_missing_unit_directory_is_a_finding(self): + with tempfile.TemporaryDirectory() as home: + r = self.run_real(home) + self.assertEqual(r.returncode, 1) + self.assertIn("no user unit directory", r.stdout) + + +class WedgedSystemctl(unittest.TestCase): + """A systemd manager that never answers must not hang the check. + + Seen live on velox 2026-08-17: the user manager spun at 96% CPU with + `is-enabled`, `cat`, and `list-unit-files` all hanging while `list-units` + still returned. Unbounded, the check stops at the first unit and never + runs checks 3 through 5, so the machine most in need of checking is the + one it reports nothing about. + """ + + def run_with_fake(self, script_body, timeout_s="1"): + """Run against a fake systemctl, with check 2's unit dir empty. + + Pointing XDG_CONFIG_HOME at an empty directory keeps check 2 from + making one call per real unit, so the test measures the bound rather + than the size of this machine's unit directory. + """ + with tempfile.TemporaryDirectory() as d: + fake = os.path.join(d, "systemctl") + with open(fake, "w") as f: + f.write(script_body) + os.chmod(fake, 0o755) + env = dict(os.environ) + env.update({"PRC_LOCAL_SCAN_ROOTS": "", "PRC_PROJECT_ROOTS": "", + "PRC_SIGNAL_ACCOUNTS": "+15045551234", + "PRC_IDLE_DAEMON": "4242", + "PRC_REPO_REMOTES": "", + "PRC_UNITS_EXPECTED_DISABLED": "", + "PRC_SYSTEMCTL": fake, + "PRC_SYSTEMCTL_TIMEOUT": timeout_s, + "XDG_CONFIG_HOME": d}) + env.pop("PRC_FAILED_UNITS", None) + env.pop("PRC_UNIT_STATES", None) + start = time.monotonic() + r = subprocess.run(["sh", CHECK], capture_output=True, + text=True, timeout=60, env=env) + return r, time.monotonic() - start + + # --- Error cases ---------------------------------------------------- + + def test_a_hanging_systemctl_is_bounded_and_reported(self): + r, _ = self.run_with_fake("#!/bin/sh\nsleep 300\n") + self.assertEqual(r.returncode, 1) + self.assertIn("could not query user units", r.stdout) + # The run must reach the end rather than stopping at the first call. + self.assertIn("check 8/8", r.stdout) + + def test_a_hanging_systemctl_does_not_stall_the_whole_run(self): + # The fake sleeps 8s against a 1s bound, so a bounded run lands near + # 2s (two calls) and an unbounded one near 16s. Deliberately short + # enough that losing the bound fails this assertion in seconds rather + # than hitting the subprocess ceiling a minute later -- a regression + # nobody waits out is a regression nobody catches. + _, elapsed = self.run_with_fake("#!/bin/sh\nsleep 8\n") + self.assertLess(elapsed, 6, + "the run was not bounded by PRC_SYSTEMCTL_TIMEOUT") + + +class Reporting(unittest.TestCase): + # --- Normal cases --------------------------------------------------- + + def test_findings_counted_in_summary(self): + # Assert the count in the summary line specifically. A bare + # assertIn("2") passes on the always-present "check 2/5" text, so it + # stays green even when the counter is arithmetically wrong. + r = run_check(failed_units="user:a.service\nuser:b.service", + unit_states="c.timer disabled") + self.assertEqual(r.returncode, 1) + summary = r.stdout.strip().splitlines()[-1] + self.assertEqual(summary, "3 finding(s) across 8 checks") + + def test_the_summary_count_tracks_every_check(self): + # One finding from each of the five, so a counter that drops or + # double-counts any single check shows up here. + with tempfile.TemporaryDirectory() as scan, \ + tempfile.TemporaryDirectory() as proj: + open(os.path.join(scan, "orphan.example"), "w").close() + os.makedirs(os.path.join(proj, ".git")) + with open(os.path.join(proj, ".gitignore"), "w") as f: + f.write(".ai/\n") + r = run_check(failed_units="user:a.service", + unit_states="b.timer disabled", + local_roots=scan, project_roots=proj, + signal_accounts="") + summary = r.stdout.strip().splitlines()[-1] + self.assertEqual(summary, "5 finding(s) across 8 checks") + + def test_help_exits_zero(self): + r = subprocess.run(["sh", CHECK, "--help"], + capture_output=True, text=True, timeout=10) + self.assertEqual(r.returncode, 0) + self.assertIn("post-rebuild-check", r.stdout) + + # --- Error cases ---------------------------------------------------- + + def test_unknown_flag_errors(self): + r = subprocess.run(["sh", CHECK, "--bogus"], + capture_output=True, text=True, timeout=10) + self.assertNotEqual(r.returncode, 0) + + +if __name__ == "__main__": + unittest.main() |