"""Driver tests.
The valuable ones run a real child process and observe it through /proc, exactly
as the tool does in production. Nothing here needs a terminal or a network.
"""
from __future__ import annotations
import contextlib
import io
import os
import signal
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 rich.console import Console # noqa: E402
from updater import icons, live # noqa: E402
from updater.buildgraph import build_graph # noqa: E402
from updater.live import BuildObserver, LiveRunner # noqa: E402
from updater.render.build import FeedEntry # noqa: E402
from updater.srcinfo import SrcInfo # noqa: E402
from updater.theme import Theme # noqa: E402
THEME = Theme.default()
GLYPHS = icons.glyphs("unicode")
def info(name: str, *deps: str) -> SrcInfo:
return SrcInfo(pkgbase=name, pkgver="1", pkgrel="1", depends=list(deps))
def fake_tool(bindir: Path, name: str, body: str) -> Path:
path = bindir / name
path.write_text(f"#!/bin/bash\n{body}\n")
path.chmod(path.stat().st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH)
return path
# --------------------------------------------------------------------------
# parallelism: the bug this exists to prevent
# --------------------------------------------------------------------------
def test_pacman_output_with_accents_does_not_re_read_lines():
"""`cut` indexed decoded characters while `len(chunk)` counted bytes.
With any multi-byte output in the same chunk the offset landed behind the
newline, so the next poll re-read consumed content and emitted a
fabricated line into the feed.
"""
import tempfile
from pathlib import Path as P
log = P(tempfile.mkdtemp()) / "x.log"
log.write_bytes(":: Über Paket — läuft\n firefox-1.2-1 downloading...\n".encode() + b"part")
obs = live.RepoObserver.__new__(live.RepoObserver)
obs.log_path, obs._offset = log, 0
assert obs._new_lines() == [":: Über Paket — läuft", " firefox-1.2-1 downloading..."]
with open(log, "ab") as fh:
fh.write(b"ial\n")
# the leftover half-line, whole -- not a re-read of what came before
assert obs._new_lines() == ["partial"]
def test_terminate_reports_whether_the_child_actually_died():
"""`returncode or 0` reported a child that survived SIGKILL as success.
A process stuck in D state, or on a hung mount, outlives the signal; the
phase was then called done and the next one started beside a live pacman
transaction holding the database lock.
"""
import subprocess
proc = subprocess.Popen(["true"], process_group=0)
proc.wait()
assert live.terminate(proc) is True, "a dead child is gone"
# one that ignores SIGINT is escalated to SIGTERM, which it accepts
proc = subprocess.Popen(["sh", "-c", "trap '' INT; sleep 30"], process_group=0)
try:
assert live.terminate(proc, grace=0.3, term_grace=2.0) is True
finally:
if proc.poll() is None:
proc.kill()
proc.wait()
def test_a_phase_that_could_not_be_stopped_is_not_reported_as_success():
"""`returncode or 0` turned a surviving child into a successful phase.
A process that outlives SIGKILL is rare and not worth faking a kernel for,
so what is checked is the contract that made it possible to miss: the
"could not stop it" code exists and is a failure.
"""
from updater import stream
assert stream.EXIT_STILL_RUNNING != 0, "a surviving child is a failure"
assert stream.EXIT_STILL_RUNNING == live.EXIT_STILL_RUNNING, (
"the two paths disagree about what it means"
)
def test_a_successful_phase_still_reports_your_zero():
from updater import stream
obs = live.BuildObserver(build_graph({}, set()), Path("/tmp"), total=0)
buf = io.StringIO()
runner = stream.StreamingRunner(obs, stream.Emitter(buf), poll_hz=20)
assert runner.run(["true"], "repo", Path(tempfile.mkdtemp()) / "x.log") == 0
def test_cores_are_divided_between_workers_not_multiplied():
env = live.build_env(workers=8, nproc=16)
assert env["MAKEFLAGS"] == "-j2", env["MAKEFLAGS"]
assert env["CARGO_BUILD_JOBS"] == "2"
assert env["NINJAFLAGS"] == "-j2"
def test_single_worker_gets_the_whole_machine():
assert live.build_env(workers=1, nproc=16)["MAKEFLAGS"] == "-j16"
def test_an_inherited_makeflags_is_overridden_not_respected():
"""`-j15` inherited from a shell is wrong for any worker count."""
env = live.build_env(workers=1, nproc=16, base={"MAKEFLAGS": "-j15"})
assert env["MAKEFLAGS"] == "-j16"
# and it must never be worse than one job on a machine with cores
for workers in (1, 2, 3, 16, 64):
env = live.build_env(workers=workers, nproc=16)
assert int(env["MAKEFLAGS"][2:]) >= 1
def test_parallelism_label_is_honest():
assert "8 workers" in live.parallelism_label(8, 16)
assert "2 jobs" in live.parallelism_label(8, 16)
assert "workers" not in live.parallelism_label(1, 16)
# --------------------------------------------------------------------------
# observer: feed integrity
# --------------------------------------------------------------------------
def graph_for(*names: str):
infos = {n: info(n) for n in names}
return build_graph(infos, set(infos))
def test_a_package_enters_the_feed_exactly_once():
obs = BuildObserver(graph_for("a"), Path("/nonexistent"))
obs._watcher.poll = lambda pid: [] # pretend it never ran
obs._watcher.seen = {"a"}
obs._watcher.finished = ["a"]
first = obs.poll(None)
assert [e.name for e in first] == ["a"]
assert len(obs.feed) == 1
# polling again must not duplicate or re-report it
for _ in range(5):
assert obs.poll(None) == []
assert len(obs.feed) == 1
assert "a" in obs.completed
def test_a_package_that_failed_is_not_also_reported_as_finished():
obs = BuildObserver(graph_for("a"), Path("/nonexistent"))
obs.fail("a", "check failed")
obs._watcher.seen = {"a"}
obs._watcher.finished = ["a"]
assert obs.poll(None) == []
assert [e.state for e in obs.feed] == ["fail"]
assert obs.failed == {"a"}
def test_seeding_supports_resuming_a_partly_built_set():
obs = BuildObserver(graph_for("a", "b", "c"), Path("/nonexistent"), total=3)
obs.seed([FeedEntry("a", "1-1", 5.0), FeedEntry("b", "1-1", 0.0, state="fail")])
assert obs.completed == {"a"}
assert obs.failed == {"b"}
assert len(obs.feed) == 2
def test_failure_blocks_dependents():
graph = build_graph({"top": info("top", "base"), "base": info("base")}, {"top", "base"})
obs = BuildObserver(graph, Path("/nonexistent"))
obs.fail("base")
assert obs.blocked == {"top"}
def test_totals_compute_an_eta_only_once_something_has_finished():
obs = BuildObserver(graph_for("a", "b"), Path("/nonexistent"), total=2)
assert obs.totals().eta is None
obs.poll(None)
obs.feed = [FeedEntry("a", "1", 2.0)]
totals = obs.totals()
assert totals.total == 2
assert totals.eta is not None and totals.eta >= 0
assert totals.elapsed >= 0
def test_download_rate_never_goes_negative_when_part_files_are_renamed():
"""`.part` files vanish on completion, so the raw total can drop."""
with tempfile.TemporaryDirectory() as td:
cache = Path(td)
obs = BuildObserver(graph_for("a"), Path("/nonexistent"), cache_dir=cache)
part = cache / "pkg-1.pkg.tar.zst.part"
part.write_bytes(b"x" * 1000)
obs._last_dl_at -= 1.0
obs.poll(None)
assert obs.download_rate > 0
# the download completes and the file is renamed away
part.unlink()
obs._last_dl_at -= 1.0
obs.poll(None)
assert obs.download_rate >= 0, "rate went negative after a rename"
assert obs.download_bytes == 0
assert obs.download_peak == 1000, "high-water mark was lost"
def test_durations_are_captured_while_the_process_is_alive():
"""Once a build's processes exit there is nothing left to measure."""
from updater.procwatch import BuildState
obs = BuildObserver(graph_for("a"), Path("/nonexistent"))
obs._watcher.poll = lambda pid: [BuildState("a", "cargo", 12.5)]
obs._watcher.new_finishes = list
obs.poll(1234)
assert obs._durations["a"] == 12.5
# the process goes away; the duration must survive into the feed entry
obs._watcher.poll = lambda pid: []
obs._watcher.new_finishes = lambda: ["a"]
obs.poll(1234)
assert obs.feed[0].elapsed == 12.5
def test_feed_line_is_a_single_permanent_line():
done = BuildObserver.feed_line(FeedEntry("pkg", "1.0-1", 41.0))
assert "\n" not in done and "pkg" in done and "41s" in done
failed = BuildObserver.feed_line(FeedEntry("pkg", "1.0-1", 3.0, state="fail"))
assert failed.startswith("x")
shared = BuildObserver.feed_line(FeedEntry("pkg", "1.0-1", 3.0, shared=True))
assert "shared" in shared
# --------------------------------------------------------------------------
# the real thing: run a child and observe it through /proc
# --------------------------------------------------------------------------
def test_runner_observes_a_real_child_process_end_to_end():
with tempfile.TemporaryDirectory() as td:
root = Path(td)
clone = root / "clone"
(clone / "pkgA").mkdir(parents=True)
bindir = root / "bin"
bindir.mkdir()
# faithful to production: we spawn paru, and paru spawns a makepkg child
# whose working directory is the package's clone dir. The launcher itself
# is deliberately NOT in the clone dir, so it must not be mistaken for a
# build.
tool = fake_tool(bindir, "paru", f'( cd "{clone}/pkgA" && exec -a makepkg sleep 1.2 )')
graph = graph_for("pkgA")
obs = BuildObserver(graph, clone, total=1)
console = Console(file=io.StringIO(), force_terminal=False, width=100)
runner = LiveRunner(
obs, theme=THEME, glyphs=GLYPHS, console=console, dashboard=False, workers=1, nproc=4
)
result = runner.run([str(tool)], root / "logs" / "build.log")
assert result.returncode == 0
assert result.log_path.exists()
assert [e.name for e in result.feed] == ["pkgA"], result.feed
assert result.feed[0].elapsed > 0, "no duration captured"
assert "pkgA" in obs.completed
assert result.elapsed >= 1.0
def test_the_launcher_itself_is_not_mistaken_for_a_build():
"""paru runs from wherever the user invoked us; that is not a package."""
with tempfile.TemporaryDirectory() as td:
root = Path(td)
clone = root / "clone"
(clone / "pkgA").mkdir(parents=True)
obs = BuildObserver(graph_for("pkgA"), clone)
# a launcher sitting directly in the clone dir must still not be read as
# a build of that package
proc = subprocess.Popen(
[sys.executable, "-c", "import time; time.sleep(0.6)"],
cwd=clone / "pkgA",
start_new_session=True,
)
try:
time.sleep(0.2)
states = obs._watcher.poll(proc.pid)
assert states == [], f"launcher misread as a build: {states}"
finally:
proc.wait(timeout=5)
def test_include_root_supports_driving_a_builder_directly():
"""The future parallel scheduler spawns makepkg itself, so the root is it."""
from updater.procwatch import descendants, snapshot
with tempfile.TemporaryDirectory() as td:
clone = Path(td) / "clone"
(clone / "pkgA").mkdir(parents=True)
proc = subprocess.Popen(["sleep", "5"], cwd=clone / "pkgA", start_new_session=True)
try:
time.sleep(0.2)
procs = snapshot()
assert descendants(proc.pid, procs) == []
included = descendants(proc.pid, procs, include_root=True)
assert [p.pid for p in included] == [proc.pid]
finally:
proc.kill()
proc.wait(timeout=5)
def test_final_drain_captures_a_package_still_running_when_the_launcher_exits():
"""Otherwise a build that ends mid-package silently loses it from the feed."""
obs = BuildObserver(graph_for("pkgA"), Path("/nonexistent"))
from updater.procwatch import BuildState
obs._watcher.poll = lambda pid: [BuildState("pkgA", "cargo", 9.0)]
obs._watcher.new_finishes = list
obs.poll(4321) # seen while alive
assert "pkgA" not in obs.completed
obs._watcher.poll = lambda pid: [] # the whole tree is gone now
obs._watcher.new_finishes = lambda: ["pkgA"]
drained = obs.poll(None)
assert [e.name for e in drained] == ["pkgA"]
assert obs.feed[0].elapsed == 9.0
def test_the_watcher_drains_when_the_root_pid_is_gone():
from updater.procwatch import Watcher
w = Watcher(Path("/nonexistent"), known={"pkgA"})
w.seen = {"pkgA"}
assert w.poll(None) == []
assert w.new_finishes() == ["pkgA"]
def test_runner_logs_child_output_instead_of_surprising_the_user():
with tempfile.TemporaryDirectory() as td:
root = Path(td)
clone = root / "clone"
(clone / "pkgA").mkdir(parents=True)
tool = fake_tool(root, "noisy", "echo 'wall of compiler text'; sleep 0.1")
obs = BuildObserver(graph_for("pkgA"), clone)
console = Console(file=io.StringIO(), force_terminal=False, width=100)
runner = LiveRunner(obs, theme=THEME, glyphs=GLYPHS, console=console, dashboard=False)
log = root / "b.log"
runner.run([str(tool)], log)
assert "wall of compiler text" in log.read_text()
def test_runner_reports_a_failing_child_without_raising():
with tempfile.TemporaryDirectory() as td:
root = Path(td)
clone = root / "clone"
clone.mkdir()
tool = fake_tool(root, "boom", "exit 7")
obs = BuildObserver(graph_for("a"), clone)
console = Console(file=io.StringIO(), force_terminal=False, width=100)
runner = LiveRunner(obs, theme=THEME, glyphs=GLYPHS, console=console, dashboard=False)
result = runner.run([str(tool)], root / "b.log")
assert result.returncode == 7
assert not result.ok
def test_flush_feed_prints_each_entry_only_once():
obs = BuildObserver(graph_for("a"), Path("/nonexistent"))
console = Console(file=io.StringIO(), force_terminal=False, width=100)
runner = LiveRunner(obs, theme=THEME, glyphs=GLYPHS, console=console)
obs.feed = [FeedEntry("a", "1", 1.0)]
runner._flush_feed()
runner._flush_feed()
out = console.file.getvalue()
assert out.count("a") == 1, f"feed line printed more than once: {out!r}"
# --------------------------------------------------------------------------
# key handling: the runtime toggle
# --------------------------------------------------------------------------
def _runner(dashboard=True):
obs = BuildObserver(graph_for("a"), Path("/nonexistent"))
console = Console(file=io.StringIO(), force_terminal=False, width=100)
return LiveRunner(obs, theme=THEME, glyphs=GLYPHS, console=console, dashboard=dashboard)
def test_v_toggles_the_dashboard_both_ways():
runner = _runner(dashboard=True)
runner._handle_keys("v")
assert runner.dashboard is False
runner._handle_keys("v")
assert runner.dashboard is True
runner._handle_keys(" ")
assert runner.dashboard is False
def test_q_leaves_the_dashboard_without_killing_the_build():
"""Killing a mid-transaction build is what breaks a package database."""
runner = _runner(dashboard=True)
runner._handle_keys("q")
assert runner.dashboard is False
def test_ctrl_c_raises_so_the_caller_can_unwind_politely():
runner = _runner()
try:
runner._handle_keys("\x03")
except KeyboardInterrupt:
pass
else:
raise AssertionError("ctrl-c did not raise")
def test_unknown_keys_are_ignored():
runner = _runner(dashboard=True)
runner._handle_keys("xyz123")
assert runner.dashboard is True
# --------------------------------------------------------------------------
# termination: polite before forceful
# --------------------------------------------------------------------------
def test_terminate_escalates_only_as_far_as_needed():
with tempfile.TemporaryDirectory() as td:
# a child that handles SIGINT and exits cleanly
nice = fake_tool(Path(td), "nice", "trap 'exit 0' INT TERM; sleep 30")
proc = subprocess.Popen([str(nice)], start_new_session=True)
time.sleep(0.3)
live.terminate(proc, grace=1.0)
assert proc.poll() is not None, "polite child was not reaped"
def test_terminate_kills_a_child_that_ignores_everything():
with tempfile.TemporaryDirectory() as td:
stubborn = fake_tool(
Path(td), "stubborn", "trap '' INT TERM; while true; do sleep 0.2; done"
)
proc = subprocess.Popen([str(stubborn)], start_new_session=True)
time.sleep(0.4)
live.terminate(proc, grace=0.3, term_grace=0.3)
assert proc.poll() is not None, "stubborn child survived"
def test_terminate_on_an_already_dead_child_is_a_no_op():
proc = subprocess.Popen(["true"], start_new_session=True)
proc.wait(timeout=5)
live.terminate(proc) # must not raise
assert proc.poll() == 0
def test_terminate_does_not_signal_an_unrelated_process_group():
"""It must target the child's group, never our own."""
mine = os.getpgrp()
proc = subprocess.Popen(["sleep", "30"], start_new_session=True)
time.sleep(0.2)
live.terminate(proc, grace=1.0)
assert os.getpgrp() == mine
assert proc.poll() is not None
def test_a_phase_keeps_its_controlling_terminal():
"""sudo keys its credential cache to the terminal.
`start_new_session=True` called setsid(), which detaches the child from the
controlling terminal. sudo then falls back to parent-PID keying, and the
parent is the backend rather than the shell the user authenticated in -- so
every privileged phase failed with "a terminal is required to read the
password".
"""
import io
kwargs = live.spawn_kwargs({"PATH": "/usr/bin"}, io.BytesIO())
assert "start_new_session" not in kwargs, kwargs
assert kwargs["process_group"] == 0, kwargs
def test_a_phase_still_cannot_read_the_keyboard():
"""The new process group is what keeps Ctrl+C off pacman; stdin stays shut
so a build cannot eat keystrokes meant for the interface."""
import io
kwargs = live.spawn_kwargs({}, io.BytesIO())
assert kwargs["stdin"] == subprocess.DEVNULL
assert kwargs["stderr"] == subprocess.STDOUT
assert kwargs["process_group"] == 0
def test_can_sudo_reports_a_prompt_as_unavailable():
"""It must ask "would sudo succeed silently", not "is sudo installed"."""
from updater import live as live_mod
calls = {}
class Done:
returncode = 1
def fake_run(argv, **kwargs):
calls["argv"] = argv
return Done()
original = live_mod.subprocess.run
live_mod.subprocess.run = fake_run
try:
assert live_mod.can_sudo() is False
finally:
live_mod.subprocess.run = original
assert "-n" in calls["argv"], calls["argv"]
def test_can_sudo_is_false_when_sudo_is_missing():
from updater import live as live_mod
def boom(argv, **kwargs):
raise FileNotFoundError("sudo")
original = live_mod.subprocess.run
live_mod.subprocess.run = boom
try:
assert live_mod.can_sudo() is False
finally:
live_mod.subprocess.run = original
def test_a_phase_is_its_own_process_group_leader():
"""The abort path signals the child's process group.
It uses killpg(getpgid(pid)), which only reaches the child and everything it
spawned if the child leads its own group. Moving off setsid must not have
cost that, or an abort would leave makepkg running.
"""
import os as _os
import subprocess as _subprocess
import time
with open(_os.devnull, "wb") as log:
proc = _subprocess.Popen(["sleep", "30"], **live.spawn_kwargs(dict(_os.environ), log))
try:
time.sleep(0.2)
assert _os.getpgid(proc.pid) == proc.pid, "not a group leader"
assert _os.getsid(proc.pid) == _os.getsid(0), "left the session"
finally:
with contextlib.suppress(ProcessLookupError):
_os.killpg(_os.getpgid(proc.pid), signal.SIGTERM)
proc.wait(timeout=5)
assert proc.returncode != 0, "the group was not signalled"
# --------------------------------------------------------------------------
# progress for a pacman transaction
# --------------------------------------------------------------------------
PACMAN_OUT = """\
:: Starting full system upgrade...
Packages (3) git-2.56.0-1 firefox-157.0-1 zed-1.22.0-1
Total Download Size: 474.12 MiB
:: Retrieving packages...
firefox-157.0-1-x86_64 downloading...
zed-1.22.0-1-x86_64 downloading...
git-2.56.0-1-x86_64 downloading...
:: Processing package changes...
upgrading firefox...
upgrading git...
upgrading zed...
:: Running post-transaction hooks...
(1/2) Updating the desktop file MIME type cache...
(2/2) Performing snapper post snapshots...
"""
def _replay(text, names=("git", "firefox", "zed")):
"""Feed the log a growing file at a time, never the whole thing at once."""
import tempfile
from updater import live
directory = Path(tempfile.mkdtemp())
log = directory / "pacman.log"
log.write_bytes(b"")
observer = live.RepoObserver(list(names), log_path=log, cache_dir=directory / "cache")
data = text.encode()
for i in range(0, len(data), 40):
log.write_bytes(data[: i + 40])
observer.poll()
return observer
def test_pacman_progress_parses_the_expected_download_size():
"""pacman is one process, so /proc has nothing to attribute.
Without this the repository phase -- the long, network-bound half of an
update -- rendered an empty dashboard while hundreds of megabytes moved.
"""
assert _replay(PACMAN_OUT).expected_bytes == int(474.12 * 2**20)
def test_every_pacman_package_reaches_the_feed():
observer = _replay(PACMAN_OUT)
assert set(observer.completed) == {"git", "firefox", "zed"}, observer.completed
assert len(observer.feed) == 3
def test_pacman_packages_are_reported_in_the_order_they_finish():
observer = _replay(PACMAN_OUT)
# the download order this sample came from, which is what pacman reports
assert [e.name for e in observer.feed] == ["firefox", "zed", "git"], [
e.name for e in observer.feed
]
def test_a_package_outside_the_plan_is_ignored():
"""pacman may touch things the plan did not name."""
observer = _replay(PACMAN_OUT.replace("upgrading git...", "upgrading elsewhere..."))
assert "elsewhere" not in observer.completed
def test_pacman_hooks_are_shown_with_their_progress():
observer = _replay(PACMAN_OUT)
assert observer.phase == "hooks"
assert any("hook 2/2" in name for name in observer.states), observer.states
def test_a_half_written_log_line_is_left_alone():
"""The log is being written while it is read."""
import tempfile
from updater import live
directory = Path(tempfile.mkdtemp())
log = directory / "pacman.log"
log.write_bytes(b"upgrading fi") # no newline, no trailing dots
observer = live.RepoObserver(["firefox"], log_path=log, cache_dir=directory)
observer.poll()
assert observer.completed == set(), observer.completed
def test_a_missing_pacman_log_is_not_an_error():
import tempfile
from updater import live
directory = Path(tempfile.mkdtemp())
observer = live.RepoObserver(["git"], log_path=directory / "nope.log", cache_dir=directory)
assert observer.poll() == []
def test_an_unreadable_cache_directory_is_not_an_error():
import tempfile
from updater import live
directory = Path(tempfile.mkdtemp())
observer = live.RepoObserver(
["git"], log_path=directory / "x.log", cache_dir=directory / "missing"
)
assert observer._cache_bytes() == 0
def test_sizes_accept_the_short_and_long_unit_spellings():
from updater.live import _to_bytes
assert _to_bytes("1", "MiB") == _to_bytes("1", "M") == 2**20
assert _to_bytes("1", "GiB") == 2**30
assert _to_bytes("nonsense", "MiB") == 0
assert _to_bytes("5", "??") == 5
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())