aboutsummaryrefslogtreecommitdiff
path: root/tools/patman/command.py
blob: 92c453b5c1374b7a5c9f1c8fb988dda25e9796e3 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
# 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