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
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
|
"""
AMAK: 20050515: This module is the test_select.py from cpython 2.4, ported to jython + unittest
"""
import errno
import os
import select
import socket
import sys
import test_socket
import unittest
from test import test_support
HOST = test_socket.HOST
PORT = test_socket.PORT + 100
class SelectWrapper:
def __init__(self):
self.read_fds = []
self.write_fds = []
self.oob_fds = []
self.timeout = None
def add_read_fd(self, fd):
self.read_fds.append(fd)
def add_write_fd(self, fd):
self.write_fds.append(fd)
def add_oob_fd(self, fd):
self.oob_fds.append(fd)
def set_timeout(self, timeout):
self.timeout = timeout
class PollWrapper:
def __init__(self):
self.timeout = None
self.poll_object = select.poll()
def add_read_fd(self, fd):
self.poll_object.register(fd, select.POLL_IN)
def add_write_fd(self, fd):
self.poll_object.register(fd, select.POLL_OUT)
def add_oob_fd(self, fd):
self.poll_object.register(fd, select.POLL_PRI)
class TestSelectInvalidParameters(unittest.TestCase):
def testBadSelectSetTypes(self):
# Test some known error conditions
for bad_select_set in [None, 1,]:
for pos in range(2): # OOB not supported on Java
args = [[], [], []]
args[pos] = bad_select_set
try:
timeout = 0 # Can't wait forever
rfd, wfd, xfd = select.select(args[0], args[1], args[2], timeout)
except (select.error, TypeError):
pass
except Exception, x:
self.fail("Selecting on '%s' raised wrong exception %s" % (str(bad_select_set), str(x)))
else:
self.fail("Selecting on '%s' should have raised TypeError" % str(bad_select_set))
def testBadSelectableTypes(self):
class Nope: pass
class Almost1:
def fileno(self):
return 'fileno'
class Almost2:
def fileno(self):
return 'fileno'
# Test some known error conditions
for bad_selectable in [None, 1, object(), Nope(), Almost1(), Almost2()]:
try:
timeout = 0 # Can't wait forever
rfd, wfd, xfd = select.select([bad_selectable], [], [], timeout)
except (TypeError, select.error), x:
pass
else:
self.fail("Selecting on '%s' should have raised TypeError or select.error" % str(bad_selectable))
def testInvalidTimeoutTypes(self):
for invalid_timeout in ['not a number']:
try:
rfd, wfd, xfd = select.select([], [], [], invalid_timeout)
except TypeError:
pass
else:
self.fail("Invalid timeout value '%s' should have raised TypeError" % invalid_timeout)
def testInvalidTimeoutValues(self):
for invalid_timeout in [-1]:
try:
rfd, wfd, xfd = select.select([], [], [], invalid_timeout)
except (ValueError, select.error):
pass
else:
self.fail("Invalid timeout value '%s' should have raised ValueError or select.error" % invalid_timeout)
class TestSelectClientSocket(unittest.TestCase):
def testUnconnectedSocket(self):
sockets = [socket.socket(socket.AF_INET, socket.SOCK_STREAM) for x in range(5)]
for pos in range(2): # OOB not supported on Java
args = [[], [], []]
args[pos] = sockets
timeout = 0 # Can't wait forever
rfd, wfd, xfd = select.select(args[0], args[1], args[2], timeout)
for s in sockets:
self.failIf(s in rfd)
self.failIf(s in wfd)
class TestPollClientSocket(unittest.TestCase):
def testEventConstants(self):
for event_name in ['IN', 'OUT', 'PRI', 'ERR', 'HUP', 'NVAL', ]:
self.failUnless(hasattr(select, 'POLL%s' % event_name))
def testUnregisterRaisesKeyError(self):
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
poll_object = select.poll()
try:
poll_object.unregister(s)
except KeyError:
pass
else:
self.fail("Unregistering socket that is not registered should have raised KeyError")
#
# using the test_socket thread based server/client management, for convenience.
#
class ThreadedPollClientSocket(test_socket.ThreadedTCPSocketTest):
HOST = HOST
PORT = PORT
def testSocketRegisteredBeforeConnected(self):
self.cli_conn = self.serv.accept()
def _testSocketRegisteredBeforeConnected(self):
timeout = 1000 # milliseconds
poll_object = select.poll()
# Register the socket before it is connected
poll_object.register(self.cli, select.POLLOUT)
result_list = poll_object.poll(timeout)
result_sockets = [r[0] for r in result_list]
self.failIf(self.cli in result_sockets, "Unconnected client socket should not have been selectable")
# Now connect the socket, but DO NOT register it again
self.cli.setblocking(0)
self.cli.connect( (self.HOST, self.PORT) )
# Now poll again, to check that the poll object has recognised that the socket is now connected
result_list = poll_object.poll(timeout)
result_sockets = [r[0] for r in result_list]
self.failUnless(self.cli in result_sockets, "Connected client socket should have been selectable")
def testSocketMustBeNonBlocking(self):
self.cli_conn = self.serv.accept()
def _testSocketMustBeNonBlocking(self):
self.cli.setblocking(1)
self.cli.connect( (self.HOST, self.PORT) )
timeout = 1000 # milliseconds
poll_object = select.poll()
try:
poll_object.register(self.cli)
except select.error, se:
self.failUnlessEqual(se[0], errno.ESOCKISBLOCKING)
except Exception, x:
self.fail("Registering blocking socket should have raised select.error, not %s" % str(x))
else:
self.fail("Registering blocking socket should have raised select.error")
def testSelectOnSocketFileno(self):
self.cli_conn = self.serv.accept()
def _testSelectOnSocketFileno(self):
self.cli.setblocking(0)
self.cli.connect( (self.HOST, self.PORT) )
return
try:
rfd, wfd, xfd = select.select([self.cli.fileno()], [], [], 1)
except Exception, x:
self.fail("Selecting on socket.fileno() should not have raised exception: %s" % str(x))
class TestPipes(unittest.TestCase):
verbose = 1
def test(self):
import sys
from test.test_support import verbose
if sys.platform[:3] in ('win', 'mac', 'os2', 'riscos'):
if verbose:
print "Can't test select easily on", sys.platform
return
cmd = 'for i in 0 1 2 3 4 5 6 7 8 9; do echo testing...; sleep 1; done'
p = os.popen(cmd, 'r')
for tout in (0, 1, 2, 4, 8, 16) + (None,)*10:
if verbose:
print 'timeout =', tout
rfd, wfd, xfd = select.select([p], [], [], tout)
if (rfd, wfd, xfd) == ([], [], []):
continue
if (rfd, wfd, xfd) == ([p], [], []):
line = p.readline()
if verbose:
print repr(line)
if not line:
if verbose:
print 'EOF'
break
continue
self.fail('Unexpected return values from select(): %s' % str(rfd, wfd, xfd))
p.close()
def test_main():
tests = [
TestSelectInvalidParameters,
TestSelectClientSocket,
TestPollClientSocket,
ThreadedPollClientSocket,
]
if sys.platform[:4] != 'java':
tests.append(TestPipes)
suites = [unittest.makeSuite(klass, 'test') for klass in tests]
test_support.run_suite(unittest.TestSuite(suites))
if __name__ == "__main__":
test_main()
|
