From b8acd634f86e89de7c96b8dd75d95ee601267af4 Mon Sep 17 00:00:00 2001 From: Kevin Veen-Birkenbach Date: Sat, 13 Dec 2025 14:29:53 +0100 Subject: [PATCH] Improve run_command error diagnostics with live output capture Switch run_command to a single-run execution model that streams stdout/stderr live while capturing both streams in memory using selectors. This guarantees that command errors (e.g. make install, pip, nix) always show full diagnostics without re-running commands or risking deadlocks. Add unit tests for preview mode, success execution, failure handling, and allow_failure behavior. Context: https://chatgpt.com/share/replace-with-this-conversation-link --- src/pkgmgr/core/command/run.py | 103 +++++++++++++++++---- tests/unit/pkgmgr/core/command/test_run.py | 47 ++++++++++ 2 files changed, 132 insertions(+), 18 deletions(-) create mode 100644 tests/unit/pkgmgr/core/command/test_run.py diff --git a/src/pkgmgr/core/command/run.py b/src/pkgmgr/core/command/run.py index b7ba5e1..9c7ec53 100644 --- a/src/pkgmgr/core/command/run.py +++ b/src/pkgmgr/core/command/run.py @@ -1,8 +1,10 @@ +from __future__ import annotations + +import selectors import subprocess import sys from typing import List, Optional, Union - CommandType = Union[str, List[str]] @@ -13,32 +15,97 @@ def run_command( allow_failure: bool = False, ) -> subprocess.CompletedProcess: """ - Run a command and optionally exit on error. + Run a command with live output while capturing stdout/stderr. - - If `cmd` is a string, it is executed with `shell=True`. - - If `cmd` is a list of strings, it is executed without a shell. + - Output is streamed live to the terminal. + - Output is captured in memory. + - On failure, captured stdout/stderr are printed again so errors are never lost. + - Command is executed exactly once. """ - if isinstance(cmd, str): - display = cmd - else: - display = " ".join(cmd) - + display = cmd if isinstance(cmd, str) else " ".join(cmd) where = cwd or "." if preview: print(f"[Preview] In '{where}': {display}") - # Fake a successful result; most callers ignore the return value anyway return subprocess.CompletedProcess(cmd, 0) # type: ignore[arg-type] print(f"Running in '{where}': {display}") - if isinstance(cmd, str): - result = subprocess.run(cmd, cwd=cwd, shell=True) - else: - result = subprocess.run(cmd, cwd=cwd) + process = subprocess.Popen( + cmd, + cwd=cwd, + shell=isinstance(cmd, str), + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + bufsize=1, + ) - if result.returncode != 0 and not allow_failure: - print(f"Command failed with exit code {result.returncode}. Exiting.") - sys.exit(result.returncode) + assert process.stdout is not None + assert process.stderr is not None - return result + sel = selectors.DefaultSelector() + sel.register(process.stdout, selectors.EVENT_READ, data="stdout") + sel.register(process.stderr, selectors.EVENT_READ, data="stderr") + + stdout_lines: List[str] = [] + stderr_lines: List[str] = [] + + try: + while sel.get_map(): + for key, _ in sel.select(): + stream = key.fileobj + which = key.data + + line = stream.readline() + if line == "": + # EOF: stop watching this stream + try: + sel.unregister(stream) + except Exception: + pass + continue + + if which == "stdout": + stdout_lines.append(line) + print(line, end="") + else: + stderr_lines.append(line) + print(line, end="", file=sys.stderr) + finally: + # Ensure we don't leak FDs + try: + sel.close() + finally: + try: + process.stdout.close() + except Exception: + pass + try: + process.stderr.close() + except Exception: + pass + + returncode = process.wait() + + if returncode != 0 and not allow_failure: + print("\n[pkgmgr] Command failed, captured diagnostics:", file=sys.stderr) + print(f"[pkgmgr] Failed command: {display}", file=sys.stderr) + + if stdout_lines: + print("----- stdout -----") + print("".join(stdout_lines), end="") + + if stderr_lines: + print("----- stderr -----", file=sys.stderr) + print("".join(stderr_lines), end="", file=sys.stderr) + + print(f"Command failed with exit code {returncode}. Exiting.") + sys.exit(returncode) + + return subprocess.CompletedProcess( + cmd, + returncode, + stdout="".join(stdout_lines), + stderr="".join(stderr_lines), + ) diff --git a/tests/unit/pkgmgr/core/command/test_run.py b/tests/unit/pkgmgr/core/command/test_run.py new file mode 100644 index 0000000..38b69d4 --- /dev/null +++ b/tests/unit/pkgmgr/core/command/test_run.py @@ -0,0 +1,47 @@ +import unittest +from unittest.mock import patch + +import pkgmgr.core.command.run as run_mod + + +class TestRunCommand(unittest.TestCase): + def test_preview_returns_success_without_running(self) -> None: + with patch.object(run_mod.subprocess, "Popen") as popen_mock: + result = run_mod.run_command("echo hi", cwd="/tmp", preview=True) + self.assertEqual(result.returncode, 0) + popen_mock.assert_not_called() + + def test_success_streams_and_returns_completed_process(self) -> None: + cmd = ["python3", "-c", "print('out'); import sys; print('err', file=sys.stderr)"] + + with patch.object(run_mod.sys, "exit") as exit_mock: + result = run_mod.run_command(cmd, allow_failure=False) + + self.assertEqual(result.returncode, 0) + self.assertIn("out", result.stdout) + self.assertIn("err", result.stderr) + exit_mock.assert_not_called() + + def test_failure_exits_when_not_allowed(self) -> None: + cmd = ["python3", "-c", "import sys; print('oops', file=sys.stderr); sys.exit(2)"] + + with patch.object(run_mod.sys, "exit", side_effect=SystemExit(2)) as exit_mock: + with self.assertRaises(SystemExit) as ctx: + run_mod.run_command(cmd, allow_failure=False) + + self.assertEqual(ctx.exception.code, 2) + exit_mock.assert_called_once_with(2) + + def test_failure_does_not_exit_when_allowed(self) -> None: + cmd = ["python3", "-c", "import sys; print('oops', file=sys.stderr); sys.exit(3)"] + + with patch.object(run_mod.sys, "exit") as exit_mock: + result = run_mod.run_command(cmd, allow_failure=True) + + self.assertEqual(result.returncode, 3) + self.assertIn("oops", result.stderr) + exit_mock.assert_not_called() + + +if __name__ == "__main__": + unittest.main()