From d6900a778a72ddb33b10550503719a13cc59bc18 Mon Sep 17 00:00:00 2001 From: Simon Glass Date: Mon, 3 Feb 2025 09:26:42 -0700 Subject: u_boot_pylib: Correct case for test_result This should be in capitals and defined at the start of the file. Update it. Signed-off-by: Simon Glass --- tools/u_boot_pylib/command.py | 23 +++++++++++------------ 1 file changed, 11 insertions(+), 12 deletions(-) (limited to 'tools/u_boot_pylib/command.py') diff --git a/tools/u_boot_pylib/command.py b/tools/u_boot_pylib/command.py index bbe95d86122..103358420dd 100644 --- a/tools/u_boot_pylib/command.py +++ b/tools/u_boot_pylib/command.py @@ -6,6 +6,13 @@ import os from u_boot_pylib import cros_subprocess +# 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 + """Shell command ease-ups for Python.""" class CommandResult: @@ -32,14 +39,6 @@ class CommandResult: self.combined = self.combined.decode('utf-8') 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, @@ -63,14 +62,14 @@ def run_pipe(pipe_list, infile=None, outfile=None, Returns: CommandResult object """ - 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 -- cgit v1.2.3 From 54ead4be04241f34967a6a591d529ee8ba66f301 Mon Sep 17 00:00:00 2001 From: Simon Glass Date: Mon, 3 Feb 2025 09:26:43 -0700 Subject: u_boot_pylib: Add an exception-class for errors Throwing an Exception is not very friendly since it is the top-level class of all exceptions. Declare a new class instead. Signed-off-by: Simon Glass --- tools/u_boot_pylib/command.py | 20 ++++++++++++++++++-- 1 file changed, 18 insertions(+), 2 deletions(-) (limited to 'tools/u_boot_pylib/command.py') diff --git a/tools/u_boot_pylib/command.py b/tools/u_boot_pylib/command.py index 103358420dd..4a9916bd814 100644 --- a/tools/u_boot_pylib/command.py +++ b/tools/u_boot_pylib/command.py @@ -13,6 +13,19 @@ from u_boot_pylib import cros_subprocess # 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 + + """Shell command ease-ups for Python.""" class CommandResult: @@ -61,6 +74,8 @@ def run_pipe(pipe_list, infile=None, outfile=None, 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__'): @@ -95,7 +110,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) @@ -106,7 +122,7 @@ 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): -- cgit v1.2.3 From f8456c91aad8259ab08bdf3654b8ee8c0187a45d Mon Sep 17 00:00:00 2001 From: Simon Glass Date: Mon, 3 Feb 2025 09:26:44 -0700 Subject: u_boot_pylib: Fix pylint warnings in command This file has a lot of warnings. Before adding any more features, fix those which are straightforward to resolve. Signed-off-by: Simon Glass --- tools/u_boot_pylib/command.py | 107 +++++++++++++++++++++++++++++++----------- 1 file changed, 80 insertions(+), 27 deletions(-) (limited to 'tools/u_boot_pylib/command.py') diff --git a/tools/u_boot_pylib/command.py b/tools/u_boot_pylib/command.py index 4a9916bd814..a98dcedd322 100644 --- a/tools/u_boot_pylib/command.py +++ b/tools/u_boot_pylib/command.py @@ -1,8 +1,11 @@ # 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 @@ -26,16 +29,16 @@ class CommandExc(Exception): self.result = result -"""Shell command ease-ups for Python.""" - 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): @@ -44,34 +47,46 @@ 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') self.combined = self.combined.decode('utf-8') return self -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: @@ -89,7 +104,7 @@ def run_pipe(pipe_list, infile=None, outfile=None, 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: @@ -125,28 +140,66 @@ def run_pipe(pipe_list, infile=None, outfile=None, 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_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 -- cgit v1.2.3 From 3d094ce28a22690c3d672988af5f161310822603 Mon Sep 17 00:00:00 2001 From: Simon Glass Date: Mon, 3 Feb 2025 09:26:45 -0700 Subject: u_boot_pylib: Add a function to run a single command Add a helper to avoid needing to use a list within a list for this simple case. Update existing users of runpipe() to use this where possible. Signed-off-by: Simon Glass --- tools/u_boot_pylib/command.py | 15 +++++++++++++++ 1 file changed, 15 insertions(+) (limited to 'tools/u_boot_pylib/command.py') diff --git a/tools/u_boot_pylib/command.py b/tools/u_boot_pylib/command.py index a98dcedd322..0e247355ef6 100644 --- a/tools/u_boot_pylib/command.py +++ b/tools/u_boot_pylib/command.py @@ -188,6 +188,21 @@ def run(*cmd, **kwargs): 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 -- cgit v1.2.3