From: Simon Glass <sjg@chromium.org>
Add a helper that runs a command with its stdout and stderr connected
to a PTY (so interactive programs flush before prompting) while the
child inherits stdin from the parent process. A reader thread copies
PTY output to fd 1 and accumulates it for the caller.
This is similar to cros_subprocess with PIPE_PTY, but cros_subprocess
always intercepts stdin which prevents the user from responding to
prompts directly. Letting Popen inherit stdin from fd 0 (rather than
referencing sys.stdin directly) keeps the helper usable inside
test runners that close sys.stdin in forked workers.
Add unit tests covering stdout capture, stderr-through-PTY capture,
the silent-failure return value, and the cwd argument. Register the
new TestRunInteractive class with both 'patman test' and the
u_boot_pylib test runner so the tests are picked up by either entry
point.
Signed-off-by: Simon Glass <sjg@chromium.org>
---
tools/patman/__main__.py | 3 +-
tools/u_boot_pylib/__main__.py | 4 ++-
tools/u_boot_pylib/command.py | 51 ++++++++++++++++++++++++++
tools/u_boot_pylib/test_command.py | 58 ++++++++++++++++++++++++++++++
4 files changed, 114 insertions(+), 2 deletions(-)
create mode 100644 tools/u_boot_pylib/test_command.py
@@ -40,13 +40,14 @@ def run_patman():
from patman import func_test
from patman import test_checkpatch
from patman import test_cseries
+ from u_boot_pylib import test_command
to_run = args.testname if args.testname not in [None, 'test'] else None
result = test_util.run_test_suites(
'patman', False, args.verbose, args.no_capture,
args.test_preserve_dirs, None, to_run, None,
[test_checkpatch.TestPatch, func_test.TestFunctional, 'settings',
- test_cseries.TestCseries])
+ test_cseries.TestCseries, test_command.TestRunInteractive])
sys.exit(0 if result.wasSuccessful() else 1)
# Process commits, produce patches files, check them, email them
@@ -29,13 +29,15 @@ def run_tests():
args = parser.parse_args()
from u_boot_pylib import test_claude
+ from u_boot_pylib import test_command
to_run = args.testname if args.testname not in [None, 'test'] else None
result = test_util.run_test_suites(
'u_boot_pylib', False, args.verbose, False,
False, None, to_run, None,
['u_boot_pylib.terminal', 'u_boot_pylib.gitutil',
- cros_subprocess.TestSubprocess, test_claude.TestClaude])
+ cros_subprocess.TestSubprocess, test_claude.TestClaude,
+ test_command.TestRunInteractive])
sys.exit(0 if result.wasSuccessful() else 1)
@@ -5,7 +5,10 @@ Shell command ease-ups for Python
Copyright (c) 2011 The Chromium OS Authors.
"""
+import os
+import pty
import subprocess
+import threading
from u_boot_pylib import cros_subprocess
@@ -160,6 +163,54 @@ def run_pipe(pipe_list, infile=None, outfile=None, capture=False,
return result
+def run_interactive(cmd, cwd=None):
+ """Run an interactive command with a PTY for correct output ordering
+
+ Similar to cros_subprocess.Popen with PIPE_PTY, but the child's stdin is
+ inherited from the parent process so the user can respond to prompts
+ directly. cros_subprocess always intercepts stdin which prevents interactive
+ use.
+
+ The child's stdout and stderr go through a PTY so interactive programs (like
+ git send-email) flush before prompting. A reader thread copies PTY output to
+ fd 1 and accumulates it for the caller.
+
+ Args:
+ cmd (list of str): Command to run
+ cwd (str or None): Working directory
+
+ Returns:
+ str: All output produced by the command
+ """
+ parent_fd, child_fd = pty.openpty()
+ captured = []
+
+ def reader():
+ """Drain the PTY: copy each chunk to fd 1 and remember it"""
+ try:
+ while True:
+ data = os.read(parent_fd, 4096)
+ if not data:
+ break
+ try:
+ os.write(1, data)
+ except OSError:
+ pass
+ captured.append(data)
+ except OSError:
+ pass
+
+ thr = threading.Thread(target=reader, daemon=True)
+ thr.start()
+
+ proc = subprocess.Popen(cmd, cwd=cwd, stdout=child_fd, stderr=child_fd)
+ os.close(child_fd)
+ proc.wait()
+ thr.join(timeout=2)
+ os.close(parent_fd)
+ return b''.join(captured).decode('utf-8', errors='replace')
+
+
def output(*cmd, **kwargs):
"""Run a command and return its output
new file mode 100644
@@ -0,0 +1,58 @@
+# SPDX-License-Identifier: GPL-2.0+
+# Copyright 2026 Simon Glass <sjg@chromium.org>
+
+"""Tests for the u_boot_pylib.command module"""
+
+import contextlib
+import os
+import unittest
+
+from u_boot_pylib import command
+
+
+@contextlib.contextmanager
+def _silence_fd1():
+ """Redirect raw fd 1 to /dev/null so PTY echo does not leak"""
+ saved = os.dup(1)
+ devnull = os.open(os.devnull, os.O_WRONLY)
+ try:
+ os.dup2(devnull, 1)
+ yield
+ finally:
+ os.dup2(saved, 1)
+ os.close(saved)
+ os.close(devnull)
+
+
+class TestRunInteractive(unittest.TestCase):
+ """Tests for command.run_interactive()"""
+
+ def test_captures_stdout(self):
+ """run_interactive() returns text written to stdout"""
+ with _silence_fd1():
+ out = command.run_interactive(['printf', 'hello'])
+ self.assertIn('hello', out)
+
+ def test_captures_stderr(self):
+ """run_interactive() also captures stderr through the PTY"""
+ with _silence_fd1():
+ out = command.run_interactive(
+ ['sh', '-c', 'printf out; printf err >&2'])
+ self.assertIn('out', out)
+ self.assertIn('err', out)
+
+ def test_silent_failing_command(self):
+ """run_interactive() returns empty for a silent failing command"""
+ with _silence_fd1():
+ out = command.run_interactive(['false'])
+ self.assertEqual('', out)
+
+ def test_cwd(self):
+ """run_interactive() honours the cwd argument"""
+ with _silence_fd1():
+ out = command.run_interactive(['pwd'], cwd='/tmp')
+ self.assertIn('/tmp', out)
+
+
+if __name__ == '__main__':
+ unittest.main()