mirror of
https://github.com/python/cpython.git
synced 2025-10-31 10:26:02 +00:00
eb3615d9f6
svn+ssh://pythondev@svn.python.org/python/trunk ........ r71785 | r.david.murray | 2009-04-21 09:06:04 -0400 (Tue, 21 Apr 2009) | 4 lines Restore skips of posix and pty tests on Windows by calling the test_support.import_module on the appropriate modules before any other imports. ........
201 lines
7.4 KiB
Python
201 lines
7.4 KiB
Python
from test.support import verbose, run_unittest, import_module
|
|
|
|
#Skip these tests if either fcntl or termios is not available
|
|
fcntl = import_module('fcntl')
|
|
import_module('termios')
|
|
|
|
import errno
|
|
import pty
|
|
import os
|
|
import sys
|
|
import signal
|
|
import unittest
|
|
|
|
TEST_STRING_1 = b"I wish to buy a fish license.\n"
|
|
TEST_STRING_2 = b"For my pet fish, Eric.\n"
|
|
|
|
if verbose:
|
|
def debug(msg):
|
|
print(msg)
|
|
else:
|
|
def debug(msg):
|
|
pass
|
|
|
|
|
|
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 and I don't have an IRIX box.
|
|
# 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')
|
|
|
|
# IRIX apparently turns \n into \r\n.
|
|
if data.endswith(b'\r\n'):
|
|
return data.replace(b'\r\n', b'\n')
|
|
|
|
return data
|
|
|
|
|
|
# Marginal testing of pty suite. Cannot do extensive 'do or fail' testing
|
|
# because pty code is not too portable.
|
|
# XXX(nnorwitz): these tests leak fds when there is an error.
|
|
class PtyTest(unittest.TestCase):
|
|
def setUp(self):
|
|
# isatty() and close() can hang on some platforms. Set an alarm
|
|
# before running the test to make sure we don't hang forever.
|
|
self.old_alarm = signal.signal(signal.SIGALRM, self.handle_sig)
|
|
signal.alarm(10)
|
|
|
|
def tearDown(self):
|
|
# remove alarm, restore old alarm handler
|
|
signal.alarm(0)
|
|
signal.signal(signal.SIGALRM, self.old_alarm)
|
|
|
|
def handle_sig(self, sig, frame):
|
|
self.fail("isatty hung")
|
|
|
|
def test_basic(self):
|
|
try:
|
|
debug("Calling master_open()")
|
|
master_fd, slave_name = pty.master_open()
|
|
debug("Got master_fd '%d', slave_name '%s'" %
|
|
(master_fd, slave_name))
|
|
debug("Calling slave_open(%r)" % (slave_name,))
|
|
slave_fd = pty.slave_open(slave_name)
|
|
debug("Got slave_fd '%d'" % slave_fd)
|
|
except OSError:
|
|
# " An optional feature could not be imported " ... ?
|
|
raise unittest.SkipTest("Pseudo-terminals (seemingly) not functional.")
|
|
|
|
self.assertTrue(os.isatty(slave_fd), 'slave_fd is not a tty')
|
|
|
|
# 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.
|
|
|
|
# Ensure the fd is non-blocking in case there's nothing to read.
|
|
orig_flags = fcntl.fcntl(master_fd, fcntl.F_GETFL)
|
|
fcntl.fcntl(master_fd, fcntl.F_SETFL, orig_flags | os.O_NONBLOCK)
|
|
try:
|
|
s1 = os.read(master_fd, 1024)
|
|
self.assertEquals(b'', s1)
|
|
except OSError as e:
|
|
if e.errno != errno.EAGAIN:
|
|
raise
|
|
# Restore the original flags.
|
|
fcntl.fcntl(master_fd, fcntl.F_SETFL, orig_flags)
|
|
|
|
debug("Writing to slave_fd")
|
|
os.write(slave_fd, TEST_STRING_1)
|
|
s1 = os.read(master_fd, 1024)
|
|
self.assertEquals(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 = os.read(master_fd, 1024)
|
|
self.assertEquals(b'For my pet fish, Eric.\n', normalize_output(s2))
|
|
|
|
os.close(slave_fd)
|
|
os.close(master_fd)
|
|
|
|
|
|
def test_fork(self):
|
|
debug("calling pty.fork()")
|
|
pid, master_fd = pty.fork()
|
|
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 throw 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 = status >> 8
|
|
debug("Child (%d) exited with status %d (%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 os.error:
|
|
## pass
|
|
##else:
|
|
## raise TestFailed("Read from master_fd did not raise exception")
|
|
|
|
os.close(master_fd)
|
|
|
|
# pty.fork() passed.
|
|
|
|
def test_main(verbose=None):
|
|
run_unittest(PtyTest)
|
|
|
|
if __name__ == "__main__":
|
|
test_main()
|