我有一个需要 tty 的可执行文件(如 stdin 和 stderr),并且希望能够测试它。 我想输入stdin,并捕获stdout和stderr的输出,这是一个示例脚本:
# test.py
import sys
print("stdin: {}".format(sys.stdin.isatty()))
print("stdout: {}".format(sys.stdout.isatty()))
print("stderr: {}".format(sys.stderr.isatty()))
sys.stdout.flush()
line = sys.stdin.readline()
sys.stderr.write("read from stdin: {}".format(line))
sys.stderr.flush()
.isatty
捕获并且每个都返回 False:
import subprocess
p = subprocess.Popen(["python", "test.py"], stdin=subprocess.PIPE, stderr=subprocess.PIPE, stdout=subprocess.PIPE)
p.stdin.write(b"abc\n")
print(p.communicate())
# (b'stdin: False\nstdout: False\nstderr: False\n', b'read from stdin: abc\n')
我想捕获 stdout 和 stderr 并让所有三个都返回 True - 作为 tty。
pty
来制作 tty 标准输入:
import subprocess
m, s = pty.openpty()
p = subprocess.Popen(["python", "test.py"], stdin=s, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
stdin = os.fdopen(m, 'wb', 0)
os.close(s)
stdin.write(b"abc\n")
(stdout, stderr) = p.communicate()
stdin.close()
print((stdout, stderr))
# (b'stdin: True\nstdout: False\nstderr: False\n', b'read from stdin: abc\n')
我尝试了很多排列来使 stdout 和 stderr tty 无济于事。
我想要的输出是:
(b'stdin: True\nstdout: True\nstderr: True\n', b'read from stdin: abc\n')
下面的代码基于jfs的答案here和here,加上你使用3个伪终端来区分stdout、stderr和stdin的想法(尽管注意有一个神秘警告,可能会出现问题(例如 OSX 上可能被截断的 stderr?)这样做)。
另请注意,从 Python 3.10 开始,文档说
pty
已在 Linux、macOS 和 FreeBSD 上进行测试,尽管它“应该适用于”其他 POSIX 平台:
import errno
import os
import pty
import select
import subprocess
def tty_capture(cmd, bytes_input):
"""Capture the output of cmd with bytes_input to stdin,
with stdin, stdout and stderr as TTYs.
Based on Andy Hayden's gist:
https://gist.github.com/hayd/4f46a68fc697ba8888a7b517a414583e
"""
mo, so = pty.openpty() # provide tty to enable line-buffering
me, se = pty.openpty()
mi, si = pty.openpty()
p = subprocess.Popen(
cmd,
bufsize=1, stdin=si, stdout=so, stderr=se,
close_fds=True)
for fd in [so, se, si]:
os.close(fd)
os.write(mi, bytes_input)
timeout = 0.04 # seconds
readable = [mo, me]
result = {mo: b'', me: b''}
try:
while readable:
ready, _, _ = select.select(readable, [], [], timeout)
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 [mo, me, mi]:
os.close(fd)
if p.poll() is None:
p.kill()
p.wait()
return result[mo], result[me]
out, err = tty_capture(["python", "test.py"], b"abc\n")
print((out, err))
产量
(b'stdin: True\r\nstdout: True\r\nstderr: True\r\n', b'read from stdin: abc\r\n')
readline
迭代器版本:
import errno
import os
import pty
import signal
import subprocess
def subprocess_tty(cmd, encoding="utf-8", **kwargs):
"""`subprocess.Popen` yielding stdout lines acting as a TTY"""
m, s = pty.openpty()
p = subprocess.Popen(cmd, stdout=s, stderr=s, **kwargs)
os.close(s)
try:
for line in open(m, encoding=encoding):
if not line: # EOF
break
yield line
except OSError as e:
if errno.EIO != e.errno: # EIO also means EOF
raise
finally:
if p.poll() is None:
p.send_signal(signal.SIGINT)
try:
p.wait(10)
except subprocess.TimeoutExpired:
p.terminate()
try:
p.wait(10)
except subprocess.TimeoutExpired:
p.kill()
p.wait()
使用示例:
import textwrap
for line in subprocess_tty(
[
"python",
"-c",
textwrap.dedent(
"""\
import sys
print(sys.stdin.isatty())
print(sys.stdout.isatty())
print(sys.stderr.isatty())
"""
),
]
):
print(f"{line!r}")