u-boot/tools/patman/command.py
Simon Glass 82012dd284 patman: Provide a way to intercept commands for testing
Add a test point for the command module. This allows tests to emulate
the execution of commands. This provides more control (since we can make
the fake 'commands' do whatever we like), makes it faster to write tests
since we don't need to set up as much environment, and speeds up test
execution.

Signed-off-by: Simon Glass <sjg@chromium.org>
2014-09-09 16:38:27 -06:00

123 lines
4.1 KiB
Python

# Copyright (c) 2011 The Chromium OS Authors.
#
# SPDX-License-Identifier: GPL-2.0+
#
import os
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):
self.stdout = None
self.stderr = None
self.combined = None
self.return_code = None
self.exception = None
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
# 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 RunPipe() call.
# When this value is None, commands are executed as normal.
test_result = None
def RunPipe(pipe_list, infile=None, outfile=None,
capture=False, capture_stderr=False, oneline=False,
raise_on_error=True, cwd=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
kwargs: Additional keyword arguments to cros_subprocess.Popen()
Returns:
CommandResult object
"""
if test_result:
if hasattr(test_result, '__call__'):
return test_result(pipe_list=pipe_list)
return test_result
result = CommandResult()
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, err:
result.exception = err
if raise_on_error:
raise Exception("Error running '%s': %s" % (user_pipestr, str))
result.return_code = 255
return result
if capture:
result.stdout, result.stderr, result.combined = (
last_pipe.CommunicateFilter(None))
if result.stdout and oneline:
result.output = result.stdout.rstrip('\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
def Output(*cmd):
return RunPipe([cmd], capture=True, raise_on_error=False).stdout
def OutputOneLine(*cmd, **kwargs):
raise_on_error = kwargs.pop('raise_on_error', True)
return (RunPipe([cmd], capture=True, oneline=True,
raise_on_error=raise_on_error,
**kwargs).stdout.strip())
def Run(*cmd, **kwargs):
return RunPipe([cmd], **kwargs).stdout
def RunList(cmd):
return RunPipe([cmd], capture=True).stdout
def StopAll():
cros_subprocess.stay_alive = False