Python code coverage for Lib/test/test_asynchat.py

#	count	content
1	n/a	# test asynchat
2	n/a
3	n/a	from test import support
4	n/a
5	n/a	# If this fails, the test will be skipped.
6	n/a	thread = support.import_module('_thread')
7	n/a
8	n/a	import asynchat
9	n/a	import asyncore
10	n/a	import errno
11	n/a	import socket
12	n/a	import sys
13	n/a	import time
14	n/a	import unittest
15	n/a	import unittest.mock
16	n/a	try:
17	n/a	import threading
18	n/a	except ImportError:
19	n/a	threading = None
20	n/a
21	n/a	HOST = support.HOST
22	n/a	SERVER_QUIT = b'QUIT\n'
23	n/a	TIMEOUT = 3.0
24	n/a
25	n/a	if threading:
26	n/a	class echo_server(threading.Thread):
27	n/a	# parameter to determine the number of bytes passed back to the
28	n/a	# client each send
29	n/a	chunk_size = 1
30	n/a
31	n/a	def __init__(self, event):
32	n/a	threading.Thread.__init__(self)
33	n/a	self.event = event
34	n/a	self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
35	n/a	self.port = support.bind_port(self.sock)
36	n/a	# This will be set if the client wants us to wait before echoing
37	n/a	# data back.
38	n/a	self.start_resend_event = None
39	n/a
40	n/a	def run(self):
41	n/a	self.sock.listen()
42	n/a	self.event.set()
43	n/a	conn, client = self.sock.accept()
44	n/a	self.buffer = b""
45	n/a	# collect data until quit message is seen
46	n/a	while SERVER_QUIT not in self.buffer:
47	n/a	data = conn.recv(1)
48	n/a	if not data:
49	n/a	break
50	n/a	self.buffer = self.buffer + data
51	n/a
52	n/a	# remove the SERVER_QUIT message
53	n/a	self.buffer = self.buffer.replace(SERVER_QUIT, b'')
54	n/a
55	n/a	if self.start_resend_event:
56	n/a	self.start_resend_event.wait()
57	n/a
58	n/a	# re-send entire set of collected data
59	n/a	try:
60	n/a	# this may fail on some tests, such as test_close_when_done,
61	n/a	# since the client closes the channel when it's done sending
62	n/a	while self.buffer:
63	n/a	n = conn.send(self.buffer[:self.chunk_size])
64	n/a	time.sleep(0.001)
65	n/a	self.buffer = self.buffer[n:]
66	n/a	except:
67	n/a	pass
68	n/a
69	n/a	conn.close()
70	n/a	self.sock.close()
71	n/a
72	n/a	class echo_client(asynchat.async_chat):
73	n/a
74	n/a	def __init__(self, terminator, server_port):
75	n/a	asynchat.async_chat.__init__(self)
76	n/a	self.contents = []
77	n/a	self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
78	n/a	self.connect((HOST, server_port))
79	n/a	self.set_terminator(terminator)
80	n/a	self.buffer = b""
81	n/a
82	n/a	def handle_connect(self):
83	n/a	pass
84	n/a
85	n/a	if sys.platform == 'darwin':
86	n/a	# select.poll returns a select.POLLHUP at the end of the tests
87	n/a	# on darwin, so just ignore it
88	n/a	def handle_expt(self):
89	n/a	pass
90	n/a
91	n/a	def collect_incoming_data(self, data):
92	n/a	self.buffer += data
93	n/a
94	n/a	def found_terminator(self):
95	n/a	self.contents.append(self.buffer)
96	n/a	self.buffer = b""
97	n/a
98	n/a	def start_echo_server():
99	n/a	event = threading.Event()
100	n/a	s = echo_server(event)
101	n/a	s.start()
102	n/a	event.wait()
103	n/a	event.clear()
104	n/a	time.sleep(0.01) # Give server time to start accepting.
105	n/a	return s, event
106	n/a
107	n/a
108	n/a	@unittest.skipUnless(threading, 'Threading required for this test.')
109	n/a	class TestAsynchat(unittest.TestCase):
110	n/a	usepoll = False
111	n/a
112	n/a	def setUp(self):
113	n/a	self._threads = support.threading_setup()
114	n/a
115	n/a	def tearDown(self):
116	n/a	support.threading_cleanup(*self._threads)
117	n/a
118	n/a	def line_terminator_check(self, term, server_chunk):
119	n/a	event = threading.Event()
120	n/a	s = echo_server(event)
121	n/a	s.chunk_size = server_chunk
122	n/a	s.start()
123	n/a	event.wait()
124	n/a	event.clear()
125	n/a	time.sleep(0.01) # Give server time to start accepting.
126	n/a	c = echo_client(term, s.port)
127	n/a	c.push(b"hello ")
128	n/a	c.push(b"world" + term)
129	n/a	c.push(b"I'm not dead yet!" + term)
130	n/a	c.push(SERVER_QUIT)
131	n/a	asyncore.loop(use_poll=self.usepoll, count=300, timeout=.01)
132	n/a	s.join(timeout=TIMEOUT)
133	n/a	if s.is_alive():
134	n/a	self.fail("join() timed out")
135	n/a
136	n/a	self.assertEqual(c.contents, [b"hello world", b"I'm not dead yet!"])
137	n/a
138	n/a	# the line terminator tests below check receiving variously-sized
139	n/a	# chunks back from the server in order to exercise all branches of
140	n/a	# async_chat.handle_read
141	n/a
142	n/a	def test_line_terminator1(self):
143	n/a	# test one-character terminator
144	n/a	for l in (1, 2, 3):
145	n/a	self.line_terminator_check(b'\n', l)
146	n/a
147	n/a	def test_line_terminator2(self):
148	n/a	# test two-character terminator
149	n/a	for l in (1, 2, 3):
150	n/a	self.line_terminator_check(b'\r\n', l)
151	n/a
152	n/a	def test_line_terminator3(self):
153	n/a	# test three-character terminator
154	n/a	for l in (1, 2, 3):
155	n/a	self.line_terminator_check(b'qqq', l)
156	n/a
157	n/a	def numeric_terminator_check(self, termlen):
158	n/a	# Try reading a fixed number of bytes
159	n/a	s, event = start_echo_server()
160	n/a	c = echo_client(termlen, s.port)
161	n/a	data = b"hello world, I'm not dead yet!\n"
162	n/a	c.push(data)
163	n/a	c.push(SERVER_QUIT)
164	n/a	asyncore.loop(use_poll=self.usepoll, count=300, timeout=.01)
165	n/a	s.join(timeout=TIMEOUT)
166	n/a	if s.is_alive():
167	n/a	self.fail("join() timed out")
168	n/a
169	n/a	self.assertEqual(c.contents, [data[:termlen]])
170	n/a
171	n/a	def test_numeric_terminator1(self):
172	n/a	# check that ints & longs both work (since type is
173	n/a	# explicitly checked in async_chat.handle_read)
174	n/a	self.numeric_terminator_check(1)
175	n/a
176	n/a	def test_numeric_terminator2(self):
177	n/a	self.numeric_terminator_check(6)
178	n/a
179	n/a	def test_none_terminator(self):
180	n/a	# Try reading a fixed number of bytes
181	n/a	s, event = start_echo_server()
182	n/a	c = echo_client(None, s.port)
183	n/a	data = b"hello world, I'm not dead yet!\n"
184	n/a	c.push(data)
185	n/a	c.push(SERVER_QUIT)
186	n/a	asyncore.loop(use_poll=self.usepoll, count=300, timeout=.01)
187	n/a	s.join(timeout=TIMEOUT)
188	n/a	if s.is_alive():
189	n/a	self.fail("join() timed out")
190	n/a
191	n/a	self.assertEqual(c.contents, [])
192	n/a	self.assertEqual(c.buffer, data)
193	n/a
194	n/a	def test_simple_producer(self):
195	n/a	s, event = start_echo_server()
196	n/a	c = echo_client(b'\n', s.port)
197	n/a	data = b"hello world\nI'm not dead yet!\n"
198	n/a	p = asynchat.simple_producer(data+SERVER_QUIT, buffer_size=8)
199	n/a	c.push_with_producer(p)
200	n/a	asyncore.loop(use_poll=self.usepoll, count=300, timeout=.01)
201	n/a	s.join(timeout=TIMEOUT)
202	n/a	if s.is_alive():
203	n/a	self.fail("join() timed out")
204	n/a
205	n/a	self.assertEqual(c.contents, [b"hello world", b"I'm not dead yet!"])
206	n/a
207	n/a	def test_string_producer(self):
208	n/a	s, event = start_echo_server()
209	n/a	c = echo_client(b'\n', s.port)
210	n/a	data = b"hello world\nI'm not dead yet!\n"
211	n/a	c.push_with_producer(data+SERVER_QUIT)
212	n/a	asyncore.loop(use_poll=self.usepoll, count=300, timeout=.01)
213	n/a	s.join(timeout=TIMEOUT)
214	n/a	if s.is_alive():
215	n/a	self.fail("join() timed out")
216	n/a
217	n/a	self.assertEqual(c.contents, [b"hello world", b"I'm not dead yet!"])
218	n/a
219	n/a	def test_empty_line(self):
220	n/a	# checks that empty lines are handled correctly
221	n/a	s, event = start_echo_server()
222	n/a	c = echo_client(b'\n', s.port)
223	n/a	c.push(b"hello world\n\nI'm not dead yet!\n")
224	n/a	c.push(SERVER_QUIT)
225	n/a	asyncore.loop(use_poll=self.usepoll, count=300, timeout=.01)
226	n/a	s.join(timeout=TIMEOUT)
227	n/a	if s.is_alive():
228	n/a	self.fail("join() timed out")
229	n/a
230	n/a	self.assertEqual(c.contents,
231	n/a	[b"hello world", b"", b"I'm not dead yet!"])
232	n/a
233	n/a	def test_close_when_done(self):
234	n/a	s, event = start_echo_server()
235	n/a	s.start_resend_event = threading.Event()
236	n/a	c = echo_client(b'\n', s.port)
237	n/a	c.push(b"hello world\nI'm not dead yet!\n")
238	n/a	c.push(SERVER_QUIT)
239	n/a	c.close_when_done()
240	n/a	asyncore.loop(use_poll=self.usepoll, count=300, timeout=.01)
241	n/a
242	n/a	# Only allow the server to start echoing data back to the client after
243	n/a	# the client has closed its connection. This prevents a race condition
244	n/a	# where the server echoes all of its data before we can check that it
245	n/a	# got any down below.
246	n/a	s.start_resend_event.set()
247	n/a	s.join(timeout=TIMEOUT)
248	n/a	if s.is_alive():
249	n/a	self.fail("join() timed out")
250	n/a
251	n/a	self.assertEqual(c.contents, [])
252	n/a	# the server might have been able to send a byte or two back, but this
253	n/a	# at least checks that it received something and didn't just fail
254	n/a	# (which could still result in the client not having received anything)
255	n/a	self.assertGreater(len(s.buffer), 0)
256	n/a
257	n/a	def test_push(self):
258	n/a	# Issue #12523: push() should raise a TypeError if it doesn't get
259	n/a	# a bytes string
260	n/a	s, event = start_echo_server()
261	n/a	c = echo_client(b'\n', s.port)
262	n/a	data = b'bytes\n'
263	n/a	c.push(data)
264	n/a	c.push(bytearray(data))
265	n/a	c.push(memoryview(data))
266	n/a	self.assertRaises(TypeError, c.push, 10)
267	n/a	self.assertRaises(TypeError, c.push, 'unicode')
268	n/a	c.push(SERVER_QUIT)
269	n/a	asyncore.loop(use_poll=self.usepoll, count=300, timeout=.01)
270	n/a	s.join(timeout=TIMEOUT)
271	n/a	self.assertEqual(c.contents, [b'bytes', b'bytes', b'bytes'])
272	n/a
273	n/a
274	n/a	class TestAsynchat_WithPoll(TestAsynchat):
275	n/a	usepoll = True
276	n/a
277	n/a
278	n/a	class TestAsynchatMocked(unittest.TestCase):
279	n/a	def test_blockingioerror(self):
280	n/a	# Issue #16133: handle_read() must ignore BlockingIOError
281	n/a	sock = unittest.mock.Mock()
282	n/a	sock.recv.side_effect = BlockingIOError(errno.EAGAIN)
283	n/a
284	n/a	dispatcher = asynchat.async_chat()
285	n/a	dispatcher.set_socket(sock)
286	n/a	self.addCleanup(dispatcher.del_channel)
287	n/a
288	n/a	with unittest.mock.patch.object(dispatcher, 'handle_error') as error:
289	n/a	dispatcher.handle_read()
290	n/a	self.assertFalse(error.called)
291	n/a
292	n/a
293	n/a	class TestHelperFunctions(unittest.TestCase):
294	n/a	def test_find_prefix_at_end(self):
295	n/a	self.assertEqual(asynchat.find_prefix_at_end("qwerty\r", "\r\n"), 1)
296	n/a	self.assertEqual(asynchat.find_prefix_at_end("qwertydkjf", "\r\n"), 0)
297	n/a
298	n/a
299	n/a	class TestNotConnected(unittest.TestCase):
300	n/a	def test_disallow_negative_terminator(self):
301	n/a	# Issue #11259
302	n/a	client = asynchat.async_chat()
303	n/a	self.assertRaises(ValueError, client.set_terminator, -1)
304	n/a
305	n/a
306	n/a
307	n/a	if __name__ == "__main__":
308	n/a	unittest.main()