ยปCore Development>Code coverage>Lib/test/test_epoll.py

Python code coverage for Lib/test/test_epoll.py

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