9 from typing import List, Optional
12 logger = logging.getLogger(__file__)
15 def cmd_showing_output(
18 """Kick off a child process. Capture and print all output that it
19 produces on stdout and stderr. Wait for the subprocess to exit
20 and return the exit value as the return code of this function.
23 line_enders = set([b'\n', b'\r'])
28 stdout=subprocess.PIPE,
29 stderr=subprocess.PIPE,
30 universal_newlines=False,
32 sel = selectors.DefaultSelector()
33 sel.register(p.stdout, selectors.EVENT_READ)
34 sel.register(p.stderr, selectors.EVENT_READ)
36 while stream_ends < 2:
37 for key, _ in sel.select():
38 char = key.fileobj.read(1)
42 if key.fileobj is p.stdout:
43 sys.stdout.buffer.write(char)
44 if char in line_enders:
47 sys.stderr.buffer.write(char)
48 if char in line_enders:
56 def cmd_with_timeout(command: str, timeout_seconds: Optional[float]) -> int:
57 """Run a command but do not let it run for more than timeout seconds.
58 Doesn't capture or rebroadcast command output. Function returns
59 the exit value of the command or raises a TimeoutExpired exception
60 if the deadline is exceeded.
62 >>> cmd_with_timeout('/bin/echo foo', 10.0)
65 >>> cmd_with_timeout('/bin/sleep 2', 0.1)
66 Traceback (most recent call last):
68 subprocess.TimeoutExpired: Command '['/bin/bash', '-c', '/bin/sleep 2']' timed out after 0.1 seconds
71 return subprocess.check_call(
72 ["/bin/bash", "-c", command], timeout=timeout_seconds
76 def cmd(command: str, timeout_seconds: Optional[float] = None) -> str:
77 """Run a command and capture its output to stdout (only) in a string.
78 Return that string as this function's output. Raises
79 subprocess.CalledProcessError or TimeoutExpired on error.
81 >>> cmd('/bin/echo foo')[:-1]
84 >>> cmd('/bin/sleep 2', 0.1)
85 Traceback (most recent call last):
87 subprocess.TimeoutExpired: Command '/bin/sleep 2' timed out after 0.1 seconds
95 timeout=timeout_seconds,
97 return ret.decode("utf-8")
100 def run_silently(command: str, timeout_seconds: Optional[float] = None) -> None:
101 """Run a command silently but raise subprocess.CalledProcessError if
104 >>> run_silently("/usr/bin/true")
106 >>> run_silently("/usr/bin/false")
107 Traceback (most recent call last):
109 subprocess.CalledProcessError: Command '/usr/bin/false' returned non-zero exit status 1.
115 stderr=subprocess.DEVNULL,
116 stdout=subprocess.DEVNULL,
117 capture_output=False,
119 timeout=timeout_seconds,
123 def cmd_in_background(
124 command: str, *, silent: bool = False
125 ) -> subprocess.Popen:
126 args = shlex.split(command)
128 subproc = subprocess.Popen(
130 stdin=subprocess.DEVNULL,
131 stdout=subprocess.DEVNULL,
132 stderr=subprocess.DEVNULL,
135 subproc = subprocess.Popen(args, stdin=subprocess.DEVNULL)
137 def kill_subproc() -> None:
139 if subproc.poll() is None:
141 "At exit handler: killing {}: {}".format(subproc, command)
144 subproc.wait(timeout=10.0)
145 except BaseException as be:
148 atexit.register(kill_subproc)
152 def cmd_list(command: List[str]) -> str:
153 """Run a command with args encapsulated in a list and return the
154 output text as a string. Raises subprocess.CalledProcessError.
156 ret = subprocess.run(command, capture_output=True, check=True).stdout
157 return ret.decode("utf-8")
160 if __name__ == '__main__':