Lib/test/test_pty.py

   1 import errno
   2 import fcntl
   3 import pty
   4 import os
   5 import sys
   6 import signal
   7 from test.test_support import verbose, TestSkipped, run_unittest
   8 import unittest
   9
  10 TEST_STRING_1 = "I wish to buy a fish license.\n"
  11 TEST_STRING_2 = "For my pet fish, Eric.\n"
  12
  13 if verbose:
  14     def debug(msg):
  15         print msg
  16 else:
  17     def debug(msg):
  18         pass
  19
  20
  21 def normalize_output(data):
  22     # Some operating systems do conversions on newline.  We could possibly
  23     # fix that by doing the appropriate termios.tcsetattr()s.  I couldn't
  24     # figure out the right combo on Tru64 and I don't have an IRIX box.
  25     # So just normalize the output and doc the problem O/Ses by allowing
  26     # certain combinations for some platforms, but avoid allowing other
  27     # differences (like extra whitespace, trailing garbage, etc.)
  28
  29     # This is about the best we can do without getting some feedback
  30     # from someone more knowledgable.
  31
  32     # OSF/1 (Tru64) apparently turns \n into \r\r\n.
  33     if data.endswith('\r\r\n'):
  34         return data.replace('\r\r\n', '\n')
  35
  36     # IRIX apparently turns \n into \r\n.
  37     if data.endswith('\r\n'):
  38         return data.replace('\r\n', '\n')
  39
  40     return data
  41
  42
  43 # Marginal testing of pty suite. Cannot do extensive 'do or fail' testing
  44 # because pty code is not too portable.
  45 # XXX(nnorwitz):  these tests leak fds when there is an error.
  46 class PtyTest(unittest.TestCase):
  47     def setUp(self):
  48         # isatty() and close() can hang on some platforms.  Set an alarm
  49         # before running the test to make sure we don't hang forever.
  50         self.old_alarm = signal.signal(signal.SIGALRM, self.handle_sig)
  51         signal.alarm(10)
  52
  53     def tearDown(self):
  54         # remove alarm, restore old alarm handler
  55         signal.alarm(0)
  56         signal.signal(signal.SIGALRM, self.old_alarm)
  57
  58     def handle_sig(self, sig, frame):
  59         self.fail("isatty hung")
  60
  61     def test_basic(self):
  62         try:
  63             debug("Calling master_open()")
  64             master_fd, slave_name = pty.master_open()
  65             debug("Got master_fd '%d', slave_name '%s'" %
  66                   (master_fd, slave_name))
  67             debug("Calling slave_open(%r)" % (slave_name,))
  68             slave_fd = pty.slave_open(slave_name)
  69             debug("Got slave_fd '%d'" % slave_fd)
  70         except OSError:
  71             # " An optional feature could not be imported " ... ?
  72             raise TestSkipped, "Pseudo-terminals (seemingly) not functional."
  73
  74         self.assertTrue(os.isatty(slave_fd), 'slave_fd is not a tty')
  75
  76         # Solaris requires reading the fd before anything is returned.
  77         # My guess is that since we open and close the slave fd
  78         # in master_open(), we need to read the EOF.
  79
  80         # Ensure the fd is non-blocking in case there's nothing to read.
  81         orig_flags = fcntl.fcntl(master_fd, fcntl.F_GETFL)
  82         fcntl.fcntl(master_fd, fcntl.F_SETFL, orig_flags | os.O_NONBLOCK)
  83         try:
  84             s1 = os.read(master_fd, 1024)
  85             self.assertEquals('', s1)
  86         except OSError, e:
  87             if e.errno != errno.EAGAIN:
  88                 raise
  89         # Restore the original flags.
  90         fcntl.fcntl(master_fd, fcntl.F_SETFL, orig_flags)
  91
  92         debug("Writing to slave_fd")
  93         os.write(slave_fd, TEST_STRING_1)
  94         s1 = os.read(master_fd, 1024)
  95         self.assertEquals('I wish to buy a fish license.\n',
  96                           normalize_output(s1))
  97
  98         debug("Writing chunked output")
  99         os.write(slave_fd, TEST_STRING_2[:5])
 100         os.write(slave_fd, TEST_STRING_2[5:])
 101         s2 = os.read(master_fd, 1024)
 102         self.assertEquals('For my pet fish, Eric.\n', normalize_output(s2))
 103
 104         os.close(slave_fd)
 105         os.close(master_fd)
 106
 107
 108     def test_fork(self):
 109         debug("calling pty.fork()")
 110         pid, master_fd = pty.fork()
 111         if pid == pty.CHILD:
 112             # stdout should be connected to a tty.
 113             if not os.isatty(1):
 114                 debug("Child's fd 1 is not a tty?!")
 115                 os._exit(3)
 116
 117             # After pty.fork(), the child should already be a session leader.
 118             # (on those systems that have that concept.)
 119             debug("In child, calling os.setsid()")
 120             try:
 121                 os.setsid()
 122             except OSError:
 123                 # Good, we already were session leader
 124                 debug("Good: OSError was raised.")
 125                 pass
 126             except AttributeError:
 127                 # Have pty, but not setsid()?
 128                 debug("No setsid() available?")
 129                 pass
 130             except:
 131                 # We don't want this error to propagate, escaping the call to
 132                 # os._exit() and causing very peculiar behavior in the calling
 133                 # regrtest.py !
 134                 # Note: could add traceback printing here.
 135                 debug("An unexpected error was raised.")
 136                 os._exit(1)
 137             else:
 138                 debug("os.setsid() succeeded! (bad!)")
 139                 os._exit(2)
 140             os._exit(4)
 141         else:
 142             debug("Waiting for child (%d) to finish." % pid)
 143             # In verbose mode, we have to consume the debug output from the
 144             # child or the child will block, causing this test to hang in the
 145             # parent's waitpid() call.  The child blocks after a
 146             # platform-dependent amount of data is written to its fd.  On
 147             # Linux 2.6, it's 4000 bytes and the child won't block, but on OS
 148             # X even the small writes in the child above will block it.  Also
 149             # on Linux, the read() will throw an OSError (input/output error)
 150             # when it tries to read past the end of the buffer but the child's
 151             # already exited, so catch and discard those exceptions.  It's not
 152             # worth checking for EIO.
 153             while True:
 154                 try:
 155                     data = os.read(master_fd, 80)
 156                 except OSError:
 157                     break
 158                 if not data:
 159                     break
 160                 sys.stdout.write(data.replace('\r\n', '\n'))
 161
 162             ##line = os.read(master_fd, 80)
 163             ##lines = line.replace('\r\n', '\n').split('\n')
 164             ##if False and lines != ['In child, calling os.setsid()',
 165             ##             'Good: OSError was raised.', '']:
 166             ##    raise TestFailed("Unexpected output from child: %r" % line)
 167
 168             (pid, status) = os.waitpid(pid, 0)
 169             res = status >> 8
 170             debug("Child (%d) exited with status %d (%d)." % (pid, res, status))
 171             if res == 1:
 172                 self.fail("Child raised an unexpected exception in os.setsid()")
 173             elif res == 2:
 174                 self.fail("pty.fork() failed to make child a session leader.")
 175             elif res == 3:
 176                 self.fail("Child spawned by pty.fork() did not have a tty as stdout")
 177             elif res != 4:
 178                 self.fail("pty.fork() failed for unknown reasons.")
 179
 180             ##debug("Reading from master_fd now that the child has exited")
 181             ##try:
 182             ##    s1 = os.read(master_fd, 1024)
 183             ##except os.error:
 184             ##    pass
 185             ##else:
 186             ##    raise TestFailed("Read from master_fd did not raise exception")
 187
 188         os.close(master_fd)
 189
 190         # pty.fork() passed.
 191
 192 def test_main(verbose=None):
 193     run_unittest(PtyTest)
 194
 195 if __name__ == "__main__":
 196     test_main()