diff options
author | Tom Rini <trini@konsulko.com> | 2025-04-07 16:40:02 -0600 |
---|---|---|
committer | Tom Rini <trini@konsulko.com> | 2025-04-08 11:43:23 -0600 |
commit | ff61d6bfd1c9534d3fc2397846a5899639f2e55d (patch) | |
tree | dcfe4bc52848a5637c975a3352b57885e5b8a06d /tools/u_boot_pylib/command.py | |
parent | 34820924edbc4ec7803eb89d9852f4b870fa760a (diff) | |
parent | f892a7f397a66d8d09f418d1e0e06dfb48bac27d (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.py | 161 |
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 |