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