Skip to content

Instantly share code, notes, and snippets.

@hayd
Last active April 24, 2021 04:33
Show Gist options
  • Save hayd/4f46a68fc697ba8888a7b517a414583e to your computer and use it in GitHub Desktop.
Save hayd/4f46a68fc697ba8888a7b517a414583e to your computer and use it in GitHub Desktop.
import os
import pty
import select
import subprocess
# Example:
# tty_capture(["python", "test.py"], b"abc\n")
# (b'stdin: True, stdout: True, stderr: True\r\n',
# b'read from stdin: abc\r\n')
def tty_capture(cmd, bytes_input):
"""Capture the output of cmd with bytes_input to stdin,
with stdin, stdout and stderr as TTYs."""
mo, so = pty.openpty() # provide tty to enable line-buffering
me, se = pty.openpty()
mi, si = pty.openpty()
fdmap = {mo: 'stdout', me: 'stderr', mi: 'stdin'}
p = subprocess.Popen(
cmd,
bufsize=1, stdin=si, stdout=so, stderr=se,
close_fds=True)
os.write(mi, bytes_input)
timeout = .04 # seconds
res = {'stdout': b'', 'stderr': b''}
while True:
ready, _, _ = select.select([mo, me], [], [], timeout)
if ready:
for fd in ready:
data = os.read(fd, 512)
if not data:
break
res[fdmap[fd]] += data
elif p.poll() is not None: # select timed-out
break # p exited
for fd in [si, so, se, mi, mo, me]:
os.close(fd) # can't do it sooner: it leads to errno.EIO error
p.wait()
return res['stdout'], res['stderr']
import sys
print("stdin: {}, stdout: {}, stderr: {}"
.format(sys.stdin.isatty(), sys.stdout.isatty(), sys.stderr.isatty()))
line = sys.stdin.readline()
sys.stderr.write("read from stdin: {}".format(line))
sys.stderr.flush()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment