1 # Copyright (c) Twisted Matrix Laboratories.
2 # See LICENSE for details.
5 Tests for epoll wrapper.
8 import socket, errno, time
10 from twisted.trial import unittest
11 from twisted.python.util import untilConcludes
14 from twisted.python import _epoll
19 class EPoll(unittest.TestCase):
21 Tests for the low-level epoll bindings.
25 Create a listening server port and a list with which to keep track
28 self.serverSocket = socket.socket()
29 self.serverSocket.bind(('127.0.0.1', 0))
30 self.serverSocket.listen(1)
31 self.connections = [self.serverSocket]
36 Close any sockets which were opened by the test.
38 for skt in self.connections:
42 def _connectedPair(self):
44 Return the two sockets which make up a new TCP connection.
46 client = socket.socket()
47 client.setblocking(False)
49 client.connect(('127.0.0.1', self.serverSocket.getsockname()[1]))
50 except socket.error, e:
51 self.assertEqual(e.args[0], errno.EINPROGRESS)
53 raise unittest.FailTest("Connect should have raised EINPROGRESS")
54 server, addr = self.serverSocket.accept()
56 self.connections.extend((client, server))
60 def test_create(self):
62 Test the creation of an epoll object.
67 raise unittest.FailTest(str(e))
72 def test_badCreate(self):
74 Test that attempting to create an epoll object with some random
75 objects raises a TypeError.
77 self.assertRaises(TypeError, _epoll.epoll, 1, 2, 3)
78 self.assertRaises(TypeError, _epoll.epoll, 'foo')
79 self.assertRaises(TypeError, _epoll.epoll, None)
80 self.assertRaises(TypeError, _epoll.epoll, ())
81 self.assertRaises(TypeError, _epoll.epoll, ['foo'])
82 self.assertRaises(TypeError, _epoll.epoll, {})
87 Test adding a socket to an epoll object.
89 server, client = self._connectedPair()
93 p._control(_epoll.CTL_ADD, server.fileno(), _epoll.IN | _epoll.OUT)
94 p._control(_epoll.CTL_ADD, client.fileno(), _epoll.IN | _epoll.OUT)
99 def test_controlAndWait(self):
101 Test waiting on an epoll object which has had some sockets added to
104 client, server = self._connectedPair()
107 p._control(_epoll.CTL_ADD, client.fileno(), _epoll.IN | _epoll.OUT |
109 p._control(_epoll.CTL_ADD, server.fileno(), _epoll.IN | _epoll.OUT |
113 events = untilConcludes(p.wait, 4, 1000)
115 self.failIf(then - now > 0.01)
118 expected = [(client.fileno(), _epoll.OUT),
119 (server.fileno(), _epoll.OUT)]
122 self.assertEqual(events, expected)
125 events = untilConcludes(p.wait, 4, 200)
127 self.failUnless(then - now > 0.1)
130 client.send("Hello!")
131 server.send("world!!!")
134 events = untilConcludes(p.wait, 4, 1000)
136 self.failIf(then - now > 0.01)
139 expected = [(client.fileno(), _epoll.IN | _epoll.OUT),
140 (server.fileno(), _epoll.IN | _epoll.OUT)]
143 self.assertEqual(events, expected)
146 EPoll.skip = "_epoll module unavailable"
151 if exc.errno == errno.ENOSYS:
153 EPoll.skip = "epoll support missing from platform"