1 # SPDX-License-Identifier: GPL-2.0+ 2 # Copyright (c) 2011 The Chromium OS Authors. 3 # 4 5 import os 6 import cros_subprocess 7 8 """Shell command ease-ups for Python.""" 9 10 class CommandResult: 11 """A class which captures the result of executing a command. 12 13 Members: 14 stdout: stdout obtained from command, as a string 15 stderr: stderr obtained from command, as a string 16 return_code: Return code from command 17 exception: Exception received, or None if all ok 18 """ 19 def __init__(self): 20 self.stdout = None 21 self.stderr = None 22 self.combined = None 23 self.return_code = None 24 self.exception = None 25 26 def __init__(self, stdout='', stderr='', combined='', return_code=0, 27 exception=None): 28 self.stdout = stdout 29 self.stderr = stderr 30 self.combined = combined 31 self.return_code = return_code 32 self.exception = exception 33 34 35 # This permits interception of RunPipe for test purposes. If it is set to 36 # a function, then that function is called with the pipe list being 37 # executed. Otherwise, it is assumed to be a CommandResult object, and is 38 # returned as the result for every RunPipe() call. 39 # When this value is None, commands are executed as normal. 40 test_result = None 41 42 def RunPipe(pipe_list, infile=None, outfile=None, 43 capture=False, capture_stderr=False, oneline=False, 44 raise_on_error=True, cwd=None, **kwargs): 45 """ 46 Perform a command pipeline, with optional input/output filenames. 47 48 Args: 49 pipe_list: List of command lines to execute. Each command line is 50 piped into the next, and is itself a list of strings. For 51 example [ ['ls', '.git'] ['wc'] ] will pipe the output of 52 'ls .git' into 'wc'. 53 infile: File to provide stdin to the pipeline 54 outfile: File to store stdout 55 capture: True to capture output 56 capture_stderr: True to capture stderr 57 oneline: True to strip newline chars from output 58 kwargs: Additional keyword arguments to cros_subprocess.Popen() 59 Returns: 60 CommandResult object 61 """ 62 if test_result: 63 if hasattr(test_result, '__call__'): 64 return test_result(pipe_list=pipe_list) 65 return test_result 66 result = CommandResult() 67 last_pipe = None 68 pipeline = list(pipe_list) 69 user_pipestr = '|'.join([' '.join(pipe) for pipe in pipe_list]) 70 kwargs['stdout'] = None 71 kwargs['stderr'] = None 72 while pipeline: 73 cmd = pipeline.pop(0) 74 if last_pipe is not None: 75 kwargs['stdin'] = last_pipe.stdout 76 elif infile: 77 kwargs['stdin'] = open(infile, 'rb') 78 if pipeline or capture: 79 kwargs['stdout'] = cros_subprocess.PIPE 80 elif outfile: 81 kwargs['stdout'] = open(outfile, 'wb') 82 if capture_stderr: 83 kwargs['stderr'] = cros_subprocess.PIPE 84 85 try: 86 last_pipe = cros_subprocess.Popen(cmd, cwd=cwd, **kwargs) 87 except Exception as err: 88 result.exception = err 89 if raise_on_error: 90 raise Exception("Error running '%s': %s" % (user_pipestr, str)) 91 result.return_code = 255 92 return result 93 94 if capture: 95 result.stdout, result.stderr, result.combined = ( 96 last_pipe.CommunicateFilter(None)) 97 if result.stdout and oneline: 98 result.output = result.stdout.rstrip('\r\n') 99 result.return_code = last_pipe.wait() 100 else: 101 result.return_code = os.waitpid(last_pipe.pid, 0)[1] 102 if raise_on_error and result.return_code: 103 raise Exception("Error running '%s'" % user_pipestr) 104 return result 105 106 def Output(*cmd, **kwargs): 107 raise_on_error = kwargs.get('raise_on_error', True) 108 return RunPipe([cmd], capture=True, raise_on_error=raise_on_error).stdout 109 110 def OutputOneLine(*cmd, **kwargs): 111 raise_on_error = kwargs.pop('raise_on_error', True) 112 return (RunPipe([cmd], capture=True, oneline=True, 113 raise_on_error=raise_on_error, 114 **kwargs).stdout.strip()) 115 116 def Run(*cmd, **kwargs): 117 return RunPipe([cmd], **kwargs).stdout 118 119 def RunList(cmd): 120 return RunPipe([cmd], capture=True).stdout 121 122 def StopAll(): 123 cros_subprocess.stay_alive = False 124
