1
|
|
|
import subprocess |
2
|
|
|
import sys |
3
|
|
|
|
4
|
|
|
|
5
|
|
|
class CLIResult: |
6
|
|
|
"""Wrap the subprocess.CompletedProcess class to make it easier to use.""" |
7
|
|
|
|
8
|
|
|
def __init__(self, completed_process: subprocess.CompletedProcess): |
9
|
|
|
self._exit_code = int(completed_process.returncode) |
10
|
|
|
self._stdout = str(completed_process.stdout, encoding='utf-8') |
11
|
|
|
self._stderr = str(completed_process.stderr, encoding='utf-8') |
12
|
|
|
|
13
|
|
|
@property |
14
|
|
|
def exit_code(self) -> int: |
15
|
|
|
return self._exit_code |
16
|
|
|
|
17
|
|
|
@property |
18
|
|
|
def stdout(self) -> str: |
19
|
|
|
return self._stdout |
20
|
|
|
|
21
|
|
|
@property |
22
|
|
|
def stderr(self) -> str: |
23
|
|
|
return self._stderr |
24
|
|
|
|
25
|
|
|
|
26
|
|
|
SUBPROCESS_RUN_MAP = { |
27
|
|
|
True: { # legacy python <= 3.6 |
28
|
|
|
'stdout': subprocess.PIPE, # capture stdout and stderr separately |
29
|
|
|
'stderr': subprocess.PIPE, |
30
|
|
|
'check': True, |
31
|
|
|
}, |
32
|
|
|
False: { # python > 3.6 |
33
|
|
|
'capture_output': True, # capture stdout and stderr separately |
34
|
|
|
'check': True, |
35
|
|
|
}, |
36
|
|
|
} |
37
|
|
|
|
38
|
|
|
|
39
|
|
|
def execute_command_in_subprocess( |
40
|
|
|
executable: str, *cli_args, **subprocess_settings |
41
|
|
|
) -> CLIResult: |
42
|
|
|
"""Execute a command in a subprocess and return the result. |
43
|
|
|
|
44
|
|
|
The subprocess is implicitly not allowed to raise an exception in case the process exits |
45
|
|
|
with a non-zero exit code. |
46
|
|
|
It is left to the client to check the 'exit_code' property and decide how to handle it. |
47
|
|
|
|
48
|
|
|
Args: |
49
|
|
|
executable (str): path to executable program/binary (ie a CLI) |
50
|
|
|
*cli_args (str): arguments to pass to the executable |
51
|
|
|
**subprocess_settings: keyword arguments to pass to subprocess.run (check=False is always set) |
52
|
|
|
|
53
|
|
|
Returns: |
54
|
|
|
CLIResult: a wrapper around the subprocess.CompletedProcess class |
55
|
|
|
""" |
56
|
|
|
|
57
|
|
|
def subprocess_run() -> CLIResult: |
58
|
|
|
kwargs_dict = SUBPROCESS_RUN_MAP[sys.version_info < (3, 7)] |
59
|
|
|
completed_process = subprocess.run( # pylint: disable=W1510 |
60
|
|
|
[executable] + list(cli_args), |
61
|
|
|
**dict(dict(kwargs_dict, **subprocess_settings), check=False) |
62
|
|
|
) |
63
|
|
|
return CLIResult(completed_process) |
64
|
|
|
|
65
|
|
|
return subprocess_run() |
66
|
|
|
|