Python code coverage for Lib/test/test_devpoll.py

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