Function run_with_tty

bpython/test/test_args.py:17–63

(command)

def run_with_tty(command):
    # based on https://stackoverflow.com/questions/52954248/capture-output-as-a-tty-in-python
    master_stdout, slave_stdout = pty.openpty()
    master_stderr, slave_stderr = pty.openpty()
    master_stdin, slave_stdin = pty.openpty()

    p = subprocess.Popen(
        command,
        stdout=slave_stdout,
        stderr=slave_stderr,
        stdin=slave_stdin,
        close_fds=True,
    )
    for fd in (slave_stdout, slave_stderr, slave_stdin):
        os.close(fd)

    readable = [master_stdout, master_stderr]
    result = {master_stdout: b"", master_stderr: b""}
    try:
        while readable:
            ready, _, _ = select.select(readable, [], [], 1)
            for fd in ready:
                try:
                    data = os.read(fd, 512)
                except OSError as e:
                    if e.errno != errno.EIO:
                        raise
                    # EIO means EOF on some systems
                    readable.remove(fd)
                else:
                    if not data:  # EOF
                        readable.remove(fd)
                    result[fd] += data
    finally:
        for fd in (master_stdout, master_stderr, master_stdin):
            os.close(fd)
        if p.poll() is None:
            p.kill()
        p.wait()

    if p.returncode:
        raise RuntimeError(f"Subprocess exited with {p.returncode}")

    return (
        result[master_stdout].decode(getpreferredencoding()),
        result[master_stderr].decode(getpreferredencoding()),
    )


class TestExecArgs(unittest.TestCase):