AppPkg/Applications/Python/Python-2.7.2/Lib/test/test_pty.py

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