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
|
import os
import shlex
import subprocess
import logging
def execute(
cmd,
wait=True,
ignore_errors=False,
include_stderr=False,
env=None,
return_exitcode=False,
shell=False,
):
"""Executes a commandline utility and returns its output
:param cmd: the command (as string) to execute, returns the program's output
:param wait: set to True to wait for command completion, False to return immediately, defaults to True
:param ignore_errors: set to True to return a string when an exception is thrown, otherwise might throw, defaults to False
:param include_stderr: set to True to include stderr output in the return value, defaults to False
:param env: provide a dict here to specify a custom execution environment, defaults to None
:param return_exitcode: set to True to return a pair, where the first member is the exit code and the message the second, defaults to False
:param shell: set to True to run command in a separate shell, defaults to False
:raises RuntimeError: the command either didn't exist or didn't exit cleanly, and ignore_errors was set to False
:return: output of cmd, or stderr, if ignore_errors is True and the command failed; or a tuple of exitcode and the previous, if return_exitcode is set to True
:rtype: string or tuple (if return_exitcode is set to True)
"""
args = cmd if shell else shlex.split(cmd)
logging.debug(cmd)
if not env:
env = os.environ.copy()
myenv = env.copy()
myenv["LC_ALL"] = "C"
if "WAYLAND_SOCKET" in myenv:
del myenv["WAYLAND_SOCKET"]
try:
proc = subprocess.Popen(
args,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT if include_stderr else subprocess.PIPE,
env=myenv,
shell=shell,
)
except FileNotFoundError as e:
raise RuntimeError("{} not found".format(cmd))
if wait:
timeout = 60
try:
out, _ = proc.communicate(timeout=timeout)
except subprocess.TimeoutExpired as e:
logging.warning(
f"""
Communication with process pid={proc.pid} hangs for more
than {timeout} seconds.
If this is not expected, the process is stale, or
you might have run in stdout / stderr deadlock.
"""
)
out, _ = proc.communicate()
if proc.returncode != 0:
err = "{} exited with code {}".format(cmd, proc.returncode)
logging.warning(err)
if ignore_errors:
return (proc.returncode, err) if return_exitcode else err
raise RuntimeError(err)
res = out.decode("utf-8")
logging.debug(res)
return (proc.returncode, res) if return_exitcode else res
return (0, "") if return_exitcode else ""
# vim: tabstop=8 expandtab shiftwidth=4 softtabstop=4
|
