From 4710c53dcad1ebf3755f3efb9e80ac24bd72a9b2 Mon Sep 17 00:00:00 2001 From: darylm503 Date: Mon, 16 Apr 2012 22:12:42 +0000 Subject: AppPkg/Applications/Python: Add Python 2.7.2 sources since the release of Python 2.7.3 made them unavailable from the python.org web site. These files are a subset of the python-2.7.2.tgz distribution from python.org. Changed files from PyMod-2.7.2 have been copied into the corresponding directories of this tree, replacing the original files in the distribution. Signed-off-by: daryl.mcdaniel@intel.com git-svn-id: https://edk2.svn.sourceforge.net/svnroot/edk2/trunk/edk2@13197 6f19259b-4bc3-4df7-8a09-765794883524 --- .../Python/Python-2.7.2/Lib/test/test_kqueue.py | 196 +++++++++++++++++++++ 1 file changed, 196 insertions(+) create mode 100644 AppPkg/Applications/Python/Python-2.7.2/Lib/test/test_kqueue.py (limited to 'AppPkg/Applications/Python/Python-2.7.2/Lib/test/test_kqueue.py') diff --git a/AppPkg/Applications/Python/Python-2.7.2/Lib/test/test_kqueue.py b/AppPkg/Applications/Python/Python-2.7.2/Lib/test/test_kqueue.py new file mode 100644 index 0000000000..a76c07caa6 --- /dev/null +++ b/AppPkg/Applications/Python/Python-2.7.2/Lib/test/test_kqueue.py @@ -0,0 +1,196 @@ +""" +Tests for kqueue wrapper. +""" +import socket +import errno +import time +import select +import sys +import unittest + +from test import test_support +if not hasattr(select, "kqueue"): + raise unittest.SkipTest("test works only on BSD") + +class TestKQueue(unittest.TestCase): + def test_create_queue(self): + kq = select.kqueue() + self.assertTrue(kq.fileno() > 0, kq.fileno()) + self.assertTrue(not kq.closed) + kq.close() + self.assertTrue(kq.closed) + self.assertRaises(ValueError, kq.fileno) + + def test_create_event(self): + fd = sys.stderr.fileno() + ev = select.kevent(fd) + other = select.kevent(1000) + self.assertEqual(ev.ident, fd) + self.assertEqual(ev.filter, select.KQ_FILTER_READ) + self.assertEqual(ev.flags, select.KQ_EV_ADD) + self.assertEqual(ev.fflags, 0) + self.assertEqual(ev.data, 0) + self.assertEqual(ev.udata, 0) + self.assertEqual(ev, ev) + self.assertNotEqual(ev, other) + self.assertEqual(cmp(ev, other), -1) + self.assertTrue(ev < other) + self.assertTrue(other >= ev) + self.assertRaises(TypeError, cmp, ev, None) + self.assertRaises(TypeError, cmp, ev, 1) + self.assertRaises(TypeError, cmp, ev, "ev") + + ev = select.kevent(fd, select.KQ_FILTER_WRITE) + self.assertEqual(ev.ident, fd) + self.assertEqual(ev.filter, select.KQ_FILTER_WRITE) + self.assertEqual(ev.flags, select.KQ_EV_ADD) + self.assertEqual(ev.fflags, 0) + self.assertEqual(ev.data, 0) + self.assertEqual(ev.udata, 0) + self.assertEqual(ev, ev) + self.assertNotEqual(ev, other) + + ev = select.kevent(fd, select.KQ_FILTER_WRITE, select.KQ_EV_ONESHOT) + self.assertEqual(ev.ident, fd) + self.assertEqual(ev.filter, select.KQ_FILTER_WRITE) + self.assertEqual(ev.flags, select.KQ_EV_ONESHOT) + self.assertEqual(ev.fflags, 0) + self.assertEqual(ev.data, 0) + self.assertEqual(ev.udata, 0) + self.assertEqual(ev, ev) + self.assertNotEqual(ev, other) + + ev = select.kevent(1, 2, 3, 4, 5, 6) + self.assertEqual(ev.ident, 1) + self.assertEqual(ev.filter, 2) + self.assertEqual(ev.flags, 3) + self.assertEqual(ev.fflags, 4) + self.assertEqual(ev.data, 5) + self.assertEqual(ev.udata, 6) + self.assertEqual(ev, ev) + self.assertNotEqual(ev, other) + + bignum = sys.maxsize * 2 + 1 + ev = select.kevent(bignum, 1, 2, 3, sys.maxsize, bignum) + self.assertEqual(ev.ident, bignum) + self.assertEqual(ev.filter, 1) + self.assertEqual(ev.flags, 2) + self.assertEqual(ev.fflags, 3) + self.assertEqual(ev.data, sys.maxsize) + self.assertEqual(ev.udata, bignum) + self.assertEqual(ev, ev) + self.assertNotEqual(ev, other) + + def test_queue_event(self): + serverSocket = socket.socket() + serverSocket.bind(('127.0.0.1', 0)) + serverSocket.listen(1) + client = socket.socket() + client.setblocking(False) + try: + client.connect(('127.0.0.1', serverSocket.getsockname()[1])) + except socket.error, e: + self.assertEqual(e.args[0], errno.EINPROGRESS) + else: + #raise AssertionError("Connect should have raised EINPROGRESS") + pass # FreeBSD doesn't raise an exception here + server, addr = serverSocket.accept() + + if sys.platform.startswith("darwin"): + flags = select.KQ_EV_ADD | select.KQ_EV_ENABLE + else: + flags = 0 + + kq = select.kqueue() + kq2 = select.kqueue.fromfd(kq.fileno()) + + ev = select.kevent(server.fileno(), + select.KQ_FILTER_WRITE, + select.KQ_EV_ADD | select.KQ_EV_ENABLE) + kq.control([ev], 0) + ev = select.kevent(server.fileno(), + select.KQ_FILTER_READ, + select.KQ_EV_ADD | select.KQ_EV_ENABLE) + kq.control([ev], 0) + ev = select.kevent(client.fileno(), + select.KQ_FILTER_WRITE, + select.KQ_EV_ADD | select.KQ_EV_ENABLE) + kq2.control([ev], 0) + ev = select.kevent(client.fileno(), + select.KQ_FILTER_READ, + select.KQ_EV_ADD | select.KQ_EV_ENABLE) + kq2.control([ev], 0) + + events = kq.control(None, 4, 1) + events = [(e.ident, e.filter, e.flags) for e in events] + events.sort() + self.assertEqual(events, [ + (client.fileno(), select.KQ_FILTER_WRITE, flags), + (server.fileno(), select.KQ_FILTER_WRITE, flags)]) + + client.send("Hello!") + server.send("world!!!") + + # We may need to call it several times + for i in range(10): + events = kq.control(None, 4, 1) + if len(events) == 4: + break + time.sleep(1.0) + else: + self.fail('timeout waiting for event notifications') + + events = [(e.ident, e.filter, e.flags) for e in events] + events.sort() + + self.assertEqual(events, [ + (client.fileno(), select.KQ_FILTER_WRITE, flags), + (client.fileno(), select.KQ_FILTER_READ, flags), + (server.fileno(), select.KQ_FILTER_WRITE, flags), + (server.fileno(), select.KQ_FILTER_READ, flags)]) + + # Remove completely client, and server read part + ev = select.kevent(client.fileno(), + select.KQ_FILTER_WRITE, + select.KQ_EV_DELETE) + kq.control([ev], 0) + ev = select.kevent(client.fileno(), + select.KQ_FILTER_READ, + select.KQ_EV_DELETE) + kq.control([ev], 0) + ev = select.kevent(server.fileno(), + select.KQ_FILTER_READ, + select.KQ_EV_DELETE) + kq.control([ev], 0, 0) + + events = kq.control([], 4, 0.99) + events = [(e.ident, e.filter, e.flags) for e in events] + events.sort() + self.assertEqual(events, [ + (server.fileno(), select.KQ_FILTER_WRITE, flags)]) + + client.close() + server.close() + serverSocket.close() + + def testPair(self): + kq = select.kqueue() + a, b = socket.socketpair() + + a.send(b'foo') + event1 = select.kevent(a, select.KQ_FILTER_READ, select.KQ_EV_ADD | select.KQ_EV_ENABLE) + event2 = select.kevent(b, select.KQ_FILTER_READ, select.KQ_EV_ADD | select.KQ_EV_ENABLE) + r = kq.control([event1, event2], 1, 1) + self.assertTrue(r) + self.assertFalse(r[0].flags & select.KQ_EV_ERROR) + self.assertEqual(b.recv(r[0].data), b'foo') + + a.close() + b.close() + kq.close() + +def test_main(): + test_support.run_unittest(TestKQueue) + +if __name__ == "__main__": + test_main() -- cgit v1.2.3