9 from typing import List, Optional
12 logger = logging.getLogger(__file__)
15 def cmd_showing_output(command: str, ) -> int:
16 """Kick off a child process. Capture and print all output that it
17 produces on stdout and stderr. Wait for the subprocess to exit
18 and return the exit value as the return code of this function.
21 line_enders = set([b'\n', b'\r'])
26 stdout=subprocess.PIPE,
27 stderr=subprocess.PIPE,
28 universal_newlines=False,
30 sel = selectors.DefaultSelector()
31 sel.register(p.stdout, selectors.EVENT_READ)
32 sel.register(p.stderr, selectors.EVENT_READ)
34 while not should_exit:
35 for key, _ in sel.select():
36 char = key.fileobj.read(1)
39 if key.fileobj is p.stdout:
40 sys.stdout.buffer.write(char)
41 if char in line_enders:
44 sys.stderr.buffer.write(char)
45 if char in line_enders:
51 def cmd_with_timeout(command: str, timeout_seconds: Optional[float]) -> int:
52 """Run a command but do not let it run for more than timeout seconds.
53 Doesn't capture or rebroadcast command output. Function returns
54 the exit value of the command or raises a TimeoutExpired exception
55 if the deadline is exceeded.
57 >>> cmd_with_timeout('/bin/echo foo', 10.0)
60 >>> cmd_with_timeout('/bin/sleep 2', 0.1)
61 Traceback (most recent call last):
63 subprocess.TimeoutExpired: Command '['/bin/bash', '-c', '/bin/sleep 2']' timed out after 0.1 seconds
66 return subprocess.check_call(
67 ["/bin/bash", "-c", command], timeout=timeout_seconds
71 def cmd(command: str, timeout_seconds: Optional[float] = None) -> str:
72 """Run a command and capture its output to stdout (only) in a string.
73 Return that string as this function's output. Raises
74 subprocess.CalledProcessError or TimeoutExpired on error.
76 >>> cmd('/bin/echo foo')[:-1]
79 >>> cmd('/bin/sleep 2', 0.1)
80 Traceback (most recent call last):
82 subprocess.TimeoutExpired: Command '/bin/sleep 2' timed out after 0.1 seconds
90 timeout=timeout_seconds,
92 return ret.decode("utf-8")
95 def run_silently(command: str, timeout_seconds: Optional[float] = None) -> None:
96 """Run a command silently but raise subprocess.CalledProcessError if
99 >>> run_silently("/usr/bin/true")
101 >>> run_silently("/usr/bin/false")
102 Traceback (most recent call last):
104 subprocess.CalledProcessError: Command '/usr/bin/false' returned non-zero exit status 1.
110 stderr=subprocess.DEVNULL,
111 stdout=subprocess.DEVNULL,
112 capture_output=False,
114 timeout=timeout_seconds,
118 def cmd_in_background(
119 command: str, *, silent: bool = False
120 ) -> subprocess.Popen:
121 args = shlex.split(command)
123 subproc = subprocess.Popen(args,
124 stdin=subprocess.DEVNULL,
125 stdout=subprocess.DEVNULL,
126 stderr=subprocess.DEVNULL)
128 subproc = subprocess.Popen(args, stdin=subprocess.DEVNULL)
130 def kill_subproc() -> None:
132 if subproc.poll() is None:
133 logger.info("At exit handler: killing {}: {}".format(subproc, command))
135 subproc.wait(timeout=10.0)
136 except BaseException as be:
138 atexit.register(kill_subproc)
142 def cmd_list(command: List[str]) -> str:
143 """Run a command with args encapsulated in a list and return the
144 output text as a string. Raises subprocess.CalledProcessError.
146 ret = subprocess.run(command, capture_output=True, check=True).stdout
147 return ret.decode("utf-8")
150 if __name__ == '__main__':