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

Python code coverage for Lib/test/test_kqueue.py

#countcontent
1n/a"""
2n/aTests for kqueue wrapper.
3n/a"""
4n/aimport errno
5n/aimport os
6n/aimport select
7n/aimport socket
8n/aimport time
9n/aimport unittest
10n/a
11n/aif not hasattr(select, "kqueue"):
12n/a raise unittest.SkipTest("test works only on BSD")
13n/a
14n/aclass TestKQueue(unittest.TestCase):
15n/a def test_create_queue(self):
16n/a kq = select.kqueue()
17n/a self.assertTrue(kq.fileno() > 0, kq.fileno())
18n/a self.assertTrue(not kq.closed)
19n/a kq.close()
20n/a self.assertTrue(kq.closed)
21n/a self.assertRaises(ValueError, kq.fileno)
22n/a
23n/a def test_create_event(self):
24n/a from operator import lt, le, gt, ge
25n/a
26n/a fd = os.open(os.devnull, os.O_WRONLY)
27n/a self.addCleanup(os.close, fd)
28n/a
29n/a ev = select.kevent(fd)
30n/a other = select.kevent(1000)
31n/a self.assertEqual(ev.ident, fd)
32n/a self.assertEqual(ev.filter, select.KQ_FILTER_READ)
33n/a self.assertEqual(ev.flags, select.KQ_EV_ADD)
34n/a self.assertEqual(ev.fflags, 0)
35n/a self.assertEqual(ev.data, 0)
36n/a self.assertEqual(ev.udata, 0)
37n/a self.assertEqual(ev, ev)
38n/a self.assertNotEqual(ev, other)
39n/a self.assertTrue(ev < other)
40n/a self.assertTrue(other >= ev)
41n/a for op in lt, le, gt, ge:
42n/a self.assertRaises(TypeError, op, ev, None)
43n/a self.assertRaises(TypeError, op, ev, 1)
44n/a self.assertRaises(TypeError, op, ev, "ev")
45n/a
46n/a ev = select.kevent(fd, select.KQ_FILTER_WRITE)
47n/a self.assertEqual(ev.ident, fd)
48n/a self.assertEqual(ev.filter, select.KQ_FILTER_WRITE)
49n/a self.assertEqual(ev.flags, select.KQ_EV_ADD)
50n/a self.assertEqual(ev.fflags, 0)
51n/a self.assertEqual(ev.data, 0)
52n/a self.assertEqual(ev.udata, 0)
53n/a self.assertEqual(ev, ev)
54n/a self.assertNotEqual(ev, other)
55n/a
56n/a ev = select.kevent(fd, select.KQ_FILTER_WRITE, select.KQ_EV_ONESHOT)
57n/a self.assertEqual(ev.ident, fd)
58n/a self.assertEqual(ev.filter, select.KQ_FILTER_WRITE)
59n/a self.assertEqual(ev.flags, select.KQ_EV_ONESHOT)
60n/a self.assertEqual(ev.fflags, 0)
61n/a self.assertEqual(ev.data, 0)
62n/a self.assertEqual(ev.udata, 0)
63n/a self.assertEqual(ev, ev)
64n/a self.assertNotEqual(ev, other)
65n/a
66n/a ev = select.kevent(1, 2, 3, 4, 5, 6)
67n/a self.assertEqual(ev.ident, 1)
68n/a self.assertEqual(ev.filter, 2)
69n/a self.assertEqual(ev.flags, 3)
70n/a self.assertEqual(ev.fflags, 4)
71n/a self.assertEqual(ev.data, 5)
72n/a self.assertEqual(ev.udata, 6)
73n/a self.assertEqual(ev, ev)
74n/a self.assertNotEqual(ev, other)
75n/a
76n/a bignum = 0x7fff
77n/a ev = select.kevent(bignum, 1, 2, 3, bignum - 1, bignum)
78n/a self.assertEqual(ev.ident, bignum)
79n/a self.assertEqual(ev.filter, 1)
80n/a self.assertEqual(ev.flags, 2)
81n/a self.assertEqual(ev.fflags, 3)
82n/a self.assertEqual(ev.data, bignum - 1)
83n/a self.assertEqual(ev.udata, bignum)
84n/a self.assertEqual(ev, ev)
85n/a self.assertNotEqual(ev, other)
86n/a
87n/a # Issue 11973
88n/a bignum = 0xffff
89n/a ev = select.kevent(0, 1, bignum)
90n/a self.assertEqual(ev.ident, 0)
91n/a self.assertEqual(ev.filter, 1)
92n/a self.assertEqual(ev.flags, bignum)
93n/a self.assertEqual(ev.fflags, 0)
94n/a self.assertEqual(ev.data, 0)
95n/a self.assertEqual(ev.udata, 0)
96n/a self.assertEqual(ev, ev)
97n/a self.assertNotEqual(ev, other)
98n/a
99n/a # Issue 11973
100n/a bignum = 0xffffffff
101n/a ev = select.kevent(0, 1, 2, bignum)
102n/a self.assertEqual(ev.ident, 0)
103n/a self.assertEqual(ev.filter, 1)
104n/a self.assertEqual(ev.flags, 2)
105n/a self.assertEqual(ev.fflags, bignum)
106n/a self.assertEqual(ev.data, 0)
107n/a self.assertEqual(ev.udata, 0)
108n/a self.assertEqual(ev, ev)
109n/a self.assertNotEqual(ev, other)
110n/a
111n/a
112n/a def test_queue_event(self):
113n/a serverSocket = socket.socket()
114n/a serverSocket.bind(('127.0.0.1', 0))
115n/a serverSocket.listen()
116n/a client = socket.socket()
117n/a client.setblocking(False)
118n/a try:
119n/a client.connect(('127.0.0.1', serverSocket.getsockname()[1]))
120n/a except OSError as e:
121n/a self.assertEqual(e.args[0], errno.EINPROGRESS)
122n/a else:
123n/a #raise AssertionError("Connect should have raised EINPROGRESS")
124n/a pass # FreeBSD doesn't raise an exception here
125n/a server, addr = serverSocket.accept()
126n/a
127n/a kq = select.kqueue()
128n/a kq2 = select.kqueue.fromfd(kq.fileno())
129n/a
130n/a ev = select.kevent(server.fileno(),
131n/a select.KQ_FILTER_WRITE,
132n/a select.KQ_EV_ADD | select.KQ_EV_ENABLE)
133n/a kq.control([ev], 0)
134n/a ev = select.kevent(server.fileno(),
135n/a select.KQ_FILTER_READ,
136n/a select.KQ_EV_ADD | select.KQ_EV_ENABLE)
137n/a kq.control([ev], 0)
138n/a ev = select.kevent(client.fileno(),
139n/a select.KQ_FILTER_WRITE,
140n/a select.KQ_EV_ADD | select.KQ_EV_ENABLE)
141n/a kq2.control([ev], 0)
142n/a ev = select.kevent(client.fileno(),
143n/a select.KQ_FILTER_READ,
144n/a select.KQ_EV_ADD | select.KQ_EV_ENABLE)
145n/a kq2.control([ev], 0)
146n/a
147n/a events = kq.control(None, 4, 1)
148n/a events = set((e.ident, e.filter) for e in events)
149n/a self.assertEqual(events, set([
150n/a (client.fileno(), select.KQ_FILTER_WRITE),
151n/a (server.fileno(), select.KQ_FILTER_WRITE)]))
152n/a
153n/a client.send(b"Hello!")
154n/a server.send(b"world!!!")
155n/a
156n/a # We may need to call it several times
157n/a for i in range(10):
158n/a events = kq.control(None, 4, 1)
159n/a if len(events) == 4:
160n/a break
161n/a time.sleep(1.0)
162n/a else:
163n/a self.fail('timeout waiting for event notifications')
164n/a
165n/a events = set((e.ident, e.filter) for e in events)
166n/a self.assertEqual(events, set([
167n/a (client.fileno(), select.KQ_FILTER_WRITE),
168n/a (client.fileno(), select.KQ_FILTER_READ),
169n/a (server.fileno(), select.KQ_FILTER_WRITE),
170n/a (server.fileno(), select.KQ_FILTER_READ)]))
171n/a
172n/a # Remove completely client, and server read part
173n/a ev = select.kevent(client.fileno(),
174n/a select.KQ_FILTER_WRITE,
175n/a select.KQ_EV_DELETE)
176n/a kq.control([ev], 0)
177n/a ev = select.kevent(client.fileno(),
178n/a select.KQ_FILTER_READ,
179n/a select.KQ_EV_DELETE)
180n/a kq.control([ev], 0)
181n/a ev = select.kevent(server.fileno(),
182n/a select.KQ_FILTER_READ,
183n/a select.KQ_EV_DELETE)
184n/a kq.control([ev], 0, 0)
185n/a
186n/a events = kq.control([], 4, 0.99)
187n/a events = set((e.ident, e.filter) for e in events)
188n/a self.assertEqual(events, set([
189n/a (server.fileno(), select.KQ_FILTER_WRITE)]))
190n/a
191n/a client.close()
192n/a server.close()
193n/a serverSocket.close()
194n/a
195n/a def testPair(self):
196n/a kq = select.kqueue()
197n/a a, b = socket.socketpair()
198n/a
199n/a a.send(b'foo')
200n/a event1 = select.kevent(a, select.KQ_FILTER_READ, select.KQ_EV_ADD | select.KQ_EV_ENABLE)
201n/a event2 = select.kevent(b, select.KQ_FILTER_READ, select.KQ_EV_ADD | select.KQ_EV_ENABLE)
202n/a r = kq.control([event1, event2], 1, 1)
203n/a self.assertTrue(r)
204n/a self.assertFalse(r[0].flags & select.KQ_EV_ERROR)
205n/a self.assertEqual(b.recv(r[0].data), b'foo')
206n/a
207n/a a.close()
208n/a b.close()
209n/a kq.close()
210n/a
211n/a def test_close(self):
212n/a open_file = open(__file__, "rb")
213n/a self.addCleanup(open_file.close)
214n/a fd = open_file.fileno()
215n/a kqueue = select.kqueue()
216n/a
217n/a # test fileno() method and closed attribute
218n/a self.assertIsInstance(kqueue.fileno(), int)
219n/a self.assertFalse(kqueue.closed)
220n/a
221n/a # test close()
222n/a kqueue.close()
223n/a self.assertTrue(kqueue.closed)
224n/a self.assertRaises(ValueError, kqueue.fileno)
225n/a
226n/a # close() can be called more than once
227n/a kqueue.close()
228n/a
229n/a # operations must fail with ValueError("I/O operation on closed ...")
230n/a self.assertRaises(ValueError, kqueue.control, None, 4)
231n/a
232n/a def test_fd_non_inheritable(self):
233n/a kqueue = select.kqueue()
234n/a self.addCleanup(kqueue.close)
235n/a self.assertEqual(os.get_inheritable(kqueue.fileno()), False)
236n/a
237n/a
238n/aif __name__ == "__main__":
239n/a unittest.main()