Python code coverage for Lib/test/test_epoll.py

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