"""repo.py tests.
The valuable ones assert agreement with pacman itself. Every number here was
originally checked by hand against `pacman -Q*`; this file keeps it that way,
because a repo reader that quietly disagrees with pacman is worse than no reader
at all -- it would report a plan that the transaction then contradicts.
"""
from __future__ import annotations
import shutil
import subprocess
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from updater import repo # noqa: E402
PYALPM = True
PACMAN = shutil.which("pacman") is not None
def _code(fn) -> str:
"""A function's source with all whitespace removed.
These assertions are about what the code does, and the exact line breaks
are the formatter's business -- pinning them made every one of these tests
fail the first time the tree was formatted.
"""
return "".join(_code(fn).split())
def pacman_lines(*args: str) -> list[str]:
proc = subprocess.run(["pacman", *args], capture_output=True, text=True, check=False)
return [line for line in proc.stdout.splitlines() if line.strip()]
# --------------------------------------------------------------------------
# configuration
# --------------------------------------------------------------------------
def _code(fn) -> str:
"""A function's source with all whitespace removed.
These assertions are about what the code does; where the line breaks fall
is the formatter's business. Stripping whitespace from both the source and
the snippet means a reformat cannot make them fail.
"""
import inspect
return _strip(inspect.getsource(fn))
def _strip(s: str) -> str:
return "".join(s.split())
def test_read_conf_finds_the_real_repositories():
config = repo.read_conf()
assert config.repos, "no repositories parsed from pacman.conf"
assert "options" not in config.repos, "'[options]' was mistaken for a repo"
for name in ("core", "extra"):
assert name in config.repos, name
def test_read_conf_missing_file_falls_back_instead_of_raising():
config = repo.read_conf("/definitely/not/here/pacman.conf")
assert config.repos
assert config.ignore_pkg == frozenset()
def test_read_conf_parses_ignore_lists(tmp: Path | None = None):
import tempfile
with tempfile.TemporaryDirectory() as td:
path = Path(td) / "pacman.conf"
path.write_text(
"[options]\n"
"IgnorePkg = linux nvidia\n"
"IgnoreGroup = modified\n"
"\n"
"[core]\n"
"Include = /etc/pacman.d/mirrorlist\n"
"\n"
"[extra]\n"
)
config = repo.read_conf(path)
assert set(config.ignore_pkg) == {"linux", "nvidia"}
assert set(config.ignore_group) == {"modified"}
assert config.repos == ("core", "extra")
# --------------------------------------------------------------------------
# the bug that made everything look up to date
# --------------------------------------------------------------------------
def test_connect_registers_every_configured_repo():
"""Without explicit registration pyalpm reports zero sync dbs.
Which means zero updates, which is indistinguishable from a system that is
fully up to date.
"""
if not PACMAN:
return
config = repo.read_conf()
with repo.connect() as handle:
registered = {db.name for db in handle.get_syncdbs()}
assert registered == set(config.repos), f"registered {registered}, expected {set(config.repos)}"
# --------------------------------------------------------------------------
# agreement with pacman
# --------------------------------------------------------------------------
def test_plan_agrees_with_pacman_on_the_number_of_upgrades():
if not PACMAN:
return
with repo.connect() as handle:
mine = len(repo.plan(handle))
assert mine == len(pacman_lines("-Qu")), "plan disagrees with pacman -Qu"
def test_orphans_match_both_of_pacmans_strictness_levels():
"""A package merely *suggested* by another is not an orphan to `-Qtd`."""
if not PACMAN:
return
with repo.connect() as handle:
strict = repo.orphans(handle)
loose = repo.orphans(handle, include_optional=True)
assert len(strict) == len(pacman_lines("-Qtdq"))
assert len(loose) == len(pacman_lines("-Qttdq"))
# and the loose set is a superset, since it counts fewer things as required
assert set(strict) <= set(loose)
def test_foreign_explicit_and_installed_agree_with_pacman():
if not PACMAN:
return
with repo.connect() as handle:
foreign = repo.foreign(handle)
explicit = repo.explicit(handle)
installed = repo.installed(handle)
assert len(foreign) == len(pacman_lines("-Qqm"))
assert len(explicit) == len(pacman_lines("-Qqe"))
assert len(installed) == len(pacman_lines("-Qq"))
# every foreign package is installed, by definition
assert set(foreign) <= set(installed)
assert set(explicit) <= set(installed)
def test_installed_reports_real_versions():
with repo.connect() as handle:
installed = repo.installed(handle)
assert installed, "no packages reported as installed"
assert "pacman" in installed
assert installed["pacman"][0].isdigit(), installed["pacman"]
def test_plan_never_reports_a_foreign_package_as_a_removal():
"""AUR packages have no sync entry and `pacman -Su` leaves them alone."""
with repo.connect() as handle:
plan = repo.plan(handle)
foreign = set(repo.foreign(handle))
assert not hasattr(plan, "removals"), "the removals field came back"
assert not (foreign & {u.name for u in plan.updates})
def test_plan_totals_are_sums_of_their_parts():
with repo.connect() as handle:
plan = repo.plan(handle)
assert plan.download_total == sum(u.download for u in plan.updates)
assert plan.installed_delta == sum(u.delta for u in plan.updates)
assert len(plan) == len(plan.updates)
# --------------------------------------------------------------------------
# refresh machinery
# --------------------------------------------------------------------------
def test_sync_argv_uses_fakeroot_when_available():
"""pacman's sandbox refuses files it does not own, so fakeroot matters."""
argv = repo._sync_argv("/tmp/example")
assert "--dbpath" in argv and "/tmp/example" in argv
if shutil.which("fakeroot"):
assert argv[0] == "fakeroot"
assert "--disable-sandbox-filesystem" in argv
# and it must never be able to touch the real database
assert "/var/lib/pacman" not in " ".join(argv)
def test_sync_db_age_is_a_number_or_none():
age = repo.sync_db_age()
assert age is None or age >= 0
def test_sync_db_age_of_a_missing_path_is_none():
assert repo.sync_db_age("/definitely/not/here") is None
def test_preview_always_yields_a_usable_handle_even_when_sync_fails():
"""Offline, or a mirror outage, must degrade to the databases on disk."""
with repo.preview(sync=False) as (handle, refreshed, note):
assert refreshed is False
assert note, "an unrefreshed preview must say so"
# and it still answers questions rather than collapsing
assert repo.installed(handle)
assert isinstance(repo.plan(handle).updates, list)
def test_preview_reports_a_reason_when_it_cannot_sync():
with repo.preview(sync=False) as (_handle, _refreshed, note):
assert "skip" in note.lower() or "refresh" in note.lower()
def test_a_preview_almost_matching_pacman_proves_the_sync_path_reads_the_same_format():
"""With sync disabled the preview must equal a plain connect()."""
with repo.preview(sync=False) as (preview_handle, _r, _n):
preview_plan = repo.plan(preview_handle)
with repo.connect() as handle:
direct_plan = repo.plan(handle)
assert len(preview_plan.updates) == len(direct_plan.updates)
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
# --------------------------------------------------------------------------
# sizes, against pacman's own numbers
# --------------------------------------------------------------------------
def _pacman_size(args: list[str], needle: str) -> dict[str, int]:
"""Parse `pacman -Qi/-Si` human sizes back into bytes, for comparison."""
import re
import subprocess
out = subprocess.run(["pacman", *args], capture_output=True, text=True, check=False).stdout
sizes: dict[str, int] = {}
name = None
units = {"B": 1, "KiB": 1024, "MiB": 1024**2, "GiB": 1024**3}
for line in out.splitlines():
if line.startswith("Name "):
name = line.split(":", 1)[1].strip()
elif needle in line and name:
m = re.search(r"([\d.]+)\s*(\w+)", line.split(":", 1)[1])
if m:
sizes[name] = int(float(m.group(1)) * units.get(m.group(2), 1))
return sizes
def test_installed_size_matches_pacman():
"""pyalpm's `isize` must agree with `pacman -Qi`.
`size` is not a substitute: it is 0 on a local package and the download size
on a sync one, which made every on-disk delta look like a full-size increase.
"""
with repo.connect() as handle:
installed = repo.installed(handle)
names = sorted(installed)[:40]
if not names:
return
truth = _pacman_size(["-Qi", *names], "Installed Size")
local = {p.name: p for p in handle.get_localdb().pkgcache}
checked = 0
for name, expected in truth.items():
pkg = local.get(name)
if pkg is None or expected == 0:
continue
got = getattr(pkg, "isize", 0)
# pacman rounds to two decimals, so allow a little slack
assert abs(got - expected) <= max(2048, expected * 0.01), (
f"{name}: isize={got:,} but pacman says {expected:,}"
)
checked += 1
assert checked >= 5, f"only compared {checked} packages"
def test_sync_size_agrees_with_pacman_and_is_not_the_download_size():
with repo.connect() as handle:
best = repo._best_versions(handle)
names = sorted(best)[:40]
if not names:
return
installed_truth = _pacman_size(["-Si", *names], "Installed Size")
download_truth = _pacman_size(["-Si", *names], "Download Size")
compared = 0
for name in names:
sp = best.get(name)
if sp is None:
continue
want_installed = installed_truth.get(name, 0)
if want_installed == 0:
continue
got = getattr(sp, "isize", 0)
assert abs(got - want_installed) <= max(2048, want_installed * 0.01), (
f"{name}: isize={got:,} but pacman says {want_installed:,}"
)
# and it must not have been confused with the download size
want_download = download_truth.get(name, 0)
if want_download and want_installed != want_download:
assert got != want_download, f"{name}: isize was populated with the download size"
compared += 1
assert compared >= 3, f"only compared {compared} packages"
def test_an_unchanged_package_has_a_zero_delta_not_a_full_size():
"""The regression: delta used to be the whole installed size."""
with repo.connect() as handle:
local = handle.get_localdb()
best = repo._best_versions(handle)
same = [
(lp, best[lp.name])
for lp in local.pkgcache
if lp.name in best and best[lp.name].version == lp.version
]
assert same, "no packages at their sync version to check"
for lp, sp in same[:50]:
delta = getattr(sp, "isize", 0) - getattr(lp, "isize", 0)
assert delta == 0, f"{lp.name}: unchanged but delta is {delta:,}"
assert delta != getattr(sp, "isize", 0) or getattr(sp, "isize", 0) == 0
def test_size_is_the_trap_and_isize_is_the_answer():
"""Documents the pyalpm semantics that caused the wrong-size bug.
On a sync package `size` is the compressed download size, not the installed
size; on a local package it is 0. `isize` is the installed size in both
cases. A tiny package can legitimately download larger than it installs, so
the only trustworthy check is against pacman, which the tests above do.
"""
with repo.connect() as handle:
best = repo._best_versions(handle)
matched = 0
for name, sp in list(best.items())[:300]:
size = getattr(sp, "size", 0)
download = getattr(sp, "download_size", 0)
if size and download:
assert size == download, f"{name}: sync `size` was expected to equal download_size"
matched += 1
assert matched >= 5, f"only checked {matched} sync packages"
# and on a local package it carries nothing useful
local = list(handle.get_localdb().pkgcache)[:200]
assert local, "no local packages to check"
zeroed = sum(1 for lp in local if getattr(lp, "size", 0) == 0)
assert zeroed > len(local) // 2, (
f"local `size` was 0 for only {zeroed}/{len(local)}; pyalpm changed"
)
if __name__ == "__main__":
raise SystemExit(main())
def test_ignore_pkg_is_a_glob_not_a_literal_name():
"""pacman.conf documents both lists as shell globs.
Comparing for equality meant `IgnorePkg = linux*` matched nothing and the
plan offered packages pacman would refuse to touch.
"""
class P:
def __init__(self, name, groups=()):
self.name, self.groups = name, list(groups)
class H:
ignorepkgs = ["linux*", "nvidia"]
ignoregrps = ["plasma*"]
assert repo._is_ignored(H(), P("linux")) is True
assert repo._is_ignored(H(), P("linux-lts")) is True
assert repo._is_ignored(H(), P("nvidia")) is True
assert repo._is_ignored(H(), P("mesa")) is False
assert repo._is_ignored(H(), P("kwin", ["plasma-desktop"])) is True
assert repo._is_ignored(H(), P("kwin", ["gnome"])) is False
def test_every_handle_registers_the_ignore_lists():
"""`connect()` did and `preview()` did not, and the plan uses preview.
So `_is_ignored` always saw an empty list and offered every IgnorePkg
package as a pending upgrade.
"""
for fn in (repo.connect, repo.preview, repo._fallback):
source = _code(fn)
assert _strip("_register_ignores") in source, fn.__name__