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(["/bin/bash", "-c", command], timeout=timeout_seconds)
74 def cmd(command: str, timeout_seconds: Optional[float] = None) -> str:
75 """Run a command and capture its output to stdout (only) in a string.
76 Return that string as this function's output. Raises
77 subprocess.CalledProcessError or TimeoutExpired on error.
79 >>> cmd('/bin/echo foo')[:-1]
82 >>> cmd('/bin/sleep 2', 0.1)
83 Traceback (most recent call last):
85 subprocess.TimeoutExpired: Command '/bin/sleep 2' timed out after 0.1 seconds
93 timeout=timeout_seconds,
95 return ret.decode("utf-8")
98 def run_silently(command: str, timeout_seconds: Optional[float] = None) -> None:
99 """Run a command silently but raise subprocess.CalledProcessError if
102 >>> run_silently("/usr/bin/true")
104 >>> run_silently("/usr/bin/false")
105 Traceback (most recent call last):
107 subprocess.CalledProcessError: Command '/usr/bin/false' returned non-zero exit status 1.
113 stderr=subprocess.DEVNULL,
114 stdout=subprocess.DEVNULL,
115 capture_output=False,
117 timeout=timeout_seconds,
121 def cmd_in_background(command: str, *, silent: bool = False) -> subprocess.Popen:
122 args = shlex.split(command)
124 subproc = subprocess.Popen(
126 stdin=subprocess.DEVNULL,
127 stdout=subprocess.DEVNULL,
128 stderr=subprocess.DEVNULL,
131 subproc = subprocess.Popen(args, stdin=subprocess.DEVNULL)
133 def kill_subproc() -> None:
135 if subproc.poll() is None:
136 logger.info("At exit handler: killing {}: {}".format(subproc, command))
138 subproc.wait(timeout=10.0)
139 except BaseException as be:
142 atexit.register(kill_subproc)
146 def cmd_list(command: List[str]) -> str:
147 """Run a command with args encapsulated in a list and return the
148 output text as a string. Raises subprocess.CalledProcessError.
150 ret = subprocess.run(command, capture_output=True, check=True).stdout
151 return ret.decode("utf-8")
154 if __name__ == '__main__':