+++ /dev/null
-# Copyright (c) 2001-2006 Twisted Matrix Laboratories.\r
-#\r
-# Permission is hereby granted, free of charge, to any person obtaining\r
-# a copy of this software and associated documentation files (the\r
-# "Software"), to deal in the Software without restriction, including\r
-# without limitation the rights to use, copy, modify, merge, publish,\r
-# distribute, sublicense, and/or sell copies of the Software, and to\r
-# permit persons to whom the Software is furnished to do so, subject to\r
-# the following conditions:\r
-#\r
-# The above copyright notice and this permission notice shall be\r
-# included in all copies or substantial portions of the Software.\r
-#\r
-# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,\r
-# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF\r
-# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND\r
-# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE\r
-# LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION\r
-# OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION\r
-# WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.\r
-"""\r
-Tests for epoll wrapper.\r
-"""\r
-import socket\r
-import errno\r
-import time\r
-import select\r
-import unittest\r
-\r
-from test import test_support\r
-if not hasattr(select, "epoll"):\r
- raise unittest.SkipTest("test works only on Linux 2.6")\r
-\r
-try:\r
- select.epoll()\r
-except IOError, e:\r
- if e.errno == errno.ENOSYS:\r
- raise unittest.SkipTest("kernel doesn't support epoll()")\r
-\r
-class TestEPoll(unittest.TestCase):\r
-\r
- def setUp(self):\r
- self.serverSocket = socket.socket()\r
- self.serverSocket.bind(('127.0.0.1', 0))\r
- self.serverSocket.listen(1)\r
- self.connections = [self.serverSocket]\r
-\r
-\r
- def tearDown(self):\r
- for skt in self.connections:\r
- skt.close()\r
-\r
- def _connected_pair(self):\r
- client = socket.socket()\r
- client.setblocking(False)\r
- try:\r
- client.connect(('127.0.0.1', self.serverSocket.getsockname()[1]))\r
- except socket.error, e:\r
- self.assertEqual(e.args[0], errno.EINPROGRESS)\r
- else:\r
- raise AssertionError("Connect should have raised EINPROGRESS")\r
- server, addr = self.serverSocket.accept()\r
-\r
- self.connections.extend((client, server))\r
- return client, server\r
-\r
- def test_create(self):\r
- try:\r
- ep = select.epoll(16)\r
- except OSError, e:\r
- raise AssertionError(str(e))\r
- self.assertTrue(ep.fileno() > 0, ep.fileno())\r
- self.assertTrue(not ep.closed)\r
- ep.close()\r
- self.assertTrue(ep.closed)\r
- self.assertRaises(ValueError, ep.fileno)\r
-\r
- def test_badcreate(self):\r
- self.assertRaises(TypeError, select.epoll, 1, 2, 3)\r
- self.assertRaises(TypeError, select.epoll, 'foo')\r
- self.assertRaises(TypeError, select.epoll, None)\r
- self.assertRaises(TypeError, select.epoll, ())\r
- self.assertRaises(TypeError, select.epoll, ['foo'])\r
- self.assertRaises(TypeError, select.epoll, {})\r
-\r
- def test_add(self):\r
- server, client = self._connected_pair()\r
-\r
- ep = select.epoll(2)\r
- try:\r
- ep.register(server.fileno(), select.EPOLLIN | select.EPOLLOUT)\r
- ep.register(client.fileno(), select.EPOLLIN | select.EPOLLOUT)\r
- finally:\r
- ep.close()\r
-\r
- # adding by object w/ fileno works, too.\r
- ep = select.epoll(2)\r
- try:\r
- ep.register(server, select.EPOLLIN | select.EPOLLOUT)\r
- ep.register(client, select.EPOLLIN | select.EPOLLOUT)\r
- finally:\r
- ep.close()\r
-\r
- ep = select.epoll(2)\r
- try:\r
- # TypeError: argument must be an int, or have a fileno() method.\r
- self.assertRaises(TypeError, ep.register, object(),\r
- select.EPOLLIN | select.EPOLLOUT)\r
- self.assertRaises(TypeError, ep.register, None,\r
- select.EPOLLIN | select.EPOLLOUT)\r
- # ValueError: file descriptor cannot be a negative integer (-1)\r
- self.assertRaises(ValueError, ep.register, -1,\r
- select.EPOLLIN | select.EPOLLOUT)\r
- # IOError: [Errno 9] Bad file descriptor\r
- self.assertRaises(IOError, ep.register, 10000,\r
- select.EPOLLIN | select.EPOLLOUT)\r
- # registering twice also raises an exception\r
- ep.register(server, select.EPOLLIN | select.EPOLLOUT)\r
- self.assertRaises(IOError, ep.register, server,\r
- select.EPOLLIN | select.EPOLLOUT)\r
- finally:\r
- ep.close()\r
-\r
- def test_fromfd(self):\r
- server, client = self._connected_pair()\r
-\r
- ep = select.epoll(2)\r
- ep2 = select.epoll.fromfd(ep.fileno())\r
-\r
- ep2.register(server.fileno(), select.EPOLLIN | select.EPOLLOUT)\r
- ep2.register(client.fileno(), select.EPOLLIN | select.EPOLLOUT)\r
-\r
- events = ep.poll(1, 4)\r
- events2 = ep2.poll(0.9, 4)\r
- self.assertEqual(len(events), 2)\r
- self.assertEqual(len(events2), 2)\r
-\r
- ep.close()\r
- try:\r
- ep2.poll(1, 4)\r
- except IOError, e:\r
- self.assertEqual(e.args[0], errno.EBADF, e)\r
- else:\r
- self.fail("epoll on closed fd didn't raise EBADF")\r
-\r
- def test_control_and_wait(self):\r
- client, server = self._connected_pair()\r
-\r
- ep = select.epoll(16)\r
- ep.register(server.fileno(),\r
- select.EPOLLIN | select.EPOLLOUT | select.EPOLLET)\r
- ep.register(client.fileno(),\r
- select.EPOLLIN | select.EPOLLOUT | select.EPOLLET)\r
-\r
- now = time.time()\r
- events = ep.poll(1, 4)\r
- then = time.time()\r
- self.assertFalse(then - now > 0.1, then - now)\r
-\r
- events.sort()\r
- expected = [(client.fileno(), select.EPOLLOUT),\r
- (server.fileno(), select.EPOLLOUT)]\r
- expected.sort()\r
-\r
- self.assertEqual(events, expected)\r
- self.assertFalse(then - now > 0.01, then - now)\r
-\r
- now = time.time()\r
- events = ep.poll(timeout=2.1, maxevents=4)\r
- then = time.time()\r
- self.assertFalse(events)\r
-\r
- client.send("Hello!")\r
- server.send("world!!!")\r
-\r
- now = time.time()\r
- events = ep.poll(1, 4)\r
- then = time.time()\r
- self.assertFalse(then - now > 0.01)\r
-\r
- events.sort()\r
- expected = [(client.fileno(), select.EPOLLIN | select.EPOLLOUT),\r
- (server.fileno(), select.EPOLLIN | select.EPOLLOUT)]\r
- expected.sort()\r
-\r
- self.assertEqual(events, expected)\r
-\r
- ep.unregister(client.fileno())\r
- ep.modify(server.fileno(), select.EPOLLOUT)\r
- now = time.time()\r
- events = ep.poll(1, 4)\r
- then = time.time()\r
- self.assertFalse(then - now > 0.01)\r
-\r
- expected = [(server.fileno(), select.EPOLLOUT)]\r
- self.assertEqual(events, expected)\r
-\r
- def test_errors(self):\r
- self.assertRaises(ValueError, select.epoll, -2)\r
- self.assertRaises(ValueError, select.epoll().register, -1,\r
- select.EPOLLIN)\r
-\r
- def test_unregister_closed(self):\r
- server, client = self._connected_pair()\r
- fd = server.fileno()\r
- ep = select.epoll(16)\r
- ep.register(server)\r
-\r
- now = time.time()\r
- events = ep.poll(1, 4)\r
- then = time.time()\r
- self.assertFalse(then - now > 0.01)\r
-\r
- server.close()\r
- ep.unregister(fd)\r
-\r
-def test_main():\r
- test_support.run_unittest(TestEPoll)\r
-\r
-if __name__ == "__main__":\r
- test_main()\r
projects / mirror_edk2.git / blobdiff