"""The streaming apply.
The important properties are protocol ones: stdout carries nothing but JSON, the
events arrive in a usable order, and the exclude lists turn into the right argv.
Everything else is detail.
"""
from __future__ import annotations
import contextlib
import io
import json
import os
import stat
import sys
import tempfile
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from updater import repo, stream # noqa: E402
from updater.buildgraph import build_graph # noqa: E402
from updater.live import BuildObserver # noqa: E402
from updater.srcinfo import SrcInfo # noqa: E402
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
# --------------------------------------------------------------------------
# command construction
# --------------------------------------------------------------------------
def test_repo_command_is_syu_and_only_ignores_when_asked():
plain = stream.repo_command([])
assert plain == ["sudo", "pacman", "-Syu", "--noconfirm"], plain
assert not any(a.startswith("--ignore") for a in plain)
held = stream.repo_command(["mesa", "linux"])
assert "--ignore=linux,mesa" in held, held # sorted, so the command is stable
assert held[:4] == ["sudo", "pacman", "-Syu", "--noconfirm"]
def test_aur_command_names_targets_instead_of_using_sua():
"""`-Sua` updates everything and offers no way to leave one package out."""
argv = stream.aur_command(["b-git", "a-bin"])
assert "-Sua" not in argv
assert argv[:5] == ["paru", "-S", "--needed", "--noconfirm", "--nobatchinstall"]
assert argv[5:] == ["a-bin", "b-git"] # sorted
assert stream.aur_command([])[5:] == []
def test_aur_installs_each_package_rather_than_batching_them():
"""paru's default batches every package into one pacman transaction.
That makes one bad package poison the whole update: a single unresolved
conflict meant none of twenty-seven packages installed. Installing each
package as it is built costs one package per failure instead of all of them.
"""
assert "--nobatchinstall" in stream.aur_command(["x"])
assert "--batchinstall" not in stream.aur_command(["x"])
def test_aur_does_not_ask_paru_to_stop_at_the_first_failure():
"""--failfast would defeat the point of installing one at a time."""
assert "--failfast" not in stream.aur_command(["x"])
def test_the_interactive_fallback_keeps_the_prompts():
"""Some questions have no correct automatic answer.
"Replace the installed scenefx0.5 with scenefx?" is a decision, not a
default, so the only place it can be put is a terminal where paru can ask.
"""
batch = stream.aur_command(["mangowm-git"])
interactive = stream.aur_command_interactive(["mangowm-git"])
assert "--noconfirm" in batch
assert "--noconfirm" not in interactive
# same targets, same per-package behaviour, only the prompts differ
assert [a for a in batch if a != "--noconfirm"] == interactive
def test_flatpak_command_is_noninteractive():
argv = stream.flatpak_command()
assert argv[:2] == ["flatpak", "update"]
assert "-y" in argv and "--noninteractive" in argv
def test_installed_progress_reads_the_local_database():
"""The repo phase has no build graph, so progress has to be measured."""
with repo.connect() as handle:
plan = repo.plan(handle)
# an empty plan is legitimate on a fresh machine, so there is nothing to
# assert about its contents -- only about how they are measured
if plan.updates:
before = stream.installed_progress(plan.updates)
assert 0 <= before <= len(plan.updates)
else:
assert stream.installed_progress([]) == 0
# --------------------------------------------------------------------------
# the emitter
# --------------------------------------------------------------------------
def test_emitter_writes_one_json_object_per_line():
buf = io.StringIO()
emit = stream.Emitter(buf)
emit("start", phases=["repo"], total=3)
emit("build", packages=[], done=1, total=3)
lines = [l for l in buf.getvalue().splitlines() if l]
assert len(lines) == 2
for line in lines:
assert "\n" not in line
json.loads(line) # must parse
assert json.loads(lines[0]) == {"event": "start", "phases": ["repo"], "total": 3}
def test_emitter_keeps_human_output_off_stdout():
"""A stray print on stdout is a parse failure in the front end."""
import contextlib
buf = io.StringIO()
emit = stream.Emitter(buf)
emit("start", total=0)
# capture the real stderr so the test does not print into the suite output
err = io.StringIO()
with contextlib.redirect_stderr(err):
emit.human("this is for a person")
assert "this is for a person" not in buf.getvalue()
assert "this is for a person" in err.getvalue()
json.loads(buf.getvalue().strip())
def test_emitter_puts_nothing_else_on_the_wire():
buf = io.StringIO()
emit = stream.Emitter(buf)
for name in ("start", "phase_start", "build", "finished", "phase_end", "done"):
emit(name, x=1)
for line in buf.getvalue().splitlines():
payload = json.loads(line)
assert set(payload) == {"event", "x"}, payload
# --------------------------------------------------------------------------
# a real run
# --------------------------------------------------------------------------
def test_streaming_runner_against_a_real_child():
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 a launcher, it spawns a builder whose
# cwd is the package's clone dir
tool = fake_tool(bindir, "paru", f'( cd "{clone}/pkgA" && exec -a makepkg sleep 1.0 )')
graph = build_graph({"pkgA": SrcInfo(pkgbase="pkgA", pkgver="1", pkgrel="1")}, {"pkgA"})
observer = BuildObserver(graph, clone, total=1)
buf = io.StringIO()
runner = stream.StreamingRunner(observer, stream.Emitter(buf), poll_hz=20)
code = runner.run([str(tool)], "aur", root / "logs" / "paru.log")
assert code == 0
events = [json.loads(l) for l in buf.getvalue().splitlines() if l]
names = [e["event"] for e in events]
assert names[0] == "start" or names[0] == "phase_start"
assert names[0] == "phase_start"
assert names[-1] == "phase_end"
assert "build" in names
start = events[0]
assert start["phase"] == "aur" and start["total"] == 1
assert "paru" in start["command"]
finished = [e for e in events if e["event"] == "finished"]
assert [e["name"] for e in finished] == ["pkgA"], finished
assert finished[0]["elapsed"] > 0
assert finished[0]["state"] == "done"
end = events[-1]
assert end["phase"] == "aur" and end["returncode"] == 0
assert end["elapsed"] > 0
def test_streaming_runner_reports_a_failing_child():
with tempfile.TemporaryDirectory() as td:
root = Path(td)
clone = root / "clone"
clone.mkdir()
tool = fake_tool(root, "boom", "exit 3")
observer = BuildObserver(build_graph({}, set()), clone, total=0)
buf = io.StringIO()
runner = stream.StreamingRunner(observer, stream.Emitter(buf), poll_hz=20)
code = runner.run([str(tool)], "repo", root / "b.log")
assert code == 3
end = [json.loads(l) for l in buf.getvalue().splitlines()][-1]
assert end["event"] == "phase_end" and end["returncode"] == 3
def test_streaming_runner_keeps_the_child_output_in_a_log():
with tempfile.TemporaryDirectory() as td:
root = Path(td)
clone = root / "clone"
clone.mkdir()
tool = fake_tool(root, "noisy", "echo 'compiler wall of text'; sleep 0.1")
observer = BuildObserver(build_graph({}, set()), clone, total=0)
buf = io.StringIO()
log = root / "phase.log"
stream.StreamingRunner(observer, stream.Emitter(buf), poll_hz=20).run(
[str(tool)], "repo", log
)
assert "compiler wall of text" in log.read_text()
# The child's output now travels too, but only as a bounded tail: the
# dashboard saying nothing about what a build is doing is the reason it
# exists. Everything must still be one valid JSON object per line, and
# the tail must stay small however loud the command is.
events = [json.loads(line) for line in buf.getvalue().splitlines()]
tails = [e.get("log_lines", []) for e in events if e["event"] == "build"]
assert any("compiler wall of text" in line for tail in tails for line in tail), (
"the output tail never reached the interface"
)
for tail in tails:
assert len(tail) <= stream.StreamingRunner.TAIL_LINES, tail
def test_a_finished_package_is_emitted_exactly_once():
with tempfile.TemporaryDirectory() as td:
root = Path(td)
clone = root / "clone"
(clone / "pkgA").mkdir(parents=True)
tool = fake_tool(root, "quick", f'( cd "{clone}/pkgA" && exec -a makepkg sleep 0.4 )')
graph = build_graph({"pkgA": SrcInfo(pkgbase="pkgA", pkgver="1", pkgrel="1")}, {"pkgA"})
observer = BuildObserver(graph, clone, total=1)
buf = io.StringIO()
stream.StreamingRunner(observer, stream.Emitter(buf), poll_hz=30).run(
[str(tool)], "aur", root / "p.log"
)
events = [json.loads(l) for l in buf.getvalue().splitlines()]
finished = [e for e in events if e["event"] == "finished"]
assert len(finished) == 1, f"emitted {len(finished)} times: {finished}"
def test_every_emitted_line_parses_as_json_end_to_end():
"""The whole contract in one assertion: the reader would never choke."""
with tempfile.TemporaryDirectory() as td:
root = Path(td)
clone = root / "clone"
(clone / "pkgA").mkdir(parents=True)
tool = fake_tool(
root, "mixed", f'( cd "{clone}/pkgA" && exec -a makepkg sleep 0.3 )\necho stray text\n'
)
graph = build_graph({"pkgA": SrcInfo(pkgbase="pkgA", pkgver="1", pkgrel="1")}, {"pkgA"})
observer = BuildObserver(graph, clone, total=1)
buf = io.StringIO()
stream.StreamingRunner(observer, stream.Emitter(buf), poll_hz=30).run(
[str(tool)], "aur", root / "p.log"
)
for i, line in enumerate(buf.getvalue().splitlines()):
try:
json.loads(line)
except ValueError as exc:
raise AssertionError(f"line {i} is not JSON: {line!r} ({exc})") from exc
# --------------------------------------------------------------------------
# a dead reader must not kill the build
# --------------------------------------------------------------------------
def test_a_broken_pipe_does_not_kill_the_running_child():
"""The dangerous case.
If the front end exits, this process's stdout breaks. Letting that exception
escape unwinds the phase loop into the `finally` that terminates pacman,
killing an upgrade halfway through and leaving the system half-upgraded.
The child must run to completion regardless.
"""
import stat as statmod
with tempfile.TemporaryDirectory() as td:
root = Path(td)
clone = root / "clone"
clone.mkdir()
marker = root / "finished"
tool = root / "slow"
tool.write_text(f"#!/bin/bash\nsleep 0.6\ntouch {marker}\n")
tool.chmod(tool.stat().st_mode | statmod.S_IXUSR | statmod.S_IXUSR)
read_fd, write_fd = os.pipe()
writer = os.fdopen(write_fd, "w")
os.close(read_fd) # the reader is gone before the child even starts
observer = BuildObserver(build_graph({}, set()), clone, total=0)
runner = stream.StreamingRunner(observer, stream.Emitter(writer), poll_hz=30)
try:
code = runner.run([str(tool)], "repo", root / "p.log")
finally:
with contextlib.suppress(Exception):
writer.close()
assert code == 0, f"the runner reported {code}"
assert marker.is_file(), "the child was killed when the reader went away"
def test_the_emitter_survives_a_dead_stream():
read_fd, write_fd = os.pipe()
writer = os.fdopen(write_fd, "w")
os.close(read_fd)
emit = stream.Emitter(writer)
emit("start", total=1) # this one hits the broken pipe
emit("build", done=1) # and this must be a harmless no-op
assert emit.alive is False
with contextlib.suppress(Exception):
writer.close()
def test_the_emitter_survives_a_closed_stringio():
buf = io.StringIO()
emit = stream.Emitter(buf)
emit("start", total=0)
buf.close() # any later write raises ValueError
emit("build", done=0)
assert emit.alive is False, "a closed stream should be treated as a dead reader"
def test_a_dead_reader_leaves_the_log_intact():
"""The build output must still land on disk even with nobody watching."""
import stat as statmod
with tempfile.TemporaryDirectory() as td:
root = Path(td)
clone = root / "clone"
clone.mkdir()
tool = root / "noisy"
tool.write_text("#!/bin/bash\necho 'still being logged'\nsleep 0.3\n")
tool.chmod(tool.stat().st_mode | statmod.S_IXUSR)
read_fd, write_fd = os.pipe()
writer = os.fdopen(write_fd, "w")
os.close(read_fd)
log = root / "phase.log"
observer = BuildObserver(build_graph({}, set()), clone, total=0)
try:
stream.StreamingRunner(observer, stream.Emitter(writer), poll_hz=30).run(
[str(tool)], "repo", log
)
finally:
with contextlib.suppress(Exception):
writer.close()
assert "still being logged" in log.read_text()
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
def test_a_skip_request_stops_the_child_and_says_so():
"""One package per invocation is what makes skipping possible at all.
Inside a single transaction there is nothing to lift out without unpicking
what pacman has already been told, so the runner only ever abandons the
child it started and reports that it did.
"""
with tempfile.TemporaryDirectory() as td:
root = Path(td)
clone = root / "clone"
clone.mkdir()
tool = fake_tool(root, "slow", "sleep 30")
observer = BuildObserver(build_graph({}, set()), clone, total=0)
buf = io.StringIO()
runner = stream.StreamingRunner(observer, stream.Emitter(buf), poll_hz=20)
# ask before the loop's first poll, which is exactly what the signal
# handler does
stream.SKIP.ask()
code = runner.run([str(tool)], "aur", root / "slow.log")
assert runner.skipped is True, "the runner did not record the skip"
# The child was terminated, so its own code is negative. The caller
# reads the flag first and never the code, which is the point: a skip
# is not a failure however the child happened to die.
assert code != 0, code
assert stream.SKIP.take() is False, "the request was left standing"
def test_a_skip_request_is_consumed_once():
stream.SKIP.take() # whatever a previous test left
assert stream.SKIP.take() is False
stream.SKIP.ask()
assert stream.SKIP.take() is True
assert stream.SKIP.take() is False
def test_the_skip_signal_sets_the_request():
"""The handler may only touch simple state, which is why the request is a
module-level object rather than something threaded through."""
stream.SKIP.take()
stream.request_skip()
assert stream.SKIP.take() is True
def test_a_skip_covers_one_package_not_every_later_one():
"""The runner is reused for every invocation in a phase.
A flag left set after one skip made every package that followed look
skipped too, so a single keypress reported seven packages as skipped at the
reader's request.
"""
with tempfile.TemporaryDirectory() as td:
root = Path(td)
clone = root / "clone"
clone.mkdir()
quick = fake_tool(root, "quick", "true")
observer = BuildObserver(build_graph({}, set()), clone, total=0)
buf = io.StringIO()
runner = stream.StreamingRunner(observer, stream.Emitter(buf), poll_hz=20)
stream.SKIP.ask()
stream.SKIP.take() # whatever a previous test left standing
stream.SKIP.ask()
slow = fake_tool(root, "slow", "sleep 30")
runner.run([str(slow)], "aur", root / "one.log")
assert runner.skipped is True, "the first skip was not recorded"
# the same runner, the next package: nothing was asked for this time
code = runner.run([str(quick)], "aur", root / "two.log")
assert runner.skipped is False, "the skip leaked into the next package"
assert code == 0, code
if __name__ == "__main__":
raise SystemExit(main())