"""Core tests. Runnable directly (`python tests/test_core.py`) or under pytest.
The interesting ones are the integration tests: a real process tree observed
through /proc, and the real .SRCINFO corpus in ~/.cache/paru/clone.
"""
from __future__ import annotations
import os
import stat
import subprocess
import sys
import tempfile
import time
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from updater import icons, layout, procwatch # noqa: E402
from updater.buildgraph import MissingMetadata, build_graph # noqa: E402
from updater.srcinfo import SrcInfo, dep_name, load, parse_pkgbuild, parse_srcinfo # noqa: E402
from updater.theme import ANSI16, Theme # noqa: E402
CLONE = Path.home() / ".cache/paru/clone"
# --------------------------------------------------------------------------
# theme
# --------------------------------------------------------------------------
def test_theme_roles_are_ansi16_or_default():
t = Theme.default()
for name in (
"text",
"muted",
"rule",
"accent",
"ok",
"warn",
"alert",
"err",
"old",
"new",
"ver",
):
style = getattr(t, name)
for token in style.split():
assert token in ANSI16 or token in ("bold", "dim", "underline", "italic", "reverse"), (
f"{name}: {token!r} is not an ANSI-16 slot or an attribute"
)
def test_theme_remap_rejects_junk():
t = Theme.default()
assert t.remap(ok="cyan").ok == "dim cyan"
for bad in ("notacolour", "bright_purple"):
try:
t.remap(ok=bad)
except ValueError:
pass
else:
raise AssertionError(f"remap accepted {bad!r}")
try:
t.remap(nonsense="red")
except ValueError:
pass
else:
raise AssertionError("remap accepted an unknown role")
# --------------------------------------------------------------------------
# icons
# --------------------------------------------------------------------------
def test_icon_detection_matrix():
local = {"TERM": "xterm-kitty"}
cases = [
(local, True, True, "nerd"),
(local, True, False, "unicode"),
({**local, "SSH_CONNECTION": "1.2.3.4 5 5.6.7.8 9"}, True, True, "unicode"),
({**local, "TMUX": "/tmp/tmux-1000/default,123,0"}, True, True, "unicode"),
({"TERM": "dumb"}, True, True, "ascii"),
({**local, "NO_ICONS": "1"}, True, True, "ascii"),
(local, False, True, "ascii"),
]
for env, tty, nerd, expected in cases:
tier, reason = icons.detect(env, is_tty=tty, nerd_font=nerd)
assert tier == expected, f"{env} -> {tier}, expected {expected} ({reason})"
assert reason
def test_icon_override_wins_and_is_validated():
tier, _ = icons.detect({"TERM": "dumb"}, is_tty=False, nerd_font=False, override="nerd")
assert tier == "nerd"
try:
icons.detect({}, override="emoji")
except ValueError:
pass
else:
raise AssertionError("bad override accepted")
def test_every_state_glyph_is_one_cell():
"""A two-cell glyph silently breaks every column to its right."""
for name, g in icons.TIERS.items():
for field in ("done", "run", "wait", "fail", "shared", "warn", "arrow"):
glyph = getattr(g, field)
assert len(glyph) == 1, f"{name}.{field} is {len(glyph)} cells: {glyph!r}"
# --------------------------------------------------------------------------
# layout
# --------------------------------------------------------------------------
def test_layout_clips_rather_than_overflowing():
"""A renderer must never hand the terminal a line it will wrap."""
grid = layout.Grid(width=20)
row = grid.row().span("x" * 25)
assert row.done().cell_len <= 20
assert row.clipped == 1, "clipping happened but was not reported"
def test_layout_reports_clipping_through_every_primitive():
grid = layout.Grid(width=20, right_margin=2)
assert grid.row().span("x" * 50).clipped == 1
assert grid.row().right("y" * 50).clipped >= 1
# `to` past the edge is clamped, not silently accepted
row = grid.row().span("abc").to(99)
assert row.done().cell_len <= 20
assert row.clipped == 1
# and nothing is clipped when it genuinely fits
assert grid.row().span("abc").right("def").clipped == 0
def test_layout_check_catches_a_hand_built_overlong_line():
from rich.text import Text as RText
try:
layout.check([RText("z" * 30)], width=20)
except layout.Overflow:
pass
else:
raise AssertionError("check() did not catch an over-wide line")
def test_layout_right_align_and_fill():
grid = layout.Grid(width=20, right_margin=2)
assert grid.row().span("a").right("end").done().plain == " a" + " " * 12 + "end"
assert len(grid.row().fill().done().plain) == 20
assert grid.right == 18
# a value wider than its column is clipped, never allowed to push past it
assert grid.row().right("w" * 40).done().cell_len <= 20
# --------------------------------------------------------------------------
# srcinfo / pkgbuild
# --------------------------------------------------------------------------
def test_dep_name_strips_constraints_and_descriptions():
assert dep_name("python>=3.14") == "python"
assert dep_name("foo<2: for bar") == "foo"
assert dep_name("plain") == "plain"
def test_parse_srcinfo_real_shape():
info = parse_srcinfo(
"pkgbase = demo\n"
"\tpkgver = 1.2.3\n\tpkgrel = 4\n\tepoch = 2\n"
"\tmakedepends = cmake\n\tcheckdepends = python-pytest\n"
"\tdepends = glibc\n\tdepends = python-attrs\n"
"\tsource = https://example.invalid/a.tar.gz\n"
"\tsha256sums = abc\n"
"\n"
"pkgname = demo\n"
"\tdepends = extra-dep\n"
"\n"
"pkgname = demo-docs\n"
)
assert info.pkgbase == "demo"
assert info.version == "2:1.2.3-4"
assert info.declared_names() == ["demo", "demo-docs"]
assert set(map(dep_name, info.depends)) == {"glibc", "python-attrs", "extra-dep"}
assert info.per_package_depends["demo"] == ["extra-dep"]
assert info.build_depends == ["cmake", "python-pytest", "glibc", "python-attrs", "extra-dep"]
def test_pkgbuild_parser_handles_arrays_quotes_and_functions():
pb = parse_pkgbuild(
"# Maintainer: someone\n"
"pkgname=demo\n"
"_name=demo\n"
"pkgver=1.0\n"
"depends=('a' 'b')\n"
"source=(\n"
' "https://example.invalid/${pkgver}.tar.gz"\n'
" 'local.patch'\n"
")\n"
'pkgdesc="has a # hash and a = sign"\n'
"\n"
"build() {\n"
' cd "$srcdir/$_name-$pkgver"\n'
" make\n"
"}\n"
"\n"
"package() {\n"
' make DESTDIR="$pkgdir" install\n'
"}\n"
)
assert pb.one("pkgver") == "1.0"
assert pb.get("depends") == ["a", "b"]
assert pb.get("source") == ["https://example.invalid/${pkgver}.tar.gz", "local.patch"]
assert pb.one("pkgdesc") == "has a # hash and a = sign"
assert sorted(pb.funcs) == ["build", "package"]
assert 'DESTDIR="$pkgdir"' in pb.funcs["package"]
def test_pkgbuild_parser_never_executes():
"""The whole point: a hostile PKGBUILD must not run at review time."""
marker = Path(tempfile.mkdtemp()) / "pwned"
pb = parse_pkgbuild(
f"pkgver=$(touch {marker})\n"
"x=$((1/0))\n"
f'source=("$(touch {marker})")\n'
f"build() {{ touch {marker}; }}\n"
)
assert not marker.exists(), "parsing executed PKGBUILD content"
assert "build" in pb.funcs
del pb
def test_real_srcinfo_corpus_parses():
if not CLONE.is_dir():
return
packages = [p.name for p in CLONE.iterdir() if p.is_dir()]
assert packages, "no cloned packages to check"
for name in packages:
info, pb = load(CLONE, name)
assert info is not None, f"{name}: no .SRCINFO"
assert pb is not None, f"{name}: no PKGBUILD"
assert info.pkgbase, f"{name}: empty pkgbase"
assert info.pkgver, f"{name}: empty pkgver"
# --------------------------------------------------------------------------
# build graph
# --------------------------------------------------------------------------
def _info(name: str, *deps: str) -> SrcInfo:
return SrcInfo(pkgbase=name, pkgver="1", pkgrel="1", depends=list(deps))
def test_graph_orders_dependencies_before_dependents():
infos = {"a": _info("a", "b"), "b": _info("b", "c"), "c": _info("c")}
g = build_graph(infos, set(infos))
assert g.order.index("c") < g.order.index("b") < g.order.index("a")
assert g.roots == ["a"]
assert not g.cycles
def test_graph_finds_diamonds():
infos = {
"top": _info("top", "left", "right"),
"left": _info("left", "shared"),
"right": _info("right", "shared"),
"shared": _info("shared"),
}
g = build_graph(infos, set(infos))
assert g.shared == {"shared"}
assert g.dependents("shared") == ["left", "right"]
def test_graph_reports_cycles_instead_of_hiding_them():
infos = {"a": _info("a", "b"), "b": _info("b", "c"), "c": _info("c", "a")}
g = build_graph(infos, set(infos))
assert g.cycles, "cycle not reported"
assert set(g.cycles[0]) >= {"a", "b", "c"}
assert not g.order
def test_graph_ignores_edges_to_installed_packages():
infos = {"only": _info("only", "glibc", "python-attrs")}
g = build_graph(infos, {"only"})
assert g.nodes["only"].deps == set()
assert g.nodes["only"].external == {"glibc", "python-attrs"}
assert g.order == ["only"]
def test_missing_metadata_is_reported_not_treated_as_no_dependencies():
"""A node with no .SRCINFO has *unknown* deps, which is not the same as none."""
infos = {"known": _info("known")}
g = build_graph(infos, {"known", "uncloned", "other"})
assert set(g.nodes) == {"known", "uncloned", "other"}
assert g.unknown == ["other", "uncloned"]
assert not g.is_orderable
# an unknown node must never be scheduled, because its deps cannot be proven
assert g.ready(set()) == ["known"]
assert "uncloned" not in g.order
def test_strict_mode_refuses_to_build_an_unorderable_graph():
try:
build_graph({"known": _info("known")}, {"known", "uncloned"}, strict=True)
except MissingMetadata as exc:
assert "uncloned" in exc.names
else:
raise AssertionError("strict mode accepted a graph with unknown metadata")
def test_dependency_on_an_unknown_package_is_not_reported_as_a_cycle():
"""These need completely different fixes, so they must not be conflated."""
infos = {"app": _info("app", "helper")}
g = build_graph(infos, {"app", "helper"})
assert g.unknown == ["helper"]
assert not g.cycles, f"reported a bogus cycle: {g.cycles}"
assert "app" in g.blocked_by_unknown
assert "app" not in g.order
def test_a_complete_graph_reports_itself_orderable():
infos = {"a": _info("a", "b"), "b": _info("b")}
g = build_graph(infos, set(infos))
assert g.is_orderable
assert g.unknown == [] and g.cycles == [] and g.blocked_by_unknown == []
def test_ready_and_blocked_drive_a_scheduler():
infos = {
"top": _info("top", "mid"),
"mid": _info("mid", "base"),
"base": _info("base"),
"free": _info("free"),
}
g = build_graph(infos, set(infos))
assert set(g.ready(set())) == {"base", "free"}
assert set(g.ready({"base"})) == {"free", "mid"}
assert g.ready({"base", "free"}) == ["mid"]
assert g.blocked({"base"}) == {"mid", "top"}
assert g.blocked({"mid"}) == {"top"}
# --------------------------------------------------------------------------
# procwatch: the integration test that matters
# --------------------------------------------------------------------------
def _fake_tool(bindir: Path, name: str) -> Path:
"""A process that behaves like `name` and stays alive.
`exec -a` sets argv0 so /proc reports the tool name, which is what a real
compiler invocation looks like.
"""
path = bindir / name
path.write_text(f"#!/bin/bash\nexec -a {name} sleep 30\n")
path.chmod(path.stat().st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH)
return path
def test_watcher_identifies_package_phase_and_elapsed():
"""A real process tree, observed only through /proc."""
with tempfile.TemporaryDirectory() as td:
root = Path(td)
clone = root / "clone"
bin_ = root / "bin"
bin_.mkdir()
# a believable layout: clone/<pkg>/src/<dir>/<subdir>
for pkg in ("alpha", "beta"):
(clone / pkg / "src" / "build" / "sub").mkdir(parents=True)
(clone / ".git").mkdir()
makepkg = _fake_tool(bin_, "makepkg")
cargo = _fake_tool(bin_, "cargo")
cc1plus = _fake_tool(bin_, "cc1plus")
cc1plus_b = _fake_tool(bin_, "cc1plus")
# makepkg at the package root, cargo a level down, cc1plus deeper still
procs = [
subprocess.Popen([str(makepkg)], cwd=clone / "alpha"),
subprocess.Popen([str(cargo)], cwd=clone / "alpha" / "src" / "build"),
subprocess.Popen([str(cc1plus)], cwd=clone / "alpha" / "src" / "build" / "sub"),
subprocess.Popen([str(cc1plus_b)], cwd=clone / "alpha" / "src" / "build"),
subprocess.Popen([str(makepkg)], cwd=clone / "beta"),
]
try:
time.sleep(0.4)
watcher = procwatch.Watcher(clone, known={"alpha", "beta"})
# observe as if we were the parent: pretend root is the test process
states = watcher.poll(os.getpid())
by_pkg = {s.package: s for s in states}
assert set(by_pkg) == {"alpha", "beta"}, by_pkg
alpha = by_pkg["alpha"]
# the phase label reflects the highest-level tool running, not
# whichever process happened to be listed first
assert alpha.phase == "cargo", alpha.phase
assert alpha.jobs >= 2, alpha # two compilers, counted separately
assert 0 <= alpha.elapsed < 10, alpha.elapsed
assert alpha.clock.endswith("s")
assert by_pkg["beta"].phase == "makepkg", by_pkg["beta"].phase
# completion is inferred once the processes are gone
assert watcher.finished == []
for proc in procs:
proc.terminate()
for proc in procs:
proc.wait(timeout=5)
time.sleep(0.2)
assert watcher.poll(os.getpid()) == []
assert set(watcher.new_finishes()) == {"alpha", "beta"}
# drains, so the caller can append to the feed exactly once
assert watcher.new_finishes() == []
finally:
for proc in procs:
if proc.poll() is None:
proc.kill()
def test_package_of_rejects_paths_outside_and_dotdirs():
clone = Path("/home/x/.cache/paru/clone")
assert procwatch.package_of("/home/x/.cache/paru/clone/foo/src", clone) == "foo"
assert procwatch.package_of("/home/x/.cache/paru/clone/.git", clone) is None
assert procwatch.package_of("/etc", clone) is None
assert procwatch.package_of(None, clone) is None
assert procwatch.package_of("/home/x/.cache/paru/clone/foo", clone, {"bar"}) is None
def test_duration_formatting():
assert procwatch.format_duration(0) == "0s"
assert procwatch.format_duration(41) == "41s"
assert procwatch.format_duration(71) == "1m11s"
assert procwatch.format_duration(3661) == "1h01m"
def test_download_progress_on_missing_dir_is_silent():
assert procwatch.download_progress(Path("/definitely/not/here")) == (0, 0)
def test_parse_proc_stat_survives_comm_with_spaces():
"""comm is parenthesised but may contain spaces and parens."""
proc = procwatch.read_proc(os.getpid(), procwatch.boot_time(), procwatch._clk_tck())
assert proc is not None
assert proc.pid == os.getpid()
assert proc.elapsed >= 0
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
if __name__ == "__main__":
raise SystemExit(main())