Lib/test/test_epoll.py

   1 # Copyright (c) 2001-2006 Twisted Matrix Laboratories.
   2 #
   3 # Permission is hereby granted, free of charge, to any person obtaining
   4 # a copy of this software and associated documentation files (the
   5 # "Software"), to deal in the Software without restriction, including
   6 # without limitation the rights to use, copy, modify, merge, publish,
   7 # distribute, sublicense, and/or sell copies of the Software, and to
   8 # permit persons to whom the Software is furnished to do so, subject to
   9 # the following conditions:
  10 #
  11 # The above copyright notice and this permission notice shall be
  12 # included in all copies or substantial portions of the Software.
  13 #
  14 # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
  15 # EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
  16 # MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
  17 # NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
  18 # LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
  19 # OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
  20 # WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
  21 """
  22 Tests for epoll wrapper.
  23 """
  24 import os
  25 import socket
  26 import errno
  27 import time
  28 import select
  29 import tempfile
  30 import unittest
  31
  32 from test import test_support
  33 if not hasattr(select, "epoll"):
  34     raise unittest.SkipTest("test works only on Linux 2.6")
  35
  36 try:
  37     select.epoll()
  38 except IOError, e:
  39     if e.errno == errno.ENOSYS:
  40         raise unittest.SkipTest("kernel doesn't support epoll()")
  41
  42 class TestEPoll(unittest.TestCase):
  43
  44     def setUp(self):
  45         self.serverSocket = socket.socket()
  46         self.serverSocket.bind(('127.0.0.1', 0))
  47         self.serverSocket.listen(1)
  48         self.connections = [self.serverSocket]
  49
  50
  51     def tearDown(self):
  52         for skt in self.connections:
  53             skt.close()
  54
  55     def _connected_pair(self):
  56         client = socket.socket()
  57         client.setblocking(False)
  58         try:
  59             client.connect(('127.0.0.1', self.serverSocket.getsockname()[1]))
  60         except socket.error, e:
  61             self.assertEquals(e.args[0], errno.EINPROGRESS)
  62         else:
  63             raise AssertionError("Connect should have raised EINPROGRESS")
  64         server, addr = self.serverSocket.accept()
  65
  66         self.connections.extend((client, server))
  67         return client, server
  68
  69     def test_create(self):
  70         try:
  71             ep = select.epoll(16)
  72         except OSError, e:
  73             raise AssertionError(str(e))
  74         self.assertTrue(ep.fileno() > 0, ep.fileno())
  75         self.assertTrue(not ep.closed)
  76         ep.close()
  77         self.assertTrue(ep.closed)
  78         self.assertRaises(ValueError, ep.fileno)
  79
  80     def test_badcreate(self):
  81         self.assertRaises(TypeError, select.epoll, 1, 2, 3)
  82         self.assertRaises(TypeError, select.epoll, 'foo')
  83         self.assertRaises(TypeError, select.epoll, None)
  84         self.assertRaises(TypeError, select.epoll, ())
  85         self.assertRaises(TypeError, select.epoll, ['foo'])
  86         self.assertRaises(TypeError, select.epoll, {})
  87
  88     def test_add(self):
  89         server, client = self._connected_pair()
  90
  91         ep = select.epoll(2)
  92         try:
  93             ep.register(server.fileno(), select.EPOLLIN | select.EPOLLOUT)
  94             ep.register(client.fileno(), select.EPOLLIN | select.EPOLLOUT)
  95         finally:
  96             ep.close()
  97
  98         # adding by object w/ fileno works, too.
  99         ep = select.epoll(2)
 100         try:
 101             ep.register(server, select.EPOLLIN | select.EPOLLOUT)
 102             ep.register(client, select.EPOLLIN | select.EPOLLOUT)
 103         finally:
 104             ep.close()
 105
 106         ep = select.epoll(2)
 107         try:
 108             # TypeError: argument must be an int, or have a fileno() method.
 109             self.assertRaises(TypeError, ep.register, object(),
 110                 select.EPOLLIN | select.EPOLLOUT)
 111             self.assertRaises(TypeError, ep.register, None,
 112                 select.EPOLLIN | select.EPOLLOUT)
 113             # ValueError: file descriptor cannot be a negative integer (-1)
 114             self.assertRaises(ValueError, ep.register, -1,
 115                 select.EPOLLIN | select.EPOLLOUT)
 116             # IOError: [Errno 9] Bad file descriptor
 117             self.assertRaises(IOError, ep.register, 10000,
 118                 select.EPOLLIN | select.EPOLLOUT)
 119             # registering twice also raises an exception
 120             ep.register(server, select.EPOLLIN | select.EPOLLOUT)
 121             self.assertRaises(IOError, ep.register, server,
 122                 select.EPOLLIN | select.EPOLLOUT)
 123         finally:
 124             ep.close()
 125
 126     def test_fromfd(self):
 127         server, client = self._connected_pair()
 128
 129         ep = select.epoll(2)
 130         ep2 = select.epoll.fromfd(ep.fileno())
 131
 132         ep2.register(server.fileno(), select.EPOLLIN | select.EPOLLOUT)
 133         ep2.register(client.fileno(), select.EPOLLIN | select.EPOLLOUT)
 134
 135         events = ep.poll(1, 4)
 136         events2 = ep2.poll(0.9, 4)
 137         self.assertEqual(len(events), 2)
 138         self.assertEqual(len(events2), 2)
 139
 140         ep.close()
 141         try:
 142             ep2.poll(1, 4)
 143         except IOError, e:
 144             self.assertEqual(e.args[0], errno.EBADF, e)
 145         else:
 146             self.fail("epoll on closed fd didn't raise EBADF")
 147
 148     def test_control_and_wait(self):
 149         client, server = self._connected_pair()
 150
 151         ep = select.epoll(16)
 152         ep.register(server.fileno(),
 153                    select.EPOLLIN | select.EPOLLOUT | select.EPOLLET)
 154         ep.register(client.fileno(),
 155                    select.EPOLLIN | select.EPOLLOUT | select.EPOLLET)
 156
 157         now = time.time()
 158         events = ep.poll(1, 4)
 159         then = time.time()
 160         self.assertFalse(then - now > 0.1, then - now)
 161
 162         events.sort()
 163         expected = [(client.fileno(), select.EPOLLOUT),
 164                     (server.fileno(), select.EPOLLOUT)]
 165         expected.sort()
 166
 167         self.assertEquals(events, expected)
 168         self.assertFalse(then - now > 0.01, then - now)
 169
 170         now = time.time()
 171         events = ep.poll(timeout=2.1, maxevents=4)
 172         then = time.time()
 173         self.assertFalse(events)
 174
 175         client.send("Hello!")
 176         server.send("world!!!")
 177
 178         now = time.time()
 179         events = ep.poll(1, 4)
 180         then = time.time()
 181         self.assertFalse(then - now > 0.01)
 182
 183         events.sort()
 184         expected = [(client.fileno(), select.EPOLLIN | select.EPOLLOUT),
 185                     (server.fileno(), select.EPOLLIN | select.EPOLLOUT)]
 186         expected.sort()
 187
 188         self.assertEquals(events, expected)
 189
 190         ep.unregister(client.fileno())
 191         ep.modify(server.fileno(), select.EPOLLOUT)
 192         now = time.time()
 193         events = ep.poll(1, 4)
 194         then = time.time()
 195         self.assertFalse(then - now > 0.01)
 196
 197         expected = [(server.fileno(), select.EPOLLOUT)]
 198         self.assertEquals(events, expected)
 199
 200     def test_errors(self):
 201         self.assertRaises(ValueError, select.epoll, -2)
 202         self.assertRaises(ValueError, select.epoll().register, -1,
 203                           select.EPOLLIN)
 204
 205     def test_unregister_closed(self):
 206         server, client = self._connected_pair()
 207         fd = server.fileno()
 208         ep = select.epoll(16)
 209         ep.register(server)
 210
 211         now = time.time()
 212         events = ep.poll(1, 4)
 213         then = time.time()
 214         self.assertFalse(then - now > 0.01)
 215
 216         server.close()
 217         ep.unregister(fd)
 218
 219 def test_main():
 220     test_support.run_unittest(TestEPoll)
 221
 222 if __name__ == "__main__":
 223     test_main()