"""Hygiene and flatpak tests. Both modules are built to degrade rather than fail: a missing tool, an unreadable directory or an unreachable remote is a reportable state, not an exception. Most of these tests are about that. """ from __future__ import annotations import os import sys import tempfile import time from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) from updater import flatpak, hygiene # noqa: E402 # -------------------------------------------------------------------------- # cache parsing # -------------------------------------------------------------------------- def test_cache_name_parsing_handles_dashes_in_package_names(): assert hygiene.parse_cache_name("linux-6.16.7.arch1-1-x86_64.pkg.tar.zst") == ( "linux", "6.16.7.arch1-1", ) assert hygiene.parse_cache_name( "python-opentelemetry-instrumentation-requests-1.0.0-1-any.pkg.tar.zst" ) == ("python-opentelemetry-instrumentation-requests", "1.0.0-1") assert hygiene.parse_cache_name("foo-1.0-1-x86_64.pkg.tar.xz") == ("foo", "1.0-1") def test_cache_name_parsing_rejects_things_that_are_not_packages(): assert hygiene.parse_cache_name("README") is None assert hygiene.parse_cache_name("random-file.txt") is None assert hygiene.parse_cache_name("only-two-1.pkg.tar.zst") is None def _cache_fixture(tmp: Path, spec: dict[str, list[str]]) -> None: for name, versions in spec.items(): for index, version in enumerate(versions): path = tmp / f"{name}-{version}-x86_64.pkg.tar.zst" path.write_bytes(b"x" * 100) # newest last, so mtime ordering is unambiguous stamp = time.time() - (len(versions) - index) * 3600 os.utime(path, (stamp, stamp)) def test_cache_keeps_the_newest_versions_and_reports_the_rest(): with tempfile.TemporaryDirectory() as td: cache = Path(td) _cache_fixture(cache, {"linux": ["6.16.1-1", "6.16.2-1", "6.16.3-1", "6.16.4-1"]}) usage = hygiene.cache_usage(cache, keep=2) assert usage.files == 4 assert usage.reclaimable_files == 2, usage assert usage.reclaimable == 200 assert usage.total == 400 def test_cache_always_keeps_the_installed_version_even_if_it_is_old(): """Downgrading is legitimate, so the installed file must survive a prune.""" with tempfile.TemporaryDirectory() as td: cache = Path(td) _cache_fixture(cache, {"linux": ["6.16.1-1", "6.16.2-1", "6.16.3-1"]}) usage = hygiene.cache_usage(cache, keep=1, installed={"linux": "6.16.1-1"}) # newest (6.16.3) plus the installed (6.16.1) are both kept assert usage.reclaimable_files == 1 assert usage.reclaimable == 100 def test_cache_usage_on_a_missing_directory_is_silent(): usage = hygiene.cache_usage(Path("/definitely/not/here")) assert usage.total == 0 and not usage.has_reclaimable def test_empty_cache_reports_nothing_reclaimable(): with tempfile.TemporaryDirectory() as td: usage = hygiene.cache_usage(Path(td)) assert usage.files == 0 and not usage.has_reclaimable # -------------------------------------------------------------------------- # config drift # -------------------------------------------------------------------------- def test_config_drift_finds_each_suffix(): with tempfile.TemporaryDirectory() as td: root = Path(td) (root / "sub").mkdir() (root / "pacman.conf.pacnew").write_text("new") (root / "sub" / "mkinitcpio.conf.pacnew").write_text("new") (root / "old.conf.pacsave").write_text("old") (root / "orig.conf.pacorig").write_text("orig") (root / "normal.conf").write_text("fine") drift = hygiene.config_drift(root) assert len(drift.pacnew) == 2 assert len(drift.pacsave) == 1 assert len(drift.pacorig) == 1 assert drift.total == 4 # only .pacnew needs merging; the others are informational assert set(drift.actionable) == set(drift.pacnew) assert all(p.name.endswith(".pacnew") for p in drift.actionable) def test_config_drift_on_a_missing_root_is_silent(): assert hygiene.config_drift(Path("/definitely/not/here")).total == 0 def test_config_drift_respects_the_depth_limit(): with tempfile.TemporaryDirectory() as td: root = Path(td) deep = root for i in range(8): deep = deep / f"d{i}" deep.mkdir(parents=True) (deep / "buried.conf.pacnew").write_text("x") assert hygiene.config_drift(root, max_depth=3).total == 0 assert hygiene.config_drift(root, max_depth=10).total == 1 # -------------------------------------------------------------------------- # reboot reasoning # -------------------------------------------------------------------------- def test_reboot_is_advised_when_the_running_kernel_vanished(): with tempfile.TemporaryDirectory() as td: modules = Path(td) (modules / "6.16.9-arch1-1").mkdir() reason = hygiene.reboot_reason(modules=modules) if hygiene.running_kernel() != "6.16.9-arch1-1": assert reason and "no longer installed" in reason else: assert reason is None def test_reboot_is_not_advised_when_the_running_kernel_is_present(): with tempfile.TemporaryDirectory() as td: modules = Path(td) (modules / hygiene.running_kernel()).mkdir() assert hygiene.reboot_reason(modules=modules) is None def test_updated_kernel_packages_trigger_a_reboot_recommendation(): with tempfile.TemporaryDirectory() as td: modules = Path(td) (modules / hygiene.running_kernel()).mkdir() assert hygiene.reboot_reason({"vim", "python"}, modules=modules) is None reason = hygiene.reboot_reason({"linux", "vim"}, modules=modules) assert reason and "linux" in reason assert hygiene.reboot_reason({"nvidia-utils"}, modules=modules) def test_missing_module_tree_does_not_produce_a_false_reboot(): """On a machine without /usr/lib/modules we cannot tell, so say nothing.""" assert hygiene.reboot_reason(modules=Path("/definitely/not/here")) is None # -------------------------------------------------------------------------- # stale processes # -------------------------------------------------------------------------- def test_stale_process_detection_runs_without_privileges(): stale = hygiene.stale_processes() assert isinstance(stale, list) for entry in stale: assert entry.pid > 0 assert entry.path.startswith("/") assert entry.name def test_stale_process_scan_is_bounded(): started = time.monotonic() hygiene.stale_processes(limit=5) assert time.monotonic() - started < 5.0 # -------------------------------------------------------------------------- # flatpak # -------------------------------------------------------------------------- def test_flatpak_parsing_handles_the_columns_format(): text = ( "org.gnome.Calculator\tstable\tflathub\t46.0\n" "org.mozilla.firefox\tstable\tflathub\t128.0\n" "\n" ) refs = flatpak._parse(text, kind="app") assert len(refs) == 2 assert refs[0].application == "org.gnome.Calculator" assert refs[0].version == "46.0" assert refs[0].origin == "flathub" assert refs[0].label == "org.gnome.Calculator 46.0" def test_flatpak_parsing_survives_missing_columns(): """Older flatpak builds may return fewer columns than requested.""" refs = flatpak._parse("org.example.App\tstable\n", kind="app") assert len(refs) == 1 assert refs[0].branch == "stable" assert refs[0].version == "" def test_flatpak_parsing_ignores_blank_and_headerless_lines(): assert flatpak._parse("\n \n", kind="app") == [] def test_flatpak_failure_is_detected_despite_a_zero_exit_code(): """flatpak prints `error:` on stderr and may still exit 0.""" assert flatpak._looks_failed(0, "") == "" assert "error" in flatpak._looks_failed(0, "error: Unable to load summary").lower() assert flatpak._looks_failed(1, "") != "" assert flatpak._looks_failed(1, "boom") == "boom" def test_flatpak_refs_are_deduped_and_ordered(): refs = [ flatpak.FlatpakRef("b.App", kind="app"), flatpak.FlatpakRef("a.App", kind="app"), flatpak.FlatpakRef("b.App", kind="app"), flatpak.FlatpakRef("runtime.Runtime", kind="runtime"), ] got = flatpak._dedupe(refs) assert [r.application for r in got] == ["a.App", "b.App", "runtime.Runtime"] def test_flatpak_collect_never_raises_even_when_broken(): """Absent, offline or cache-locked flatpak is a reportable state.""" state = flatpak.collect() assert isinstance(state, flatpak.FlatpakState) assert isinstance(state.available, bool) if not state.available: assert state.reason, "an unavailable flatpak must explain itself" def test_flatpak_short_reason_is_usable_in_a_one_line_row(): state = flatpak.FlatpakState( reason="error: Unable to load summary from remote flathub: " 'Error opening file "/home/x/.cache/flatpak/summaries/flathub.idx": ' "Read-only file system" ) short = state.short_reason assert short == "remote flathub unreachable", short assert len(short) <= 48 assert "/home" not in short def test_flatpak_short_reason_handles_other_failures(): assert ( flatpak.FlatpakState(reason="flatpak is not installed").short_reason == "flatpak is not installed" ) readonly = flatpak.FlatpakState(reason="Error opening file: Read-only file system") assert "read-only" in readonly.short_reason assert flatpak.FlatpakState(reason="").short_reason == "" def test_flatpak_update_command_is_noninteractive(): argv = flatpak.update_command() assert argv[:2] == ["flatpak", "update"] assert "-y" in argv and "--noninteractive" in argv # -------------------------------------------------------------------------- # the aggregate # -------------------------------------------------------------------------- def test_hygiene_collect_gathers_everything_without_raising(): report = hygiene.collect(include_processes=False) assert isinstance(report.cache, hygiene.CacheUsage) assert isinstance(report.drift, hygiene.Drift) assert isinstance(report.orphans, list) assert report.stale == [] def test_hygiene_collect_can_be_pointed_at_fixture_directories(): with tempfile.TemporaryDirectory() as td: root = Path(td) cache, etc = root / "cache", root / "etc" cache.mkdir() etc.mkdir() _cache_fixture(cache, {"foo": ["1-1", "2-1", "3-1", "4-1"]}) (etc / "x.conf.pacnew").write_text("x") # no `installed` mapping here: pinning the oldest version as installed # would legitimately keep every file, which is not what this test is about report = hygiene.collect( cache_dir=cache, etc=etc, orphans=["orphan"], include_processes=False ) assert report.cache.reclaimable_files == 1, report.cache assert len(report.drift.pacnew) == 1 assert report.orphans == ["orphan"] assert report.has_anything def test_an_empty_hygiene_report_says_it_has_nothing(): with tempfile.TemporaryDirectory() as td: root = Path(td) cache, etc = root / "cache", root / "etc" modules = root / "modules" for d in (cache, etc, modules): d.mkdir() # an empty machine, including its module tree: the running kernel # check reads the real one otherwise, and a host that has just # installed a kernel reports a pending reboot no matter what the # caller passed for the other paths. report = hygiene.collect(cache_dir=cache, etc=etc, modules=modules, include_processes=False) assert not report.has_anything, report.reboot def test_hygiene_checks_offer_a_command_to_run(): """A check that says "run paccache" should be able to run it.""" from updater import cli from updater.render import review row = review.CheckRow(False, "cache", "948 MiB reclaimable", action=["paccache", "-rk3"]) assert row.action[0] == "paccache" # the default is no action, so nothing runs by accident assert review.CheckRow(True, "pacnew", "clean").action == [] assert cli is not None def test_a_destructive_action_carries_a_warning(): """Removing orphans is the one hygiene action that can break a system.""" from updater.render import review row = review.CheckRow(False, "orphans", "12 packages", warning="removes packages") assert row.warning def test_leftovers_explain_includes_the_command_that_clears_them(): import tempfile directory = Path(tempfile.mkdtemp()) for i in range(12): (directory / f"download-{i}").write_bytes(b"") got = hygiene.incomplete_downloads(directory) text = "\n".join(got.explain(directory)) assert "find" in text and "-delete" in text, text assert "zero bytes" in text, "the empty-file clue is the useful part" def test_leftovers_do_not_claim_a_write_failure_when_data_is_present(): """Partial data means the transfer failed, which is a different story.""" import tempfile directory = Path(tempfile.mkdtemp()) for i in range(12): (directory / f"download-{i}").write_bytes(b"x" * 10) got = hygiene.incomplete_downloads(directory) assert got.bytes > 0 assert not got.all_empty assert "zero bytes" not in "\n".join(got.explain(directory)) def main() -> int: tests = [(n, f) for n, f in sorted(globals().items()) if n.startswith("test_") and callable(f)] failed = 0 for name, fn in tests: try: fn() except Exception as exc: # noqa: BLE001 failed += 1 print(f"FAIL {name}\n {type(exc).__name__}: {exc}") else: print(f"ok {name}") print(f"\n{len(tests) - failed}/{len(tests)} passed") return 1 if failed else 0 # -------------------------------------------------------------------------- # abandoned downloads # -------------------------------------------------------------------------- def test_incomplete_downloads_counts_scratch_files(): import tempfile as tf with tf.TemporaryDirectory() as td: d = Path(td) for i in range(4): (d / f"download-abc{i}").write_bytes(b"") (d / "download-big").write_bytes(b"x" * 100) # a real package and a signature must not be counted (d / "linux-6.16-1-x86_64.pkg.tar.zst").write_bytes(b"x" * 10) (d / "linux-6.16-1-x86_64.pkg.tar.zst.sig").write_bytes(b"x" * 10) got = hygiene.incomplete_downloads(d) assert got.count == 5, got assert got.bytes == 100, got def test_a_missing_cache_directory_is_not_an_error(): got = hygiene.incomplete_downloads(Path("/definitely/not/here")) assert got.count == 0 and got.bytes == 0 def test_a_handful_of_failures_is_not_worth_mentioning(): """Occasional failed downloads are normal; a pile of them is a symptom.""" assert not hygiene.Leftovers(count=1).worth_mentioning assert not hygiene.Leftovers(count=9).worth_mentioning assert hygiene.Leftovers(count=10).worth_mentioning assert hygiene.Leftovers(count=445).worth_mentioning def test_the_oldest_leftover_is_tracked(): import os as osmod import tempfile as tf import time as timemod with tf.TemporaryDirectory() as td: d = Path(td) old = d / "download-old" new = d / "download-new" old.write_bytes(b"") new.write_bytes(b"") past = timemod.time() - 86400 * 30 osmod.utime(old, (past, past)) got = hygiene.incomplete_downloads(d) assert got.count == 2 assert abs(got.oldest - past) < 5, got.oldest def test_the_real_cache_is_reported_without_raising(): """Runs against whatever is actually on this machine.""" got = hygiene.incomplete_downloads() assert got.count >= 0 and got.bytes >= 0 if got.count: assert got.oldest > 0, "a counted leftover must carry an mtime" def test_leftovers_reach_the_hygiene_report(): report = hygiene.collect(include_processes=False) assert isinstance(report.leftovers, hygiene.Leftovers) if report.leftovers.worth_mentioning: assert report.has_anything, "a notable leftover must make the report non-empty" # -------------------------------------------------------------------------- # processes running a replaced binary # -------------------------------------------------------------------------- def _listed(pid: int) -> bool: return pid in {s.pid for s in hygiene.stale_processes()} def test_a_process_running_a_replaced_binary_is_reported(): """The case the check exists for, and the one it used to skip. Replacing a package's files unlinks the old ones, so the kernel marks `/proc//exe` with ` (deleted)`. The check discarded exactly those, so it reported almost nothing. """ import shutil import subprocess import tempfile directory = Path(tempfile.mkdtemp()) binary = directory / "sleeper" shutil.copy("/usr/bin/sleep", binary) proc = subprocess.Popen([str(binary), "30"]) try: import time as timemod timemod.sleep(0.3) assert not _listed(proc.pid), "a fresh process should not be stale" os.remove(binary) shutil.copy("/usr/bin/sleep", binary) # a new inode at the same path timemod.sleep(0.2) hits = [s for s in hygiene.stale_processes() if s.pid == proc.pid] assert hits, "a replaced binary was not reported as stale" assert " (deleted)" not in hits[0].path, hits[0].path assert hits[0].path == str(binary) finally: proc.kill() proc.wait() shutil.rmtree(directory, ignore_errors=True) def test_a_process_whose_binary_was_deleted_is_reported(): import shutil import subprocess import tempfile import time as timemod directory = Path(tempfile.mkdtemp()) binary = directory / "sleeper" shutil.copy("/usr/bin/sleep", binary) proc = subprocess.Popen([str(binary), "30"]) try: timemod.sleep(0.3) os.remove(binary) timemod.sleep(0.2) assert _listed(proc.pid), "a deleted binary is the clearest staleness" finally: proc.kill() proc.wait() shutil.rmtree(directory, ignore_errors=True) def test_a_current_process_is_not_reported(): import subprocess import time as timemod proc = subprocess.Popen(["sleep", "30"]) try: timemod.sleep(0.3) assert not _listed(proc.pid) finally: proc.kill() proc.wait() def test_stale_processes_needs_no_privileges_and_does_not_raise(): got = hygiene.stale_processes() assert isinstance(got, list) for entry in got: assert entry.pid > 0 assert entry.path.startswith("/"), entry.path assert entry.name, "a reported process must have a name" def test_a_tiny_limit_is_respected(): """The scan is bounded so a machine with thousands of processes stays quick.""" import time as timemod started = timemod.monotonic() hygiene.stale_processes(limit=1) assert timemod.monotonic() - started < 5 # -------------------------------------------------------------------------- # configuration drift # -------------------------------------------------------------------------- def test_config_drift_finds_every_suffix_and_nothing_else(): import tempfile base = Path(tempfile.mkdtemp()) etc = base / "etc" etc.mkdir() (etc / "pacman.conf.pacnew").write_text("x") (etc / "pacman.conf.pacsave").write_text("x") (etc / "hosts.pacorig").write_text("x") (etc / "sub").mkdir() (etc / "sub" / "nested.conf.pacnew").write_text("x") # things that must be ignored (etc / "notapacnew").write_text("x") (etc / "weird.pacnew.bak").write_text("x") (etc / ".git").mkdir() (etc / ".git" / "x.pacnew").write_text("x") (etc / "node_modules").mkdir() (etc / "node_modules" / "y.pacnew").write_text("x") got = hygiene.config_drift(etc) names = lambda bucket: sorted(p.name for p in bucket) # noqa: E731 assert names(got.pacnew) == ["nested.conf.pacnew", "pacman.conf.pacnew"], names(got.pacnew) assert names(got.pacsave) == ["pacman.conf.pacsave"] assert names(got.pacorig) == ["hosts.pacorig"] def test_config_drift_respects_max_depth(): import tempfile base = Path(tempfile.mkdtemp()) etc = base / "etc" etc.mkdir() deep = etc for i in range(9): deep = deep / f"l{i}" deep.mkdir() (deep / "too-deep.conf.pacnew").write_text("x") assert not hygiene.config_drift(etc).pacnew def test_config_drift_of_a_missing_directory_is_empty(): got = hygiene.config_drift(Path("/definitely/not/here")) assert got.pacnew == [] and got.pacsave == [] and got.pacorig == [] def test_config_drift_on_the_real_etc_matches_a_plain_scan(): """Whatever is actually there, the report must agree with the filesystem.""" got = sorted(str(p) for p in hygiene.config_drift().pacnew) raw = sorted(str(p) for p in Path("/etc").rglob("*.pacnew")) assert got == raw # -------------------------------------------------------------------------- # reboot advice # -------------------------------------------------------------------------- def test_no_reboot_is_advised_when_the_running_kernel_is_installed(): """A false "reboot now" is worse than none: it teaches people to ignore it.""" import tempfile modules = Path(tempfile.mkdtemp()) (modules / hygiene.running_kernel()).mkdir() assert hygiene.reboot_reason(modules=modules) is None def test_a_reboot_is_advised_when_the_running_kernel_is_gone(): import tempfile modules = Path(tempfile.mkdtemp()) (modules / "9.9.9-other").mkdir() reason = hygiene.reboot_reason(modules=modules) assert reason and hygiene.running_kernel() in reason, reason def test_an_updated_kernel_package_triggers_the_advice_without_module_info(): """The module tree is authoritative, but an empty one must not silence it.""" import tempfile empty = Path(tempfile.mkdtemp()) reason = hygiene.reboot_reason({"linux"}, modules=empty) assert reason and "linux" in reason, reason def test_an_unrelated_update_does_not_trigger_a_reboot(): import tempfile empty = Path(tempfile.mkdtemp()) assert hygiene.reboot_reason({"vim", "htop"}, modules=empty) is None if __name__ == "__main__": raise SystemExit(main())