cpython/Lib/test/test_pty.py

427 lines
16 KiB
Python

from test.support import verbose, reap_children
from test.support.import_helper import import_module
# Skip these tests if termios is not available
import_module('termios')
import errno
import pty
import os
import sys
import select
import signal
import socket
import io # readline
import unittest
import struct
import tty
import fcntl
import warnings
TEST_STRING_1 = b"I wish to buy a fish license.\n"
TEST_STRING_2 = b"For my pet fish, Eric.\n"
try:
_TIOCGWINSZ = tty.TIOCGWINSZ
_TIOCSWINSZ = tty.TIOCSWINSZ
_HAVE_WINSZ = True
except AttributeError:
_HAVE_WINSZ = False
if verbose:
def debug(msg):
print(msg)
else:
def debug(msg):
pass
# Note that os.read() is nondeterministic so we need to be very careful
# to make the test suite deterministic. A normal call to os.read() may
# give us less than expected.
#
# Beware, on my Linux system, if I put 'foo\n' into a terminal fd, I get
# back 'foo\r\n' at the other end. The behavior depends on the termios
# setting. The newline translation may be OS-specific. To make the
# test suite deterministic and OS-independent, the functions _readline
# and normalize_output can be used.
def normalize_output(data):
# Some operating systems do conversions on newline. We could possibly fix
# that by doing the appropriate termios.tcsetattr()s. I couldn't figure out
# the right combo on Tru64. So, just normalize the output and doc the
# problem O/Ses by allowing certain combinations for some platforms, but
# avoid allowing other differences (like extra whitespace, trailing garbage,
# etc.)
# This is about the best we can do without getting some feedback
# from someone more knowledgable.
# OSF/1 (Tru64) apparently turns \n into \r\r\n.
if data.endswith(b'\r\r\n'):
return data.replace(b'\r\r\n', b'\n')
if data.endswith(b'\r\n'):
return data.replace(b'\r\n', b'\n')
return data
def _readline(fd):
"""Read one line. May block forever if no newline is read."""
reader = io.FileIO(fd, mode='rb', closefd=False)
return reader.readline()
def expectedFailureIfStdinIsTTY(fun):
# avoid isatty()
try:
tty.tcgetattr(pty.STDIN_FILENO)
return unittest.expectedFailure(fun)
except tty.error:
pass
return fun
def _get_term_winsz(fd):
s = struct.pack("HHHH", 0, 0, 0, 0)
return fcntl.ioctl(fd, _TIOCGWINSZ, s)
def _set_term_winsz(fd, winsz):
fcntl.ioctl(fd, _TIOCSWINSZ, winsz)
# Marginal testing of pty suite. Cannot do extensive 'do or fail' testing
# because pty code is not too portable.
class PtyTest(unittest.TestCase):
def setUp(self):
old_alarm = signal.signal(signal.SIGALRM, self.handle_sig)
self.addCleanup(signal.signal, signal.SIGALRM, old_alarm)
old_sighup = signal.signal(signal.SIGHUP, self.handle_sighup)
self.addCleanup(signal.signal, signal.SIGHUP, old_sighup)
# isatty() and close() can hang on some platforms. Set an alarm
# before running the test to make sure we don't hang forever.
self.addCleanup(signal.alarm, 0)
signal.alarm(10)
# Save original stdin window size
self.stdin_rows = None
self.stdin_cols = None
if _HAVE_WINSZ:
try:
stdin_dim = os.get_terminal_size(pty.STDIN_FILENO)
self.stdin_rows = stdin_dim.lines
self.stdin_cols = stdin_dim.columns
old_stdin_winsz = struct.pack("HHHH", self.stdin_rows,
self.stdin_cols, 0, 0)
self.addCleanup(_set_term_winsz, pty.STDIN_FILENO, old_stdin_winsz)
except OSError:
pass
def handle_sig(self, sig, frame):
self.fail("isatty hung")
@staticmethod
def handle_sighup(signum, frame):
# bpo-38547: if the process is the session leader, os.close(master_fd)
# of "master_fd, slave_name = pty.master_open()" raises SIGHUP
# signal: just ignore the signal.
#
# NOTE: the above comment is from an older version of the test;
# master_open() is not being used anymore.
pass
@expectedFailureIfStdinIsTTY
def test_openpty(self):
try:
mode = tty.tcgetattr(pty.STDIN_FILENO)
except tty.error:
# not a tty or bad/closed fd
debug("tty.tcgetattr(pty.STDIN_FILENO) failed")
mode = None
new_stdin_winsz = None
if self.stdin_rows != None and self.stdin_cols != None:
try:
# Modify pty.STDIN_FILENO window size; we need to
# check if pty.openpty() is able to set pty slave
# window size accordingly.
debug("Setting pty.STDIN_FILENO window size")
debug(f"original size: (rows={self.stdin_rows}, cols={self.stdin_cols})")
target_stdin_rows = self.stdin_rows + 1
target_stdin_cols = self.stdin_cols + 1
debug(f"target size: (rows={target_stdin_rows}, cols={target_stdin_cols})")
target_stdin_winsz = struct.pack("HHHH", target_stdin_rows,
target_stdin_cols, 0, 0)
_set_term_winsz(pty.STDIN_FILENO, target_stdin_winsz)
# Were we able to set the window size
# of pty.STDIN_FILENO successfully?
new_stdin_winsz = _get_term_winsz(pty.STDIN_FILENO)
self.assertEqual(new_stdin_winsz, target_stdin_winsz,
"pty.STDIN_FILENO window size unchanged")
except OSError:
warnings.warn("Failed to set pty.STDIN_FILENO window size")
pass
try:
debug("Calling pty.openpty()")
try:
master_fd, slave_fd = pty.openpty(mode, new_stdin_winsz)
except TypeError:
master_fd, slave_fd = pty.openpty()
debug(f"Got master_fd '{master_fd}', slave_fd '{slave_fd}'")
except OSError:
# " An optional feature could not be imported " ... ?
raise unittest.SkipTest("Pseudo-terminals (seemingly) not functional.")
# closing master_fd can raise a SIGHUP if the process is
# the session leader: we installed a SIGHUP signal handler
# to ignore this signal.
self.addCleanup(os.close, master_fd)
self.addCleanup(os.close, slave_fd)
self.assertTrue(os.isatty(slave_fd), "slave_fd is not a tty")
if mode:
self.assertEqual(tty.tcgetattr(slave_fd), mode,
"openpty() failed to set slave termios")
if new_stdin_winsz:
self.assertEqual(_get_term_winsz(slave_fd), new_stdin_winsz,
"openpty() failed to set slave window size")
# Solaris requires reading the fd before anything is returned.
# My guess is that since we open and close the slave fd
# in master_open(), we need to read the EOF.
#
# NOTE: the above comment is from an older version of the test;
# master_open() is not being used anymore.
# Ensure the fd is non-blocking in case there's nothing to read.
blocking = os.get_blocking(master_fd)
try:
os.set_blocking(master_fd, False)
try:
s1 = os.read(master_fd, 1024)
self.assertEqual(b'', s1)
except OSError as e:
if e.errno != errno.EAGAIN:
raise
finally:
# Restore the original flags.
os.set_blocking(master_fd, blocking)
debug("Writing to slave_fd")
os.write(slave_fd, TEST_STRING_1)
s1 = _readline(master_fd)
self.assertEqual(b'I wish to buy a fish license.\n',
normalize_output(s1))
debug("Writing chunked output")
os.write(slave_fd, TEST_STRING_2[:5])
os.write(slave_fd, TEST_STRING_2[5:])
s2 = _readline(master_fd)
self.assertEqual(b'For my pet fish, Eric.\n', normalize_output(s2))
def test_fork(self):
debug("calling pty.fork()")
pid, master_fd = pty.fork()
self.addCleanup(os.close, master_fd)
if pid == pty.CHILD:
# stdout should be connected to a tty.
if not os.isatty(1):
debug("Child's fd 1 is not a tty?!")
os._exit(3)
# After pty.fork(), the child should already be a session leader.
# (on those systems that have that concept.)
debug("In child, calling os.setsid()")
try:
os.setsid()
except OSError:
# Good, we already were session leader
debug("Good: OSError was raised.")
pass
except AttributeError:
# Have pty, but not setsid()?
debug("No setsid() available?")
pass
except:
# We don't want this error to propagate, escaping the call to
# os._exit() and causing very peculiar behavior in the calling
# regrtest.py !
# Note: could add traceback printing here.
debug("An unexpected error was raised.")
os._exit(1)
else:
debug("os.setsid() succeeded! (bad!)")
os._exit(2)
os._exit(4)
else:
debug("Waiting for child (%d) to finish." % pid)
# In verbose mode, we have to consume the debug output from the
# child or the child will block, causing this test to hang in the
# parent's waitpid() call. The child blocks after a
# platform-dependent amount of data is written to its fd. On
# Linux 2.6, it's 4000 bytes and the child won't block, but on OS
# X even the small writes in the child above will block it. Also
# on Linux, the read() will raise an OSError (input/output error)
# when it tries to read past the end of the buffer but the child's
# already exited, so catch and discard those exceptions. It's not
# worth checking for EIO.
while True:
try:
data = os.read(master_fd, 80)
except OSError:
break
if not data:
break
sys.stdout.write(str(data.replace(b'\r\n', b'\n'),
encoding='ascii'))
##line = os.read(master_fd, 80)
##lines = line.replace('\r\n', '\n').split('\n')
##if False and lines != ['In child, calling os.setsid()',
## 'Good: OSError was raised.', '']:
## raise TestFailed("Unexpected output from child: %r" % line)
(pid, status) = os.waitpid(pid, 0)
res = os.waitstatus_to_exitcode(status)
debug("Child (%d) exited with code %d (status %d)." % (pid, res, status))
if res == 1:
self.fail("Child raised an unexpected exception in os.setsid()")
elif res == 2:
self.fail("pty.fork() failed to make child a session leader.")
elif res == 3:
self.fail("Child spawned by pty.fork() did not have a tty as stdout")
elif res != 4:
self.fail("pty.fork() failed for unknown reasons.")
##debug("Reading from master_fd now that the child has exited")
##try:
## s1 = os.read(master_fd, 1024)
##except OSError:
## pass
##else:
## raise TestFailed("Read from master_fd did not raise exception")
def test_master_read(self):
# XXX(nnorwitz): this test leaks fds when there is an error.
debug("Calling pty.openpty()")
master_fd, slave_fd = pty.openpty()
debug(f"Got master_fd '{master_fd}', slave_fd '{slave_fd}'")
self.addCleanup(os.close, master_fd)
debug("Closing slave_fd")
os.close(slave_fd)
debug("Reading from master_fd")
try:
data = os.read(master_fd, 1)
except OSError: # Linux
data = b""
self.assertEqual(data, b"")
class SmallPtyTests(unittest.TestCase):
"""These tests don't spawn children or hang."""
def setUp(self):
self.orig_stdin_fileno = pty.STDIN_FILENO
self.orig_stdout_fileno = pty.STDOUT_FILENO
self.orig_pty_select = pty.select
self.fds = [] # A list of file descriptors to close.
self.files = []
self.select_rfds_lengths = []
self.select_rfds_results = []
def tearDown(self):
pty.STDIN_FILENO = self.orig_stdin_fileno
pty.STDOUT_FILENO = self.orig_stdout_fileno
pty.select = self.orig_pty_select
for file in self.files:
try:
file.close()
except OSError:
pass
for fd in self.fds:
try:
os.close(fd)
except OSError:
pass
def _pipe(self):
pipe_fds = os.pipe()
self.fds.extend(pipe_fds)
return pipe_fds
def _socketpair(self):
socketpair = socket.socketpair()
self.files.extend(socketpair)
return socketpair
def _mock_select(self, rfds, wfds, xfds, timeout=0):
# This will raise IndexError when no more expected calls exist.
# This ignores the timeout
self.assertEqual(self.select_rfds_lengths.pop(0), len(rfds))
return self.select_rfds_results.pop(0), [], []
def test__copy_to_each(self):
"""Test the normal data case on both master_fd and stdin."""
read_from_stdout_fd, mock_stdout_fd = self._pipe()
pty.STDOUT_FILENO = mock_stdout_fd
mock_stdin_fd, write_to_stdin_fd = self._pipe()
pty.STDIN_FILENO = mock_stdin_fd
socketpair = self._socketpair()
masters = [s.fileno() for s in socketpair]
# Feed data. Smaller than PIPEBUF. These writes will not block.
os.write(masters[1], b'from master')
os.write(write_to_stdin_fd, b'from stdin')
# Expect two select calls, the last one will cause IndexError
pty.select = self._mock_select
self.select_rfds_lengths.append(2)
self.select_rfds_results.append([mock_stdin_fd, masters[0]])
self.select_rfds_lengths.append(2)
with self.assertRaises(IndexError):
pty._copy(masters[0])
# Test that the right data went to the right places.
rfds = select.select([read_from_stdout_fd, masters[1]], [], [], 0)[0]
self.assertEqual([read_from_stdout_fd, masters[1]], rfds)
self.assertEqual(os.read(read_from_stdout_fd, 20), b'from master')
self.assertEqual(os.read(masters[1], 20), b'from stdin')
def test__copy_eof_on_all(self):
"""Test the empty read EOF case on both master_fd and stdin."""
read_from_stdout_fd, mock_stdout_fd = self._pipe()
pty.STDOUT_FILENO = mock_stdout_fd
mock_stdin_fd, write_to_stdin_fd = self._pipe()
pty.STDIN_FILENO = mock_stdin_fd
socketpair = self._socketpair()
masters = [s.fileno() for s in socketpair]
socketpair[1].close()
os.close(write_to_stdin_fd)
# Expect two select calls, the last one will cause IndexError
pty.select = self._mock_select
self.select_rfds_lengths.append(2)
self.select_rfds_results.append([mock_stdin_fd, masters[0]])
# We expect that both fds were removed from the fds list as they
# both encountered an EOF before the second select call.
self.select_rfds_lengths.append(0)
with self.assertRaises(IndexError):
pty._copy(masters[0])
def tearDownModule():
reap_children()
if __name__ == "__main__":
unittest.main()