AppPkg/Applications/Python: Add Python 2.7.2 sources since the release of Python...

[mirror_edk2.git] / AppPkg / Applications / Python / Python-2.7.2 / Lib / test / test_poll.py

diff --git a/AppPkg/Applications/Python/Python-2.7.2/Lib/test/test_poll.py b/AppPkg/Applications/Python/Python-2.7.2/Lib/test/test_poll.py
new file mode 100644 (file)
index 0000000..f1b7c04
--- /dev/null
+++ b/AppPkg/Applications/Python/Python-2.7.2/Lib/test/test_poll.py
@@ -0,0 +1,157 @@
+# Test case for the os.poll() function\r
+\r
+import os, select, random, unittest\r
+from test.test_support import TESTFN, run_unittest\r
+\r
+try:\r
+    select.poll\r
+except AttributeError:\r
+    raise unittest.SkipTest, "select.poll not defined -- skipping test_poll"\r
+\r
+\r
+def find_ready_matching(ready, flag):\r
+    match = []\r
+    for fd, mode in ready:\r
+        if mode & flag:\r
+            match.append(fd)\r
+    return match\r
+\r
+class PollTests(unittest.TestCase):\r
+\r
+    def test_poll1(self):\r
+        # Basic functional test of poll object\r
+        # Create a bunch of pipe and test that poll works with them.\r
+\r
+        p = select.poll()\r
+\r
+        NUM_PIPES = 12\r
+        MSG = " This is a test."\r
+        MSG_LEN = len(MSG)\r
+        readers = []\r
+        writers = []\r
+        r2w = {}\r
+        w2r = {}\r
+\r
+        for i in range(NUM_PIPES):\r
+            rd, wr = os.pipe()\r
+            p.register(rd)\r
+            p.modify(rd, select.POLLIN)\r
+            p.register(wr, select.POLLOUT)\r
+            readers.append(rd)\r
+            writers.append(wr)\r
+            r2w[rd] = wr\r
+            w2r[wr] = rd\r
+\r
+        bufs = []\r
+\r
+        while writers:\r
+            ready = p.poll()\r
+            ready_writers = find_ready_matching(ready, select.POLLOUT)\r
+            if not ready_writers:\r
+                raise RuntimeError, "no pipes ready for writing"\r
+            wr = random.choice(ready_writers)\r
+            os.write(wr, MSG)\r
+\r
+            ready = p.poll()\r
+            ready_readers = find_ready_matching(ready, select.POLLIN)\r
+            if not ready_readers:\r
+                raise RuntimeError, "no pipes ready for reading"\r
+            rd = random.choice(ready_readers)\r
+            buf = os.read(rd, MSG_LEN)\r
+            self.assertEqual(len(buf), MSG_LEN)\r
+            bufs.append(buf)\r
+            os.close(r2w[rd]) ; os.close( rd )\r
+            p.unregister( r2w[rd] )\r
+            p.unregister( rd )\r
+            writers.remove(r2w[rd])\r
+\r
+        self.assertEqual(bufs, [MSG] * NUM_PIPES)\r
+\r
+    def poll_unit_tests(self):\r
+        # returns NVAL for invalid file descriptor\r
+        FD = 42\r
+        try:\r
+            os.close(FD)\r
+        except OSError:\r
+            pass\r
+        p = select.poll()\r
+        p.register(FD)\r
+        r = p.poll()\r
+        self.assertEqual(r[0], (FD, select.POLLNVAL))\r
+\r
+        f = open(TESTFN, 'w')\r
+        fd = f.fileno()\r
+        p = select.poll()\r
+        p.register(f)\r
+        r = p.poll()\r
+        self.assertEqual(r[0][0], fd)\r
+        f.close()\r
+        r = p.poll()\r
+        self.assertEqual(r[0], (fd, select.POLLNVAL))\r
+        os.unlink(TESTFN)\r
+\r
+        # type error for invalid arguments\r
+        p = select.poll()\r
+        self.assertRaises(TypeError, p.register, p)\r
+        self.assertRaises(TypeError, p.unregister, p)\r
+\r
+        # can't unregister non-existent object\r
+        p = select.poll()\r
+        self.assertRaises(KeyError, p.unregister, 3)\r
+\r
+        # Test error cases\r
+        pollster = select.poll()\r
+        class Nope:\r
+            pass\r
+\r
+        class Almost:\r
+            def fileno(self):\r
+                return 'fileno'\r
+\r
+        self.assertRaises(TypeError, pollster.register, Nope(), 0)\r
+        self.assertRaises(TypeError, pollster.register, Almost(), 0)\r
+\r
+    # Another test case for poll().  This is copied from the test case for\r
+    # select(), modified to use poll() instead.\r
+\r
+    def test_poll2(self):\r
+        cmd = 'for i in 0 1 2 3 4 5 6 7 8 9; do echo testing...; sleep 1; done'\r
+        p = os.popen(cmd, 'r')\r
+        pollster = select.poll()\r
+        pollster.register( p, select.POLLIN )\r
+        for tout in (0, 1000, 2000, 4000, 8000, 16000) + (-1,)*10:\r
+            fdlist = pollster.poll(tout)\r
+            if (fdlist == []):\r
+                continue\r
+            fd, flags = fdlist[0]\r
+            if flags & select.POLLHUP:\r
+                line = p.readline()\r
+                if line != "":\r
+                    self.fail('error: pipe seems to be closed, but still returns data')\r
+                continue\r
+\r
+            elif flags & select.POLLIN:\r
+                line = p.readline()\r
+                if not line:\r
+                    break\r
+                continue\r
+            else:\r
+                self.fail('Unexpected return value from select.poll: %s' % fdlist)\r
+        p.close()\r
+\r
+    def test_poll3(self):\r
+        # test int overflow\r
+        pollster = select.poll()\r
+        pollster.register(1)\r
+\r
+        self.assertRaises(OverflowError, pollster.poll, 1L << 64)\r
+\r
+        x = 2 + 3\r
+        if x != 5:\r
+            self.fail('Overflow must have occurred')\r
+\r
+def test_main():\r
+    run_unittest(PollTests)\r
+\r
+if __name__ == '__main__':\r
+    test_main()\r