(self, command: str, shell: str, env: Dict[str, Any])
| 1333 | ) |
| 1334 | |
| 1335 | def start(self, command: str, shell: str, env: Dict[str, Any]) -> None: |
| 1336 | if self.using_pty: |
| 1337 | if pty is None: # Encountered ImportError |
| 1338 | err = "You indicated pty=True, but your platform doesn't support the 'pty' module!" # noqa |
| 1339 | sys.exit(err) |
| 1340 | cols, rows = pty_size() |
| 1341 | self.pid, self.parent_fd = pty.fork() |
| 1342 | # If we're the child process, load up the actual command in a |
| 1343 | # shell, just as subprocess does; this replaces our process - whose |
| 1344 | # pipes are all hooked up to the PTY - with the "real" one. |
| 1345 | if self.pid == 0: |
| 1346 | # TODO: both pty.spawn() and pexpect.spawn() do a lot of |
| 1347 | # setup/teardown involving tty.setraw, getrlimit, signal. |
| 1348 | # Ostensibly we'll want some of that eventually, but if |
| 1349 | # possible write tests - integration-level if necessary - |
| 1350 | # before adding it! |
| 1351 | # |
| 1352 | # Set pty window size based on what our own controlling |
| 1353 | # terminal's window size appears to be. |
| 1354 | # TODO: make subroutine? |
| 1355 | winsize = struct.pack("HHHH", rows, cols, 0, 0) |
| 1356 | fcntl.ioctl(sys.stdout.fileno(), termios.TIOCSWINSZ, winsize) |
| 1357 | # Use execvpe for bare-minimum "exec w/ variable # args + env" |
| 1358 | # behavior. |
| 1359 | # NOTE: stdlib subprocess (actually its posix flavor, which is |
| 1360 | # written in C) uses either execve or execv, depending. |
| 1361 | os.execvpe(shell, [shell, "-c", command], env) |
| 1362 | else: |
| 1363 | self.process = Popen( |
| 1364 | command, |
| 1365 | shell=True, |
| 1366 | executable=shell, |
| 1367 | env=env, |
| 1368 | stdout=PIPE, |
| 1369 | stderr=PIPE, |
| 1370 | stdin=PIPE, |
| 1371 | ) |
| 1372 | |
| 1373 | def get_pid(self) -> int: |
| 1374 | return self.pid if self.using_pty else self.process.pid |
no test coverage detected