Python code coverage for Lib/test/test_poll.py

#	count	content
1	n/a	# Test case for the os.poll() function
2	n/a
3	n/a	import os
4	n/a	import subprocess
5	n/a	import random
6	n/a	import select
7	n/a	try:
8	n/a	import threading
9	n/a	except ImportError:
10	n/a	threading = None
11	n/a	import time
12	n/a	import unittest
13	n/a	from test.support import TESTFN, run_unittest, reap_threads, cpython_only
14	n/a
15	n/a	try:
16	n/a	select.poll
17	n/a	except AttributeError:
18	n/a	raise unittest.SkipTest("select.poll not defined")
19	n/a
20	n/a
21	n/a	def find_ready_matching(ready, flag):
22	n/a	match = []
23	n/a	for fd, mode in ready:
24	n/a	if mode & flag:
25	n/a	match.append(fd)
26	n/a	return match
27	n/a
28	n/a	class PollTests(unittest.TestCase):
29	n/a
30	n/a	def test_poll1(self):
31	n/a	# Basic functional test of poll object
32	n/a	# Create a bunch of pipe and test that poll works with them.
33	n/a
34	n/a	p = select.poll()
35	n/a
36	n/a	NUM_PIPES = 12
37	n/a	MSG = b" This is a test."
38	n/a	MSG_LEN = len(MSG)
39	n/a	readers = []
40	n/a	writers = []
41	n/a	r2w = {}
42	n/a	w2r = {}
43	n/a
44	n/a	for i in range(NUM_PIPES):
45	n/a	rd, wr = os.pipe()
46	n/a	p.register(rd)
47	n/a	p.modify(rd, select.POLLIN)
48	n/a	p.register(wr, select.POLLOUT)
49	n/a	readers.append(rd)
50	n/a	writers.append(wr)
51	n/a	r2w[rd] = wr
52	n/a	w2r[wr] = rd
53	n/a
54	n/a	bufs = []
55	n/a
56	n/a	while writers:
57	n/a	ready = p.poll()
58	n/a	ready_writers = find_ready_matching(ready, select.POLLOUT)
59	n/a	if not ready_writers:
60	n/a	raise RuntimeError("no pipes ready for writing")
61	n/a	wr = random.choice(ready_writers)
62	n/a	os.write(wr, MSG)
63	n/a
64	n/a	ready = p.poll()
65	n/a	ready_readers = find_ready_matching(ready, select.POLLIN)
66	n/a	if not ready_readers:
67	n/a	raise RuntimeError("no pipes ready for reading")
68	n/a	rd = random.choice(ready_readers)
69	n/a	buf = os.read(rd, MSG_LEN)
70	n/a	self.assertEqual(len(buf), MSG_LEN)
71	n/a	bufs.append(buf)
72	n/a	os.close(r2w[rd]) ; os.close( rd )
73	n/a	p.unregister( r2w[rd] )
74	n/a	p.unregister( rd )
75	n/a	writers.remove(r2w[rd])
76	n/a
77	n/a	self.assertEqual(bufs, [MSG] * NUM_PIPES)
78	n/a
79	n/a	def test_poll_unit_tests(self):
80	n/a	# returns NVAL for invalid file descriptor
81	n/a	FD, w = os.pipe()
82	n/a	os.close(FD)
83	n/a	os.close(w)
84	n/a	p = select.poll()
85	n/a	p.register(FD)
86	n/a	r = p.poll()
87	n/a	self.assertEqual(r[0], (FD, select.POLLNVAL))
88	n/a
89	n/a	f = open(TESTFN, 'w')
90	n/a	fd = f.fileno()
91	n/a	p = select.poll()
92	n/a	p.register(f)
93	n/a	r = p.poll()
94	n/a	self.assertEqual(r[0][0], fd)
95	n/a	f.close()
96	n/a	r = p.poll()
97	n/a	self.assertEqual(r[0], (fd, select.POLLNVAL))
98	n/a	os.unlink(TESTFN)
99	n/a
100	n/a	# type error for invalid arguments
101	n/a	p = select.poll()
102	n/a	self.assertRaises(TypeError, p.register, p)
103	n/a	self.assertRaises(TypeError, p.unregister, p)
104	n/a
105	n/a	# can't unregister non-existent object
106	n/a	p = select.poll()
107	n/a	self.assertRaises(KeyError, p.unregister, 3)
108	n/a
109	n/a	# Test error cases
110	n/a	pollster = select.poll()
111	n/a	class Nope:
112	n/a	pass
113	n/a
114	n/a	class Almost:
115	n/a	def fileno(self):
116	n/a	return 'fileno'
117	n/a
118	n/a	self.assertRaises(TypeError, pollster.register, Nope(), 0)
119	n/a	self.assertRaises(TypeError, pollster.register, Almost(), 0)
120	n/a
121	n/a	# Another test case for poll(). This is copied from the test case for
122	n/a	# select(), modified to use poll() instead.
123	n/a
124	n/a	def test_poll2(self):
125	n/a	cmd = 'for i in 0 1 2 3 4 5 6 7 8 9; do echo testing...; sleep 1; done'
126	n/a	proc = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE,
127	n/a	bufsize=0)
128	n/a	proc.__enter__()
129	n/a	self.addCleanup(proc.__exit__, None, None, None)
130	n/a	p = proc.stdout
131	n/a	pollster = select.poll()
132	n/a	pollster.register( p, select.POLLIN )
133	n/a	for tout in (0, 1000, 2000, 4000, 8000, 16000) + (-1,)*10:
134	n/a	fdlist = pollster.poll(tout)
135	n/a	if (fdlist == []):
136	n/a	continue
137	n/a	fd, flags = fdlist[0]
138	n/a	if flags & select.POLLHUP:
139	n/a	line = p.readline()
140	n/a	if line != b"":
141	n/a	self.fail('error: pipe seems to be closed, but still returns data')
142	n/a	continue
143	n/a
144	n/a	elif flags & select.POLLIN:
145	n/a	line = p.readline()
146	n/a	if not line:
147	n/a	break
148	n/a	self.assertEqual(line, b'testing...\n')
149	n/a	continue
150	n/a	else:
151	n/a	self.fail('Unexpected return value from select.poll: %s' % fdlist)
152	n/a
153	n/a	def test_poll3(self):
154	n/a	# test int overflow
155	n/a	pollster = select.poll()
156	n/a	pollster.register(1)
157	n/a
158	n/a	self.assertRaises(OverflowError, pollster.poll, 1 << 64)
159	n/a
160	n/a	x = 2 + 3
161	n/a	if x != 5:
162	n/a	self.fail('Overflow must have occurred')
163	n/a
164	n/a	# Issues #15989, #17919
165	n/a	self.assertRaises(OverflowError, pollster.register, 0, -1)
166	n/a	self.assertRaises(OverflowError, pollster.register, 0, 1 << 64)
167	n/a	self.assertRaises(OverflowError, pollster.modify, 1, -1)
168	n/a	self.assertRaises(OverflowError, pollster.modify, 1, 1 << 64)
169	n/a
170	n/a	@cpython_only
171	n/a	def test_poll_c_limits(self):
172	n/a	from _testcapi import USHRT_MAX, INT_MAX, UINT_MAX
173	n/a	pollster = select.poll()
174	n/a	pollster.register(1)
175	n/a
176	n/a	# Issues #15989, #17919
177	n/a	self.assertRaises(OverflowError, pollster.register, 0, USHRT_MAX + 1)
178	n/a	self.assertRaises(OverflowError, pollster.modify, 1, USHRT_MAX + 1)
179	n/a	self.assertRaises(OverflowError, pollster.poll, INT_MAX + 1)
180	n/a	self.assertRaises(OverflowError, pollster.poll, UINT_MAX + 1)
181	n/a
182	n/a	@unittest.skipUnless(threading, 'Threading required for this test.')
183	n/a	@reap_threads
184	n/a	def test_threaded_poll(self):
185	n/a	r, w = os.pipe()
186	n/a	self.addCleanup(os.close, r)
187	n/a	self.addCleanup(os.close, w)
188	n/a	rfds = []
189	n/a	for i in range(10):
190	n/a	fd = os.dup(r)
191	n/a	self.addCleanup(os.close, fd)
192	n/a	rfds.append(fd)
193	n/a	pollster = select.poll()
194	n/a	for fd in rfds:
195	n/a	pollster.register(fd, select.POLLIN)
196	n/a
197	n/a	t = threading.Thread(target=pollster.poll)
198	n/a	t.start()
199	n/a	try:
200	n/a	time.sleep(0.5)
201	n/a	# trigger ufds array reallocation
202	n/a	for fd in rfds:
203	n/a	pollster.unregister(fd)
204	n/a	pollster.register(w, select.POLLOUT)
205	n/a	self.assertRaises(RuntimeError, pollster.poll)
206	n/a	finally:
207	n/a	# and make the call to poll() from the thread return
208	n/a	os.write(w, b'spam')
209	n/a	t.join()
210	n/a
211	n/a
212	n/a	def test_main():
213	n/a	run_unittest(PollTests)
214	n/a
215	n/a	if __name__ == '__main__':
216	n/a	test_main()