1# SPDX-License-Identifier: GPL-2.0+ 2# Copyright (c) 2011 The Chromium OS Authors. 3# 4 5import os 6import cros_subprocess 7 8"""Shell command ease-ups for Python.""" 9 10class 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. 40test_result = None 41 42def 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 result = test_result(pipe_list=pipe_list) 65 if result: 66 return result 67 else: 68 return test_result 69 # No result: fall through to normal processing 70 result = CommandResult() 71 last_pipe = None 72 pipeline = list(pipe_list) 73 user_pipestr = '|'.join([' '.join(pipe) for pipe in pipe_list]) 74 kwargs['stdout'] = None 75 kwargs['stderr'] = None 76 while pipeline: 77 cmd = pipeline.pop(0) 78 if last_pipe is not None: 79 kwargs['stdin'] = last_pipe.stdout 80 elif infile: 81 kwargs['stdin'] = open(infile, 'rb') 82 if pipeline or capture: 83 kwargs['stdout'] = cros_subprocess.PIPE 84 elif outfile: 85 kwargs['stdout'] = open(outfile, 'wb') 86 if capture_stderr: 87 kwargs['stderr'] = cros_subprocess.PIPE 88 89 try: 90 last_pipe = cros_subprocess.Popen(cmd, cwd=cwd, **kwargs) 91 except Exception as err: 92 result.exception = err 93 if raise_on_error: 94 raise Exception("Error running '%s': %s" % (user_pipestr, str)) 95 result.return_code = 255 96 return result 97 98 if capture: 99 result.stdout, result.stderr, result.combined = ( 100 last_pipe.CommunicateFilter(None)) 101 if result.stdout and oneline: 102 result.output = result.stdout.rstrip('\r\n') 103 result.return_code = last_pipe.wait() 104 else: 105 result.return_code = os.waitpid(last_pipe.pid, 0)[1] 106 if raise_on_error and result.return_code: 107 raise Exception("Error running '%s'" % user_pipestr) 108 return result 109 110def Output(*cmd, **kwargs): 111 raise_on_error = kwargs.get('raise_on_error', True) 112 return RunPipe([cmd], capture=True, raise_on_error=raise_on_error).stdout 113 114def OutputOneLine(*cmd, **kwargs): 115 raise_on_error = kwargs.pop('raise_on_error', True) 116 return (RunPipe([cmd], capture=True, oneline=True, 117 raise_on_error=raise_on_error, 118 **kwargs).stdout.strip()) 119 120def Run(*cmd, **kwargs): 121 return RunPipe([cmd], **kwargs).stdout 122 123def RunList(cmd): 124 return RunPipe([cmd], capture=True).stdout 125 126def StopAll(): 127 cros_subprocess.stay_alive = False 128