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

Python code coverage for Lib/test/test_devpoll.py

#countcontent
1n/a# Test case for the select.devpoll() function
2n/a
3n/a# Initial tests are copied as is from "test_poll.py"
4n/a
5n/aimport os
6n/aimport random
7n/aimport select
8n/aimport unittest
9n/afrom test.support import run_unittest, cpython_only
10n/a
11n/aif not hasattr(select, 'devpoll') :
12n/a raise unittest.SkipTest('test works only on Solaris OS family')
13n/a
14n/a
15n/adef find_ready_matching(ready, flag):
16n/a match = []
17n/a for fd, mode in ready:
18n/a if mode & flag:
19n/a match.append(fd)
20n/a return match
21n/a
22n/aclass DevPollTests(unittest.TestCase):
23n/a
24n/a def test_devpoll1(self):
25n/a # Basic functional test of poll object
26n/a # Create a bunch of pipe and test that poll works with them.
27n/a
28n/a p = select.devpoll()
29n/a
30n/a NUM_PIPES = 12
31n/a MSG = b" This is a test."
32n/a MSG_LEN = len(MSG)
33n/a readers = []
34n/a writers = []
35n/a r2w = {}
36n/a w2r = {}
37n/a
38n/a for i in range(NUM_PIPES):
39n/a rd, wr = os.pipe()
40n/a p.register(rd)
41n/a p.modify(rd, select.POLLIN)
42n/a p.register(wr, select.POLLOUT)
43n/a readers.append(rd)
44n/a writers.append(wr)
45n/a r2w[rd] = wr
46n/a w2r[wr] = rd
47n/a
48n/a bufs = []
49n/a
50n/a while writers:
51n/a ready = p.poll()
52n/a ready_writers = find_ready_matching(ready, select.POLLOUT)
53n/a if not ready_writers:
54n/a self.fail("no pipes ready for writing")
55n/a wr = random.choice(ready_writers)
56n/a os.write(wr, MSG)
57n/a
58n/a ready = p.poll()
59n/a ready_readers = find_ready_matching(ready, select.POLLIN)
60n/a if not ready_readers:
61n/a self.fail("no pipes ready for reading")
62n/a self.assertEqual([w2r[wr]], ready_readers)
63n/a rd = ready_readers[0]
64n/a buf = os.read(rd, MSG_LEN)
65n/a self.assertEqual(len(buf), MSG_LEN)
66n/a bufs.append(buf)
67n/a os.close(r2w[rd]) ; os.close(rd)
68n/a p.unregister(r2w[rd])
69n/a p.unregister(rd)
70n/a writers.remove(r2w[rd])
71n/a
72n/a self.assertEqual(bufs, [MSG] * NUM_PIPES)
73n/a
74n/a def test_timeout_overflow(self):
75n/a pollster = select.devpoll()
76n/a w, r = os.pipe()
77n/a pollster.register(w)
78n/a
79n/a pollster.poll(-1)
80n/a self.assertRaises(OverflowError, pollster.poll, -2)
81n/a self.assertRaises(OverflowError, pollster.poll, -1 << 31)
82n/a self.assertRaises(OverflowError, pollster.poll, -1 << 64)
83n/a
84n/a pollster.poll(0)
85n/a pollster.poll(1)
86n/a pollster.poll(1 << 30)
87n/a self.assertRaises(OverflowError, pollster.poll, 1 << 31)
88n/a self.assertRaises(OverflowError, pollster.poll, 1 << 63)
89n/a self.assertRaises(OverflowError, pollster.poll, 1 << 64)
90n/a
91n/a def test_close(self):
92n/a open_file = open(__file__, "rb")
93n/a self.addCleanup(open_file.close)
94n/a fd = open_file.fileno()
95n/a devpoll = select.devpoll()
96n/a
97n/a # test fileno() method and closed attribute
98n/a self.assertIsInstance(devpoll.fileno(), int)
99n/a self.assertFalse(devpoll.closed)
100n/a
101n/a # test close()
102n/a devpoll.close()
103n/a self.assertTrue(devpoll.closed)
104n/a self.assertRaises(ValueError, devpoll.fileno)
105n/a
106n/a # close() can be called more than once
107n/a devpoll.close()
108n/a
109n/a # operations must fail with ValueError("I/O operation on closed ...")
110n/a self.assertRaises(ValueError, devpoll.modify, fd, select.POLLIN)
111n/a self.assertRaises(ValueError, devpoll.poll)
112n/a self.assertRaises(ValueError, devpoll.register, fd, fd, select.POLLIN)
113n/a self.assertRaises(ValueError, devpoll.unregister, fd)
114n/a
115n/a def test_fd_non_inheritable(self):
116n/a devpoll = select.devpoll()
117n/a self.addCleanup(devpoll.close)
118n/a self.assertEqual(os.get_inheritable(devpoll.fileno()), False)
119n/a
120n/a def test_events_mask_overflow(self):
121n/a pollster = select.devpoll()
122n/a w, r = os.pipe()
123n/a pollster.register(w)
124n/a # Issue #17919
125n/a self.assertRaises(OverflowError, pollster.register, 0, -1)
126n/a self.assertRaises(OverflowError, pollster.register, 0, 1 << 64)
127n/a self.assertRaises(OverflowError, pollster.modify, 1, -1)
128n/a self.assertRaises(OverflowError, pollster.modify, 1, 1 << 64)
129n/a
130n/a @cpython_only
131n/a def test_events_mask_overflow_c_limits(self):
132n/a from _testcapi import USHRT_MAX
133n/a pollster = select.devpoll()
134n/a w, r = os.pipe()
135n/a pollster.register(w)
136n/a # Issue #17919
137n/a self.assertRaises(OverflowError, pollster.register, 0, USHRT_MAX + 1)
138n/a self.assertRaises(OverflowError, pollster.modify, 1, USHRT_MAX + 1)
139n/a
140n/a
141n/adef test_main():
142n/a run_unittest(DevPollTests)
143n/a
144n/aif __name__ == '__main__':
145n/a test_main()