Improve run_command error diagnostics with live output capture
Some checks failed
Mark stable commit / test-unit (push) Has been cancelled
Mark stable commit / test-integration (push) Has been cancelled
Mark stable commit / test-env-virtual (push) Has been cancelled
Mark stable commit / test-env-nix (push) Has been cancelled
Mark stable commit / test-e2e (push) Has been cancelled
Mark stable commit / test-virgin-user (push) Has been cancelled
Mark stable commit / test-virgin-root (push) Has been cancelled
Mark stable commit / codesniffer-shellcheck (push) Has been cancelled
Mark stable commit / codesniffer-ruff (push) Has been cancelled
Mark stable commit / mark-stable (push) Has been cancelled

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
This commit is contained in:
Kevin Veen-Birkenbach
2025-12-13 14:29:53 +01:00
parent fb68b325d6
commit b8acd634f8
2 changed files with 132 additions and 18 deletions

View File

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