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
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
|
#!/usr/bin/env python3
"""Helper methods concerned with executing subprocesses."""
import atexit
import logging
import os
import selectors
import shlex
import subprocess
import sys
from typing import List, Optional
logger = logging.getLogger(__file__)
def cmd_showing_output(
command: str,
) -> int:
"""Kick off a child process. Capture and print all output that it
produces on stdout and stderr. Wait for the subprocess to exit
and return the exit value as the return code of this function.
"""
line_enders = set([b'\n', b'\r'])
sel = selectors.DefaultSelector()
with subprocess.Popen(
command,
shell=True,
bufsize=0,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
universal_newlines=False,
) as p:
sel.register(p.stdout, selectors.EVENT_READ) # type: ignore
sel.register(p.stderr, selectors.EVENT_READ) # type: ignore
done = False
while not done:
for key, _ in sel.select():
char = key.fileobj.read(1) # type: ignore
if not char:
sel.unregister(key.fileobj)
if len(sel.get_map()) == 0:
sys.stdout.flush()
sys.stderr.flush()
sel.close()
done = True
if key.fileobj is p.stdout:
# sys.stdout.buffer.write(char)
os.write(sys.stdout.fileno(), char)
if char in line_enders:
sys.stdout.flush()
else:
# sys.stderr.buffer.write(char)
os.write(sys.stderr.fileno(), char)
if char in line_enders:
sys.stderr.flush()
p.wait()
return p.returncode
def cmd_with_timeout(command: str, timeout_seconds: Optional[float]) -> int:
"""Run a command but do not let it run for more than timeout seconds.
Doesn't capture or rebroadcast command output. Function returns
the exit value of the command or raises a TimeoutExpired exception
if the deadline is exceeded.
>>> cmd_with_timeout('/bin/echo foo', 10.0)
0
>>> cmd_with_timeout('/bin/sleep 2', 0.1)
Traceback (most recent call last):
...
subprocess.TimeoutExpired: Command '['/bin/bash', '-c', '/bin/sleep 2']' timed out after 0.1 seconds
"""
return subprocess.check_call(["/bin/bash", "-c", command], timeout=timeout_seconds)
def cmd(command: str, timeout_seconds: Optional[float] = None) -> str:
"""Run a command and capture its output to stdout (only) in a string.
Return that string as this function's output. Raises
subprocess.CalledProcessError or TimeoutExpired on error.
>>> cmd('/bin/echo foo')[:-1]
'foo'
>>> cmd('/bin/sleep 2', 0.1)
Traceback (most recent call last):
...
subprocess.TimeoutExpired: Command '/bin/sleep 2' timed out after 0.1 seconds
"""
ret = subprocess.run(
command,
shell=True,
capture_output=True,
check=True,
timeout=timeout_seconds,
).stdout
return ret.decode("utf-8")
def run_silently(command: str, timeout_seconds: Optional[float] = None) -> None:
"""Run a command silently but raise subprocess.CalledProcessError if
it fails.
>>> run_silently("/usr/bin/true")
>>> run_silently("/usr/bin/false")
Traceback (most recent call last):
...
subprocess.CalledProcessError: Command '/usr/bin/false' returned non-zero exit status 1.
"""
subprocess.run(
command,
shell=True,
stderr=subprocess.DEVNULL,
stdout=subprocess.DEVNULL,
capture_output=False,
check=True,
timeout=timeout_seconds,
)
def cmd_in_background(command: str, *, silent: bool = False) -> subprocess.Popen:
args = shlex.split(command)
if silent:
subproc = subprocess.Popen(
args,
stdin=subprocess.DEVNULL,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
else:
subproc = subprocess.Popen(args, stdin=subprocess.DEVNULL)
def kill_subproc() -> None:
try:
if subproc.poll() is None:
logger.info('At exit handler: killing %s (%s)', subproc, command)
subproc.terminate()
subproc.wait(timeout=10.0)
except BaseException as be:
logger.exception(be)
atexit.register(kill_subproc)
return subproc
def cmd_list(command: List[str]) -> str:
"""Run a command with args encapsulated in a list and return the
output text as a string. Raises subprocess.CalledProcessError.
"""
ret = subprocess.run(command, capture_output=True, check=True).stdout
return ret.decode("utf-8")
if __name__ == '__main__':
import doctest
doctest.testmod()
|
