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