Commit 9decc0dc authored by Georg Brandl's avatar Georg Brandl

Patch #1675471: convert test_pty to unittest.

parent 05c075d6
test_pty
I wish to buy a fish license.
For my pet fish, Eric.
import pty, os, sys, signal import pty
from test.test_support import verbose, TestFailed, TestSkipped import os
import signal
from test.test_support import verbose, TestSkipped, run_unittest
import unittest
TEST_STRING_1 = "I wish to buy a fish license.\n" TEST_STRING_1 = "I wish to buy a fish license.\n"
TEST_STRING_2 = "For my pet fish, Eric.\n" TEST_STRING_2 = "For my pet fish, Eric.\n"
...@@ -11,6 +14,7 @@ else: ...@@ -11,6 +14,7 @@ else:
def debug(msg): def debug(msg):
pass pass
def normalize_output(data): def normalize_output(data):
# Some operating systems do conversions on newline. We could possibly # Some operating systems do conversions on newline. We could possibly
# fix that by doing the appropriate termios.tcsetattr()s. I couldn't # fix that by doing the appropriate termios.tcsetattr()s. I couldn't
...@@ -32,116 +36,122 @@ def normalize_output(data): ...@@ -32,116 +36,122 @@ def normalize_output(data):
return data return data
# Marginal testing of pty suite. Cannot do extensive 'do or fail' testing # Marginal testing of pty suite. Cannot do extensive 'do or fail' testing
# because pty code is not too portable. # because pty code is not too portable.
class PtyTest(unittest.TestCase):
def test_basic_pty(): def setUp(self):
try: # isatty() and close() can hang on some platforms. Set an alarm
debug("Calling master_open()") # before running the test to make sure we don't hang forever.
master_fd, slave_name = pty.master_open() self.old_alarm = signal.signal(signal.SIGALRM, self.handle_sig)
debug("Got master_fd '%d', slave_name '%s'"%(master_fd, slave_name)) signal.alarm(10)
debug("Calling slave_open(%r)"%(slave_name,))
slave_fd = pty.slave_open(slave_name) def tearDown(self):
debug("Got slave_fd '%d'"%slave_fd) # remove alarm, restore old alarm handler
except OSError: signal.alarm(0)
# " An optional feature could not be imported " ... ? signal.signal(signal.SIGALRM, self.old_alarm)
raise TestSkipped, "Pseudo-terminals (seemingly) not functional."
def handle_sig(self, sig, frame):
if not os.isatty(slave_fd): self.fail("isatty hung")
raise TestFailed, "slave_fd is not a tty"
def test_basic(self):
debug("Writing to slave_fd") try:
os.write(slave_fd, TEST_STRING_1) debug("Calling master_open()")
s1 = os.read(master_fd, 1024) master_fd, slave_name = pty.master_open()
sys.stdout.write(normalize_output(s1)) debug("Got master_fd '%d', slave_name '%s'" %
(master_fd, slave_name))
debug("Writing chunked output") debug("Calling slave_open(%r)" % (slave_name,))
os.write(slave_fd, TEST_STRING_2[:5]) slave_fd = pty.slave_open(slave_name)
os.write(slave_fd, TEST_STRING_2[5:]) debug("Got slave_fd '%d'" % slave_fd)
s2 = os.read(master_fd, 1024) except OSError:
sys.stdout.write(normalize_output(s2)) # " An optional feature could not be imported " ... ?
raise TestSkipped, "Pseudo-terminals (seemingly) not functional."
os.close(slave_fd)
os.close(master_fd) self.assertTrue(os.isatty(slave_fd), 'slave_fd is not a tty')
def handle_sig(sig, frame): debug("Writing to slave_fd")
raise TestFailed, "isatty hung" os.write(slave_fd, TEST_STRING_1)
s1 = os.read(master_fd, 1024)
# isatty() and close() can hang on some platforms self.assertEquals('I wish to buy a fish license.\n',
# set an alarm before running the test to make sure we don't hang forever normalize_output(s1))
old_alarm = signal.signal(signal.SIGALRM, handle_sig)
signal.alarm(10) debug("Writing chunked output")
os.write(slave_fd, TEST_STRING_2[:5])
try: os.write(slave_fd, TEST_STRING_2[5:])
test_basic_pty() s2 = os.read(master_fd, 1024)
finally: self.assertEquals('For my pet fish, Eric.\n', normalize_output(s2))
# remove alarm, restore old alarm handler
signal.alarm(0) os.close(slave_fd)
signal.signal(signal.SIGALRM, old_alarm) os.close(master_fd)
# basic pty passed.
def test_fork(self):
debug("calling pty.fork()") debug("calling pty.fork()")
pid, master_fd = pty.fork() pid, master_fd = pty.fork()
if pid == pty.CHILD: if pid == pty.CHILD:
# stdout should be connected to a tty. # stdout should be connected to a tty.
if not os.isatty(1): if not os.isatty(1):
debug("Child's fd 1 is not a tty?!") debug("Child's fd 1 is not a tty?!")
os._exit(3) os._exit(3)
# After pty.fork(), the child should already be a session leader. # After pty.fork(), the child should already be a session leader.
# (on those systems that have that concept.) # (on those systems that have that concept.)
debug("In child, calling os.setsid()") debug("In child, calling os.setsid()")
try: try:
os.setsid() os.setsid()
except OSError: except OSError:
# Good, we already were session leader # Good, we already were session leader
debug("Good: OSError was raised.") debug("Good: OSError was raised.")
pass pass
except AttributeError: except AttributeError:
# Have pty, but not setsid() ? # Have pty, but not setsid()?
debug("No setsid() available ?") debug("No setsid() available?")
pass pass
except: except:
# We don't want this error to propagate, escaping the call to # We don't want this error to propagate, escaping the call to
# os._exit() and causing very peculiar behavior in the calling # os._exit() and causing very peculiar behavior in the calling
# regrtest.py ! # regrtest.py !
# Note: could add traceback printing here. # Note: could add traceback printing here.
debug("An unexpected error was raised.") debug("An unexpected error was raised.")
os._exit(1) os._exit(1)
else: else:
debug("os.setsid() succeeded! (bad!)") debug("os.setsid() succeeded! (bad!)")
os._exit(2) os._exit(2)
os._exit(4) os._exit(4)
else: else:
debug("Waiting for child (%d) to finish."%pid) debug("Waiting for child (%d) to finish." % pid)
##line = os.read(master_fd, 80) ##line = os.read(master_fd, 80)
##lines = line.replace('\r\n', '\n').split('\n') ##lines = line.replace('\r\n', '\n').split('\n')
##if False and lines != ['In child, calling os.setsid()', ##if False and lines != ['In child, calling os.setsid()',
## 'Good: OSError was raised.', '']: ## 'Good: OSError was raised.', '']:
## raise TestFailed("Unexpected output from child: %r" % line) ## raise TestFailed("Unexpected output from child: %r" % line)
(pid, status) = os.waitpid(pid, 0) (pid, status) = os.waitpid(pid, 0)
res = status >> 8 res = status >> 8
debug("Child (%d) exited with status %d (%d)."%(pid, res, status)) debug("Child (%d) exited with status %d (%d)." % (pid, res, status))
if res == 1: if res == 1:
raise TestFailed, "Child raised an unexpected exception in os.setsid()" self.fail("Child raised an unexpected exception in os.setsid()")
elif res == 2: elif res == 2:
raise TestFailed, "pty.fork() failed to make child a session leader." self.fail("pty.fork() failed to make child a session leader.")
elif res == 3: elif res == 3:
raise TestFailed, "Child spawned by pty.fork() did not have a tty as stdout" self.fail("Child spawned by pty.fork() did not have a tty as stdout")
elif res != 4: elif res != 4:
raise TestFailed, "pty.fork() failed for unknown reasons." self.fail("pty.fork() failed for unknown reasons.")
##debug("Reading from master_fd now that the child has exited") ##debug("Reading from master_fd now that the child has exited")
##try: ##try:
## s1 = os.read(master_fd, 1024) ## s1 = os.read(master_fd, 1024)
##except os.error: ##except os.error:
## pass ## pass
##else: ##else:
## raise TestFailed("Read from master_fd did not raise exception") ## raise TestFailed("Read from master_fd did not raise exception")
os.close(master_fd)
os.close(master_fd)
# pty.fork() passed.
# pty.fork() passed.
def test_main(verbose=None):
run_unittest(PtyTest)
if __name__ == "__main__":
test_main()
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment