"""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/<pid>/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())