1 files changed, 289 insertions, 0 deletions
diff --git a/lib/python2.7/test/test_pty.py b/lib/python2.7/test/test_pty.py
new file mode 100644
index 0000000..bec38c4
--- /dev/null
+++ b/lib/python2.7/test/test_pty.py
@@ -0,0 +1,289 @@
+from test.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 select
+import signal
+import socket
+import unittest
+
+TEST_STRING_1 = "I wish to buy a fish license.\n"
+TEST_STRING_2 = "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('\r\r\n'):
+        return data.replace('\r\r\n', '\n')
+
+    # IRIX apparently turns \n into \r\n.
+    if data.endswith('\r\n'):
+        return data.replace('\r\n', '\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.assertEqual('', s1)
+        except OSError, 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.assertEqual('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.assertEqual('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 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(data.replace('\r\n', '\n'))
+
+            ##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.
+
+
+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.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 fd in self.fds:
+            try:
+                os.close(fd)
+            except:
+                pass
+
+    def _pipe(self):
+        pipe_fds = os.pipe()
+        self.fds.extend(pipe_fds)
+        return pipe_fds
+
+    def _mock_select(self, rfds, wfds, xfds):
+        # This will raise IndexError when no more expected calls exist.
+        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 = socket.socketpair()
+        masters = [s.fileno() for s in socketpair]
+        self.fds.extend(masters)
+
+        # 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 = socket.socketpair()
+        masters = [s.fileno() for s in socketpair]
+        self.fds.extend(masters)
+
+        os.close(masters[1])
+        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 test_main(verbose=None):
+    run_unittest(SmallPtyTests, PtyTest)
+
+if __name__ == "__main__":
+    test_main()