summaryrefslogtreecommitdiff
path: root/tools/u_boot_pylib/command.py
diff options
context:
space:
mode:
authorTom Rini <trini@konsulko.com>2025-04-07 16:40:02 -0600
committerTom Rini <trini@konsulko.com>2025-04-08 11:43:23 -0600
commitff61d6bfd1c9534d3fc2397846a5899639f2e55d (patch)
treedcfe4bc52848a5637c975a3352b57885e5b8a06d /tools/u_boot_pylib/command.py
parent34820924edbc4ec7803eb89d9852f4b870fa760a (diff)
parentf892a7f397a66d8d09f418d1e0e06dfb48bac27d (diff)
Merge branch 'next'
Note that this undoes the changes of commit cf6d4535cc4c ("x86: emulation: Disable bloblist for now") as that was intended only for the release due to time.
Diffstat (limited to 'tools/u_boot_pylib/command.py')
-rw-r--r--tools/u_boot_pylib/command.py161
1 files changed, 122 insertions, 39 deletions
diff --git a/tools/u_boot_pylib/command.py b/tools/u_boot_pylib/command.py
index bbe95d86122..0e247355ef6 100644
--- a/tools/u_boot_pylib/command.py
+++ b/tools/u_boot_pylib/command.py
@@ -1,21 +1,44 @@
# SPDX-License-Identifier: GPL-2.0+
-# Copyright (c) 2011 The Chromium OS Authors.
-#
+"""
+Shell command ease-ups for Python
-import os
+Copyright (c) 2011 The Chromium OS Authors.
+"""
+
+import subprocess
from u_boot_pylib import cros_subprocess
-"""Shell command ease-ups for Python."""
+# This permits interception of RunPipe for test purposes. If it is set to
+# a function, then that function is called with the pipe list being
+# executed. Otherwise, it is assumed to be a CommandResult object, and is
+# returned as the result for every run_pipe() call.
+# When this value is None, commands are executed as normal.
+TEST_RESULT = None
+
+
+class CommandExc(Exception):
+ """Reports an exception to the caller"""
+ def __init__(self, msg, result):
+ """Set up a new exception object
+
+ Args:
+ result (CommandResult): Execution result so far
+ """
+ super().__init__(msg)
+ self.result = result
+
class CommandResult:
"""A class which captures the result of executing a command.
Members:
- stdout: stdout obtained from command, as a string
- stderr: stderr obtained from command, as a string
- return_code: Return code from command
- exception: Exception received, or None if all ok
+ stdout (bytes): stdout obtained from command, as a string
+ stderr (bytes): stderr obtained from command, as a string
+ combined (bytes): stdout and stderr interleaved
+ return_code (int): Return code from command
+ exception (Exception): Exception received, or None if all ok
+ output (str or None): Returns output as a single line if requested
"""
def __init__(self, stdout='', stderr='', combined='', return_code=0,
exception=None):
@@ -24,8 +47,16 @@ class CommandResult:
self.combined = combined
self.return_code = return_code
self.exception = exception
+ self.output = None
def to_output(self, binary):
+ """Converts binary output to its final form
+
+ Args:
+ binary (bool): True to report binary output, False to use strings
+ Returns:
+ self
+ """
if not binary:
self.stdout = self.stdout.decode('utf-8')
self.stderr = self.stderr.decode('utf-8')
@@ -33,49 +64,47 @@ class CommandResult:
return self
-# This permits interception of RunPipe for test purposes. If it is set to
-# a function, then that function is called with the pipe list being
-# executed. Otherwise, it is assumed to be a CommandResult object, and is
-# returned as the result for every run_pipe() call.
-# When this value is None, commands are executed as normal.
-test_result = None
-
-def run_pipe(pipe_list, infile=None, outfile=None,
- capture=False, capture_stderr=False, oneline=False,
- raise_on_error=True, cwd=None, binary=False,
- output_func=None, **kwargs):
+def run_pipe(pipe_list, infile=None, outfile=None, capture=False,
+ capture_stderr=False, oneline=False, raise_on_error=True, cwd=None,
+ binary=False, output_func=None, **kwargs):
"""
Perform a command pipeline, with optional input/output filenames.
Args:
- pipe_list: List of command lines to execute. Each command line is
- piped into the next, and is itself a list of strings. For
+ pipe_list (list of list): List of command lines to execute. Each command
+ line is piped into the next, and is itself a list of strings. For
example [ ['ls', '.git'] ['wc'] ] will pipe the output of
'ls .git' into 'wc'.
- infile: File to provide stdin to the pipeline
- outfile: File to store stdout
- capture: True to capture output
- capture_stderr: True to capture stderr
- oneline: True to strip newline chars from output
- output_func: Output function to call with each output fragment
- (if it returns True the function terminates)
- kwargs: Additional keyword arguments to cros_subprocess.Popen()
+ infile (str): File to provide stdin to the pipeline
+ outfile (str): File to store stdout
+ capture (bool): True to capture output
+ capture_stderr (bool): True to capture stderr
+ oneline (bool): True to strip newline chars from output
+ raise_on_error (bool): True to raise on an error, False to return it in
+ the CommandResult
+ cwd (str or None): Directory to run the command in
+ binary (bool): True to report binary output, False to use strings
+ output_func (function): Output function to call with each output
+ fragment (if it returns True the function terminates)
+ **kwargs: Additional keyword arguments to cros_subprocess.Popen()
Returns:
CommandResult object
+ Raises:
+ CommandExc if an exception happens
"""
- if test_result:
- if hasattr(test_result, '__call__'):
+ if TEST_RESULT:
+ if hasattr(TEST_RESULT, '__call__'):
# pylint: disable=E1102
- result = test_result(pipe_list=pipe_list)
+ result = TEST_RESULT(pipe_list=pipe_list)
if result:
return result
else:
- return test_result
+ return TEST_RESULT
# No result: fall through to normal processing
result = CommandResult(b'', b'', b'')
last_pipe = None
pipeline = list(pipe_list)
- user_pipestr = '|'.join([' '.join(pipe) for pipe in pipe_list])
+ user_pipestr = '|'.join([' '.join(pipe) for pipe in pipe_list])
kwargs['stdout'] = None
kwargs['stderr'] = None
while pipeline:
@@ -96,7 +125,8 @@ def run_pipe(pipe_list, infile=None, outfile=None,
except Exception as err:
result.exception = err
if raise_on_error:
- raise Exception("Error running '%s': %s" % (user_pipestr, str))
+ raise CommandExc(f"Error running '{user_pipestr}': {err}",
+ result) from err
result.return_code = 255
return result.to_output(binary)
@@ -107,31 +137,84 @@ def run_pipe(pipe_list, infile=None, outfile=None,
result.output = result.stdout.rstrip(b'\r\n')
result.return_code = last_pipe.wait()
if raise_on_error and result.return_code:
- raise Exception("Error running '%s'" % user_pipestr)
+ raise CommandExc(f"Error running '{user_pipestr}'", result)
return result.to_output(binary)
+
def output(*cmd, **kwargs):
+ """Run a command and return its output
+
+ Args:
+ *cmd (list of str): Command to run
+ **kwargs (dict of args): Extra arguments to pass in
+
+ Returns:
+ str: command output
+ """
kwargs['raise_on_error'] = kwargs.get('raise_on_error', True)
return run_pipe([cmd], capture=True, **kwargs).stdout
+
def output_one_line(*cmd, **kwargs):
"""Run a command and output it as a single-line string
- The command us expected to produce a single line of output
+ The command is expected to produce a single line of output
+
+ Args:
+ *cmd (list of str): Command to run
+ **kwargs (dict of args): Extra arguments to pass in
Returns:
- String containing output of command
+ str: output of command with all newlines removed
"""
raise_on_error = kwargs.pop('raise_on_error', True)
result = run_pipe([cmd], capture=True, oneline=True,
- raise_on_error=raise_on_error, **kwargs).stdout.strip()
+ raise_on_error=raise_on_error, **kwargs).stdout.strip()
return result
+
def run(*cmd, **kwargs):
+ """Run a command
+
+ Note that you must add 'capture' to kwargs to obtain non-empty output
+
+ Args:
+ *cmd (list of str): Command to run
+ **kwargs (dict of args): Extra arguments to pass in
+
+ Returns:
+ str: output of command
+ """
return run_pipe([cmd], **kwargs).stdout
+
+def run_one(*cmd, **kwargs):
+ """Run a single command
+
+ Note that you must add 'capture' to kwargs to obtain non-empty output
+
+ Args:
+ *cmd (list of str): Command to run
+ **kwargs (dict of args): Extra arguments to pass in
+
+ Returns:
+ CommandResult: output of command
+ """
+ return run_pipe([cmd], **kwargs)
+
+
def run_list(cmd):
+ """Run a command and return its output
+
+ Args:
+ cmd (list of str): Command to run
+
+ Returns:
+ str: output of command
+ """
return run_pipe([cmd], capture=True).stdout
+
def stop_all():
+ """Stop all subprocesses initiated with cros_subprocess"""
cros_subprocess.stay_alive = False