+++ /dev/null
-from test.test_support import verbose, run_unittest, import_module\r
-\r
-#Skip these tests if either fcntl or termios is not available\r
-fcntl = import_module('fcntl')\r
-import_module('termios')\r
-\r
-import errno\r
-import pty\r
-import os\r
-import sys\r
-import signal\r
-import unittest\r
-\r
-TEST_STRING_1 = "I wish to buy a fish license.\n"\r
-TEST_STRING_2 = "For my pet fish, Eric.\n"\r
-\r
-if verbose:\r
- def debug(msg):\r
- print msg\r
-else:\r
- def debug(msg):\r
- pass\r
-\r
-\r
-def normalize_output(data):\r
- # Some operating systems do conversions on newline. We could possibly\r
- # fix that by doing the appropriate termios.tcsetattr()s. I couldn't\r
- # figure out the right combo on Tru64 and I don't have an IRIX box.\r
- # So just normalize the output and doc the problem O/Ses by allowing\r
- # certain combinations for some platforms, but avoid allowing other\r
- # differences (like extra whitespace, trailing garbage, etc.)\r
-\r
- # This is about the best we can do without getting some feedback\r
- # from someone more knowledgable.\r
-\r
- # OSF/1 (Tru64) apparently turns \n into \r\r\n.\r
- if data.endswith('\r\r\n'):\r
- return data.replace('\r\r\n', '\n')\r
-\r
- # IRIX apparently turns \n into \r\n.\r
- if data.endswith('\r\n'):\r
- return data.replace('\r\n', '\n')\r
-\r
- return data\r
-\r
-\r
-# Marginal testing of pty suite. Cannot do extensive 'do or fail' testing\r
-# because pty code is not too portable.\r
-# XXX(nnorwitz): these tests leak fds when there is an error.\r
-class PtyTest(unittest.TestCase):\r
- def setUp(self):\r
- # isatty() and close() can hang on some platforms. Set an alarm\r
- # before running the test to make sure we don't hang forever.\r
- self.old_alarm = signal.signal(signal.SIGALRM, self.handle_sig)\r
- signal.alarm(10)\r
-\r
- def tearDown(self):\r
- # remove alarm, restore old alarm handler\r
- signal.alarm(0)\r
- signal.signal(signal.SIGALRM, self.old_alarm)\r
-\r
- def handle_sig(self, sig, frame):\r
- self.fail("isatty hung")\r
-\r
- def test_basic(self):\r
- try:\r
- debug("Calling master_open()")\r
- master_fd, slave_name = pty.master_open()\r
- debug("Got master_fd '%d', slave_name '%s'" %\r
- (master_fd, slave_name))\r
- debug("Calling slave_open(%r)" % (slave_name,))\r
- slave_fd = pty.slave_open(slave_name)\r
- debug("Got slave_fd '%d'" % slave_fd)\r
- except OSError:\r
- # " An optional feature could not be imported " ... ?\r
- raise unittest.SkipTest, "Pseudo-terminals (seemingly) not functional."\r
-\r
- self.assertTrue(os.isatty(slave_fd), 'slave_fd is not a tty')\r
-\r
- # Solaris requires reading the fd before anything is returned.\r
- # My guess is that since we open and close the slave fd\r
- # in master_open(), we need to read the EOF.\r
-\r
- # Ensure the fd is non-blocking in case there's nothing to read.\r
- orig_flags = fcntl.fcntl(master_fd, fcntl.F_GETFL)\r
- fcntl.fcntl(master_fd, fcntl.F_SETFL, orig_flags | os.O_NONBLOCK)\r
- try:\r
- s1 = os.read(master_fd, 1024)\r
- self.assertEqual('', s1)\r
- except OSError, e:\r
- if e.errno != errno.EAGAIN:\r
- raise\r
- # Restore the original flags.\r
- fcntl.fcntl(master_fd, fcntl.F_SETFL, orig_flags)\r
-\r
- debug("Writing to slave_fd")\r
- os.write(slave_fd, TEST_STRING_1)\r
- s1 = os.read(master_fd, 1024)\r
- self.assertEqual('I wish to buy a fish license.\n',\r
- normalize_output(s1))\r
-\r
- debug("Writing chunked output")\r
- os.write(slave_fd, TEST_STRING_2[:5])\r
- os.write(slave_fd, TEST_STRING_2[5:])\r
- s2 = os.read(master_fd, 1024)\r
- self.assertEqual('For my pet fish, Eric.\n', normalize_output(s2))\r
-\r
- os.close(slave_fd)\r
- os.close(master_fd)\r
-\r
-\r
- def test_fork(self):\r
- debug("calling pty.fork()")\r
- pid, master_fd = pty.fork()\r
- if pid == pty.CHILD:\r
- # stdout should be connected to a tty.\r
- if not os.isatty(1):\r
- debug("Child's fd 1 is not a tty?!")\r
- os._exit(3)\r
-\r
- # After pty.fork(), the child should already be a session leader.\r
- # (on those systems that have that concept.)\r
- debug("In child, calling os.setsid()")\r
- try:\r
- os.setsid()\r
- except OSError:\r
- # Good, we already were session leader\r
- debug("Good: OSError was raised.")\r
- pass\r
- except AttributeError:\r
- # Have pty, but not setsid()?\r
- debug("No setsid() available?")\r
- pass\r
- except:\r
- # We don't want this error to propagate, escaping the call to\r
- # os._exit() and causing very peculiar behavior in the calling\r
- # regrtest.py !\r
- # Note: could add traceback printing here.\r
- debug("An unexpected error was raised.")\r
- os._exit(1)\r
- else:\r
- debug("os.setsid() succeeded! (bad!)")\r
- os._exit(2)\r
- os._exit(4)\r
- else:\r
- debug("Waiting for child (%d) to finish." % pid)\r
- # In verbose mode, we have to consume the debug output from the\r
- # child or the child will block, causing this test to hang in the\r
- # parent's waitpid() call. The child blocks after a\r
- # platform-dependent amount of data is written to its fd. On\r
- # Linux 2.6, it's 4000 bytes and the child won't block, but on OS\r
- # X even the small writes in the child above will block it. Also\r
- # on Linux, the read() will throw an OSError (input/output error)\r
- # when it tries to read past the end of the buffer but the child's\r
- # already exited, so catch and discard those exceptions. It's not\r
- # worth checking for EIO.\r
- while True:\r
- try:\r
- data = os.read(master_fd, 80)\r
- except OSError:\r
- break\r
- if not data:\r
- break\r
- sys.stdout.write(data.replace('\r\n', '\n'))\r
-\r
- ##line = os.read(master_fd, 80)\r
- ##lines = line.replace('\r\n', '\n').split('\n')\r
- ##if False and lines != ['In child, calling os.setsid()',\r
- ## 'Good: OSError was raised.', '']:\r
- ## raise TestFailed("Unexpected output from child: %r" % line)\r
-\r
- (pid, status) = os.waitpid(pid, 0)\r
- res = status >> 8\r
- debug("Child (%d) exited with status %d (%d)." % (pid, res, status))\r
- if res == 1:\r
- self.fail("Child raised an unexpected exception in os.setsid()")\r
- elif res == 2:\r
- self.fail("pty.fork() failed to make child a session leader.")\r
- elif res == 3:\r
- self.fail("Child spawned by pty.fork() did not have a tty as stdout")\r
- elif res != 4:\r
- self.fail("pty.fork() failed for unknown reasons.")\r
-\r
- ##debug("Reading from master_fd now that the child has exited")\r
- ##try:\r
- ## s1 = os.read(master_fd, 1024)\r
- ##except os.error:\r
- ## pass\r
- ##else:\r
- ## raise TestFailed("Read from master_fd did not raise exception")\r
-\r
- os.close(master_fd)\r
-\r
- # pty.fork() passed.\r
-\r
-def test_main(verbose=None):\r
- run_unittest(PtyTest)\r
-\r
-if __name__ == "__main__":\r
- test_main()\r
projects / mirror_edk2.git / blobdiff