aboutsummaryrefslogtreecommitdiff
path: root/tools/patman/command.py
diff options
context:
space:
mode:
Diffstat (limited to 'tools/patman/command.py')
-rw-r--r--tools/patman/command.py139
1 files changed, 0 insertions, 139 deletions
diff --git a/tools/patman/command.py b/tools/patman/command.py
deleted file mode 100644
index 92c453b5c1..0000000000
--- a/tools/patman/command.py
+++ /dev/null
@@ -1,139 +0,0 @@
-# SPDX-License-Identifier: GPL-2.0+
-# Copyright (c) 2011 The Chromium OS Authors.
-#
-
-import os
-
-from patman import cros_subprocess
-
-"""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
- """
- def __init__(self, stdout='', stderr='', combined='', return_code=0,
- exception=None):
- self.stdout = stdout
- self.stderr = stderr
- self.combined = combined
- self.return_code = return_code
- self.exception = exception
-
- def to_output(self, binary):
- 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
-
-
-# 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):
- """
- 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
- 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()
- Returns:
- CommandResult object
- """
- if test_result:
- if hasattr(test_result, '__call__'):
- # pylint: disable=E1102
- result = test_result(pipe_list=pipe_list)
- if result:
- return result
- else:
- 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])
- kwargs['stdout'] = None
- kwargs['stderr'] = None
- while pipeline:
- cmd = pipeline.pop(0)
- if last_pipe is not None:
- kwargs['stdin'] = last_pipe.stdout
- elif infile:
- kwargs['stdin'] = open(infile, 'rb')
- if pipeline or capture:
- kwargs['stdout'] = cros_subprocess.PIPE
- elif outfile:
- kwargs['stdout'] = open(outfile, 'wb')
- if capture_stderr:
- kwargs['stderr'] = cros_subprocess.PIPE
-
- try:
- last_pipe = cros_subprocess.Popen(cmd, cwd=cwd, **kwargs)
- except Exception as err:
- result.exception = err
- if raise_on_error:
- raise Exception("Error running '%s': %s" % (user_pipestr, str))
- result.return_code = 255
- return result.to_output(binary)
-
- if capture:
- result.stdout, result.stderr, result.combined = (
- last_pipe.communicate_filter(output_func))
- if result.stdout and oneline:
- result.output = result.stdout.rstrip(b'\r\n')
- result.return_code = last_pipe.wait()
- else:
- result.return_code = os.waitpid(last_pipe.pid, 0)[1]
- if raise_on_error and result.return_code:
- raise Exception("Error running '%s'" % user_pipestr)
- return result.to_output(binary)
-
-def output(*cmd, **kwargs):
- 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
-
- Returns:
- String containing output of command
- """
- 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()
- return result
-
-def run(*cmd, **kwargs):
- return run_pipe([cmd], **kwargs).stdout
-
-def run_list(cmd):
- return run_pipe([cmd], capture=True).stdout
-
-def stop_all():
- cros_subprocess.stay_alive = False