1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
|
# Copyright (c) 2001-2006 Twisted Matrix Laboratories.
# See LICENSE for details.
"""
Tests for epoll wrapper.
"""
import socket, errno, time
from twisted.trial import unittest
from twisted.python.util import untilConcludes
try:
from twisted.python import _epoll
except ImportError:
_epoll = None
class EPoll(unittest.TestCase):
"""
Tests for the low-level epoll bindings.
"""
def setUp(self):
"""
Create a listening server port and a list with which to keep track
of created sockets.
"""
self.serverSocket = socket.socket()
self.serverSocket.bind(('127.0.0.1', 0))
self.serverSocket.listen(1)
self.connections = [self.serverSocket]
def tearDown(self):
"""
Close any sockets which were opened by the test.
"""
for skt in self.connections:
skt.close()
def _connectedPair(self):
"""
Return the two sockets which make up a new TCP connection.
"""
client = socket.socket()
client.setblocking(False)
try:
client.connect(('127.0.0.1', self.serverSocket.getsockname()[1]))
except socket.error, e:
self.assertEquals(e.args[0], errno.EINPROGRESS)
else:
raise unittest.FailTest("Connect should have raised EINPROGRESS")
server, addr = self.serverSocket.accept()
self.connections.extend((client, server))
return client, server
def test_create(self):
"""
Test the creation of an epoll object.
"""
try:
p = _epoll.epoll(16)
except OSError, e:
raise unittest.FailTest(str(e))
else:
p.close()
def test_badCreate(self):
"""
Test that attempting to create an epoll object with some random
objects raises a TypeError.
"""
self.assertRaises(TypeError, _epoll.epoll, 1, 2, 3)
self.assertRaises(TypeError, _epoll.epoll, 'foo')
self.assertRaises(TypeError, _epoll.epoll, None)
self.assertRaises(TypeError, _epoll.epoll, ())
self.assertRaises(TypeError, _epoll.epoll, ['foo'])
self.assertRaises(TypeError, _epoll.epoll, {})
self.assertRaises(TypeError, _epoll.epoll)
def test_add(self):
"""
Test adding a socket to an epoll object.
"""
server, client = self._connectedPair()
p = _epoll.epoll(2)
try:
p._control(_epoll.CTL_ADD, server.fileno(), _epoll.IN | _epoll.OUT)
p._control(_epoll.CTL_ADD, client.fileno(), _epoll.IN | _epoll.OUT)
finally:
p.close()
def test_controlAndWait(self):
"""
Test waiting on an epoll object which has had some sockets added to
it.
"""
client, server = self._connectedPair()
p = _epoll.epoll(16)
p._control(_epoll.CTL_ADD, client.fileno(), _epoll.IN | _epoll.OUT |
_epoll.ET)
p._control(_epoll.CTL_ADD, server.fileno(), _epoll.IN | _epoll.OUT |
_epoll.ET)
now = time.time()
events = untilConcludes(p.wait, 4, 1000)
then = time.time()
self.failIf(then - now > 0.01)
events.sort()
expected = [(client.fileno(), _epoll.OUT),
(server.fileno(), _epoll.OUT)]
expected.sort()
self.assertEquals(events, expected)
now = time.time()
events = untilConcludes(p.wait, 4, 200)
then = time.time()
self.failUnless(then - now > 0.1)
self.failIf(events)
client.send("Hello!")
server.send("world!!!")
now = time.time()
events = untilConcludes(p.wait, 4, 1000)
then = time.time()
self.failIf(then - now > 0.01)
events.sort()
expected = [(client.fileno(), _epoll.IN | _epoll.OUT),
(server.fileno(), _epoll.IN | _epoll.OUT)]
expected.sort()
self.assertEquals(events, expected)
if _epoll is None:
EPoll.skip = "_epoll module unavailable"
else:
try:
e = _epoll.epoll(16)
except IOError, exc:
if exc.errno == errno.ENOSYS:
del exc
EPoll.skip = "epoll support missing from platform"
else:
raise
else:
e.close()
del e
|
