diff options
Diffstat (limited to 'tests/post-rebuild-check/test_post_rebuild_check.py')
| -rw-r--r-- | tests/post-rebuild-check/test_post_rebuild_check.py | 1126 |
1 files changed, 1126 insertions, 0 deletions
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..a034f87 --- /dev/null +++ b/tests/post-rebuild-check/test_post_rebuild_check.py @@ -0,0 +1,1126 @@ +"""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/", ".claude/", "todo.org"]) + r = run_check(project_roots=root) + self.assertEqual(r.returncode, 1) + for missing in (".ai", ".claude", "todo.org"): + self.assertIn(missing, 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: +15045173983 ...") + 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() |