Python code coverage for Lib/test/test_asyncio/test_streams.py

#	count	content
1	n/a	"""Tests for streams.py."""
2	n/a
3	n/a	import gc
4	n/a	import os
5	n/a	import queue
6	n/a	import socket
7	n/a	import sys
8	n/a	import threading
9	n/a	import unittest
10	n/a	from unittest import mock
11	n/a	try:
12	n/a	import ssl
13	n/a	except ImportError:
14	n/a	ssl = None
15	n/a
16	n/a	import asyncio
17	n/a	from asyncio import test_utils
18	n/a
19	n/a
20	n/a	class StreamReaderTests(test_utils.TestCase):
21	n/a
22	n/a	DATA = b'line1\nline2\nline3\n'
23	n/a
24	n/a	def setUp(self):
25	n/a	super().setUp()
26	n/a	self.loop = asyncio.new_event_loop()
27	n/a	self.set_event_loop(self.loop)
28	n/a
29	n/a	def tearDown(self):
30	n/a	# just in case if we have transport close callbacks
31	n/a	test_utils.run_briefly(self.loop)
32	n/a
33	n/a	self.loop.close()
34	n/a	gc.collect()
35	n/a	super().tearDown()
36	n/a
37	n/a	@mock.patch('asyncio.streams.events')
38	n/a	def test_ctor_global_loop(self, m_events):
39	n/a	stream = asyncio.StreamReader()
40	n/a	self.assertIs(stream._loop, m_events.get_event_loop.return_value)
41	n/a
42	n/a	def _basetest_open_connection(self, open_connection_fut):
43	n/a	reader, writer = self.loop.run_until_complete(open_connection_fut)
44	n/a	writer.write(b'GET / HTTP/1.0\r\n\r\n')
45	n/a	f = reader.readline()
46	n/a	data = self.loop.run_until_complete(f)
47	n/a	self.assertEqual(data, b'HTTP/1.0 200 OK\r\n')
48	n/a	f = reader.read()
49	n/a	data = self.loop.run_until_complete(f)
50	n/a	self.assertTrue(data.endswith(b'\r\n\r\nTest message'))
51	n/a	writer.close()
52	n/a
53	n/a	def test_open_connection(self):
54	n/a	with test_utils.run_test_server() as httpd:
55	n/a	conn_fut = asyncio.open_connection(*httpd.address,
56	n/a	loop=self.loop)
57	n/a	self._basetest_open_connection(conn_fut)
58	n/a
59	n/a	@unittest.skipUnless(hasattr(socket, 'AF_UNIX'), 'No UNIX Sockets')
60	n/a	def test_open_unix_connection(self):
61	n/a	with test_utils.run_test_unix_server() as httpd:
62	n/a	conn_fut = asyncio.open_unix_connection(httpd.address,
63	n/a	loop=self.loop)
64	n/a	self._basetest_open_connection(conn_fut)
65	n/a
66	n/a	def _basetest_open_connection_no_loop_ssl(self, open_connection_fut):
67	n/a	try:
68	n/a	reader, writer = self.loop.run_until_complete(open_connection_fut)
69	n/a	finally:
70	n/a	asyncio.set_event_loop(None)
71	n/a	writer.write(b'GET / HTTP/1.0\r\n\r\n')
72	n/a	f = reader.read()
73	n/a	data = self.loop.run_until_complete(f)
74	n/a	self.assertTrue(data.endswith(b'\r\n\r\nTest message'))
75	n/a
76	n/a	writer.close()
77	n/a
78	n/a	@unittest.skipIf(ssl is None, 'No ssl module')
79	n/a	def test_open_connection_no_loop_ssl(self):
80	n/a	with test_utils.run_test_server(use_ssl=True) as httpd:
81	n/a	conn_fut = asyncio.open_connection(
82	n/a	*httpd.address,
83	n/a	ssl=test_utils.dummy_ssl_context(),
84	n/a	loop=self.loop)
85	n/a
86	n/a	self._basetest_open_connection_no_loop_ssl(conn_fut)
87	n/a
88	n/a	@unittest.skipIf(ssl is None, 'No ssl module')
89	n/a	@unittest.skipUnless(hasattr(socket, 'AF_UNIX'), 'No UNIX Sockets')
90	n/a	def test_open_unix_connection_no_loop_ssl(self):
91	n/a	with test_utils.run_test_unix_server(use_ssl=True) as httpd:
92	n/a	conn_fut = asyncio.open_unix_connection(
93	n/a	httpd.address,
94	n/a	ssl=test_utils.dummy_ssl_context(),
95	n/a	server_hostname='',
96	n/a	loop=self.loop)
97	n/a
98	n/a	self._basetest_open_connection_no_loop_ssl(conn_fut)
99	n/a
100	n/a	def _basetest_open_connection_error(self, open_connection_fut):
101	n/a	reader, writer = self.loop.run_until_complete(open_connection_fut)
102	n/a	writer._protocol.connection_lost(ZeroDivisionError())
103	n/a	f = reader.read()
104	n/a	with self.assertRaises(ZeroDivisionError):
105	n/a	self.loop.run_until_complete(f)
106	n/a	writer.close()
107	n/a	test_utils.run_briefly(self.loop)
108	n/a
109	n/a	def test_open_connection_error(self):
110	n/a	with test_utils.run_test_server() as httpd:
111	n/a	conn_fut = asyncio.open_connection(*httpd.address,
112	n/a	loop=self.loop)
113	n/a	self._basetest_open_connection_error(conn_fut)
114	n/a
115	n/a	@unittest.skipUnless(hasattr(socket, 'AF_UNIX'), 'No UNIX Sockets')
116	n/a	def test_open_unix_connection_error(self):
117	n/a	with test_utils.run_test_unix_server() as httpd:
118	n/a	conn_fut = asyncio.open_unix_connection(httpd.address,
119	n/a	loop=self.loop)
120	n/a	self._basetest_open_connection_error(conn_fut)
121	n/a
122	n/a	def test_feed_empty_data(self):
123	n/a	stream = asyncio.StreamReader(loop=self.loop)
124	n/a
125	n/a	stream.feed_data(b'')
126	n/a	self.assertEqual(b'', stream._buffer)
127	n/a
128	n/a	def test_feed_nonempty_data(self):
129	n/a	stream = asyncio.StreamReader(loop=self.loop)
130	n/a
131	n/a	stream.feed_data(self.DATA)
132	n/a	self.assertEqual(self.DATA, stream._buffer)
133	n/a
134	n/a	def test_read_zero(self):
135	n/a	# Read zero bytes.
136	n/a	stream = asyncio.StreamReader(loop=self.loop)
137	n/a	stream.feed_data(self.DATA)
138	n/a
139	n/a	data = self.loop.run_until_complete(stream.read(0))
140	n/a	self.assertEqual(b'', data)
141	n/a	self.assertEqual(self.DATA, stream._buffer)
142	n/a
143	n/a	def test_read(self):
144	n/a	# Read bytes.
145	n/a	stream = asyncio.StreamReader(loop=self.loop)
146	n/a	read_task = asyncio.Task(stream.read(30), loop=self.loop)
147	n/a
148	n/a	def cb():
149	n/a	stream.feed_data(self.DATA)
150	n/a	self.loop.call_soon(cb)
151	n/a
152	n/a	data = self.loop.run_until_complete(read_task)
153	n/a	self.assertEqual(self.DATA, data)
154	n/a	self.assertEqual(b'', stream._buffer)
155	n/a
156	n/a	def test_read_line_breaks(self):
157	n/a	# Read bytes without line breaks.
158	n/a	stream = asyncio.StreamReader(loop=self.loop)
159	n/a	stream.feed_data(b'line1')
160	n/a	stream.feed_data(b'line2')
161	n/a
162	n/a	data = self.loop.run_until_complete(stream.read(5))
163	n/a
164	n/a	self.assertEqual(b'line1', data)
165	n/a	self.assertEqual(b'line2', stream._buffer)
166	n/a
167	n/a	def test_read_eof(self):
168	n/a	# Read bytes, stop at eof.
169	n/a	stream = asyncio.StreamReader(loop=self.loop)
170	n/a	read_task = asyncio.Task(stream.read(1024), loop=self.loop)
171	n/a
172	n/a	def cb():
173	n/a	stream.feed_eof()
174	n/a	self.loop.call_soon(cb)
175	n/a
176	n/a	data = self.loop.run_until_complete(read_task)
177	n/a	self.assertEqual(b'', data)
178	n/a	self.assertEqual(b'', stream._buffer)
179	n/a
180	n/a	def test_read_until_eof(self):
181	n/a	# Read all bytes until eof.
182	n/a	stream = asyncio.StreamReader(loop=self.loop)
183	n/a	read_task = asyncio.Task(stream.read(-1), loop=self.loop)
184	n/a
185	n/a	def cb():
186	n/a	stream.feed_data(b'chunk1\n')
187	n/a	stream.feed_data(b'chunk2')
188	n/a	stream.feed_eof()
189	n/a	self.loop.call_soon(cb)
190	n/a
191	n/a	data = self.loop.run_until_complete(read_task)
192	n/a
193	n/a	self.assertEqual(b'chunk1\nchunk2', data)
194	n/a	self.assertEqual(b'', stream._buffer)
195	n/a
196	n/a	def test_read_exception(self):
197	n/a	stream = asyncio.StreamReader(loop=self.loop)
198	n/a	stream.feed_data(b'line\n')
199	n/a
200	n/a	data = self.loop.run_until_complete(stream.read(2))
201	n/a	self.assertEqual(b'li', data)
202	n/a
203	n/a	stream.set_exception(ValueError())
204	n/a	self.assertRaises(
205	n/a	ValueError, self.loop.run_until_complete, stream.read(2))
206	n/a
207	n/a	def test_invalid_limit(self):
208	n/a	with self.assertRaisesRegex(ValueError, 'imit'):
209	n/a	asyncio.StreamReader(limit=0, loop=self.loop)
210	n/a
211	n/a	with self.assertRaisesRegex(ValueError, 'imit'):
212	n/a	asyncio.StreamReader(limit=-1, loop=self.loop)
213	n/a
214	n/a	def test_read_limit(self):
215	n/a	stream = asyncio.StreamReader(limit=3, loop=self.loop)
216	n/a	stream.feed_data(b'chunk')
217	n/a	data = self.loop.run_until_complete(stream.read(5))
218	n/a	self.assertEqual(b'chunk', data)
219	n/a	self.assertEqual(b'', stream._buffer)
220	n/a
221	n/a	def test_readline(self):
222	n/a	# Read one line. 'readline' will need to wait for the data
223	n/a	# to come from 'cb'
224	n/a	stream = asyncio.StreamReader(loop=self.loop)
225	n/a	stream.feed_data(b'chunk1 ')
226	n/a	read_task = asyncio.Task(stream.readline(), loop=self.loop)
227	n/a
228	n/a	def cb():
229	n/a	stream.feed_data(b'chunk2 ')
230	n/a	stream.feed_data(b'chunk3 ')
231	n/a	stream.feed_data(b'\n chunk4')
232	n/a	self.loop.call_soon(cb)
233	n/a
234	n/a	line = self.loop.run_until_complete(read_task)
235	n/a	self.assertEqual(b'chunk1 chunk2 chunk3 \n', line)
236	n/a	self.assertEqual(b' chunk4', stream._buffer)
237	n/a
238	n/a	def test_readline_limit_with_existing_data(self):
239	n/a	# Read one line. The data is in StreamReader's buffer
240	n/a	# before the event loop is run.
241	n/a
242	n/a	stream = asyncio.StreamReader(limit=3, loop=self.loop)
243	n/a	stream.feed_data(b'li')
244	n/a	stream.feed_data(b'ne1\nline2\n')
245	n/a
246	n/a	self.assertRaises(
247	n/a	ValueError, self.loop.run_until_complete, stream.readline())
248	n/a	# The buffer should contain the remaining data after exception
249	n/a	self.assertEqual(b'line2\n', stream._buffer)
250	n/a
251	n/a	stream = asyncio.StreamReader(limit=3, loop=self.loop)
252	n/a	stream.feed_data(b'li')
253	n/a	stream.feed_data(b'ne1')
254	n/a	stream.feed_data(b'li')
255	n/a
256	n/a	self.assertRaises(
257	n/a	ValueError, self.loop.run_until_complete, stream.readline())
258	n/a	# No b'\n' at the end. The 'limit' is set to 3. So before
259	n/a	# waiting for the new data in buffer, 'readline' will consume
260	n/a	# the entire buffer, and since the length of the consumed data
261	n/a	# is more than 3, it will raise a ValueError. The buffer is
262	n/a	# expected to be empty now.
263	n/a	self.assertEqual(b'', stream._buffer)
264	n/a
265	n/a	def test_at_eof(self):
266	n/a	stream = asyncio.StreamReader(loop=self.loop)
267	n/a	self.assertFalse(stream.at_eof())
268	n/a
269	n/a	stream.feed_data(b'some data\n')
270	n/a	self.assertFalse(stream.at_eof())
271	n/a
272	n/a	self.loop.run_until_complete(stream.readline())
273	n/a	self.assertFalse(stream.at_eof())
274	n/a
275	n/a	stream.feed_data(b'some data\n')
276	n/a	stream.feed_eof()
277	n/a	self.loop.run_until_complete(stream.readline())
278	n/a	self.assertTrue(stream.at_eof())
279	n/a
280	n/a	def test_readline_limit(self):
281	n/a	# Read one line. StreamReaders are fed with data after
282	n/a	# their 'readline' methods are called.
283	n/a
284	n/a	stream = asyncio.StreamReader(limit=7, loop=self.loop)
285	n/a	def cb():
286	n/a	stream.feed_data(b'chunk1')
287	n/a	stream.feed_data(b'chunk2')
288	n/a	stream.feed_data(b'chunk3\n')
289	n/a	stream.feed_eof()
290	n/a	self.loop.call_soon(cb)
291	n/a
292	n/a	self.assertRaises(
293	n/a	ValueError, self.loop.run_until_complete, stream.readline())
294	n/a	# The buffer had just one line of data, and after raising
295	n/a	# a ValueError it should be empty.
296	n/a	self.assertEqual(b'', stream._buffer)
297	n/a
298	n/a	stream = asyncio.StreamReader(limit=7, loop=self.loop)
299	n/a	def cb():
300	n/a	stream.feed_data(b'chunk1')
301	n/a	stream.feed_data(b'chunk2\n')
302	n/a	stream.feed_data(b'chunk3\n')
303	n/a	stream.feed_eof()
304	n/a	self.loop.call_soon(cb)
305	n/a
306	n/a	self.assertRaises(
307	n/a	ValueError, self.loop.run_until_complete, stream.readline())
308	n/a	self.assertEqual(b'chunk3\n', stream._buffer)
309	n/a
310	n/a	# check strictness of the limit
311	n/a	stream = asyncio.StreamReader(limit=7, loop=self.loop)
312	n/a	stream.feed_data(b'1234567\n')
313	n/a	line = self.loop.run_until_complete(stream.readline())
314	n/a	self.assertEqual(b'1234567\n', line)
315	n/a	self.assertEqual(b'', stream._buffer)
316	n/a
317	n/a	stream.feed_data(b'12345678\n')
318	n/a	with self.assertRaises(ValueError) as cm:
319	n/a	self.loop.run_until_complete(stream.readline())
320	n/a	self.assertEqual(b'', stream._buffer)
321	n/a
322	n/a	stream.feed_data(b'12345678')
323	n/a	with self.assertRaises(ValueError) as cm:
324	n/a	self.loop.run_until_complete(stream.readline())
325	n/a	self.assertEqual(b'', stream._buffer)
326	n/a
327	n/a	def test_readline_nolimit_nowait(self):
328	n/a	# All needed data for the first 'readline' call will be
329	n/a	# in the buffer.
330	n/a	stream = asyncio.StreamReader(loop=self.loop)
331	n/a	stream.feed_data(self.DATA[:6])
332	n/a	stream.feed_data(self.DATA[6:])
333	n/a
334	n/a	line = self.loop.run_until_complete(stream.readline())
335	n/a
336	n/a	self.assertEqual(b'line1\n', line)
337	n/a	self.assertEqual(b'line2\nline3\n', stream._buffer)
338	n/a
339	n/a	def test_readline_eof(self):
340	n/a	stream = asyncio.StreamReader(loop=self.loop)
341	n/a	stream.feed_data(b'some data')
342	n/a	stream.feed_eof()
343	n/a
344	n/a	line = self.loop.run_until_complete(stream.readline())
345	n/a	self.assertEqual(b'some data', line)
346	n/a
347	n/a	def test_readline_empty_eof(self):
348	n/a	stream = asyncio.StreamReader(loop=self.loop)
349	n/a	stream.feed_eof()
350	n/a
351	n/a	line = self.loop.run_until_complete(stream.readline())
352	n/a	self.assertEqual(b'', line)
353	n/a
354	n/a	def test_readline_read_byte_count(self):
355	n/a	stream = asyncio.StreamReader(loop=self.loop)
356	n/a	stream.feed_data(self.DATA)
357	n/a
358	n/a	self.loop.run_until_complete(stream.readline())
359	n/a
360	n/a	data = self.loop.run_until_complete(stream.read(7))
361	n/a
362	n/a	self.assertEqual(b'line2\nl', data)
363	n/a	self.assertEqual(b'ine3\n', stream._buffer)
364	n/a
365	n/a	def test_readline_exception(self):
366	n/a	stream = asyncio.StreamReader(loop=self.loop)
367	n/a	stream.feed_data(b'line\n')
368	n/a
369	n/a	data = self.loop.run_until_complete(stream.readline())
370	n/a	self.assertEqual(b'line\n', data)
371	n/a
372	n/a	stream.set_exception(ValueError())
373	n/a	self.assertRaises(
374	n/a	ValueError, self.loop.run_until_complete, stream.readline())
375	n/a	self.assertEqual(b'', stream._buffer)
376	n/a
377	n/a	def test_readuntil_separator(self):
378	n/a	stream = asyncio.StreamReader(loop=self.loop)
379	n/a	with self.assertRaisesRegex(ValueError, 'Separator should be'):
380	n/a	self.loop.run_until_complete(stream.readuntil(separator=b''))
381	n/a
382	n/a	def test_readuntil_multi_chunks(self):
383	n/a	stream = asyncio.StreamReader(loop=self.loop)
384	n/a
385	n/a	stream.feed_data(b'lineAAA')
386	n/a	data = self.loop.run_until_complete(stream.readuntil(separator=b'AAA'))
387	n/a	self.assertEqual(b'lineAAA', data)
388	n/a	self.assertEqual(b'', stream._buffer)
389	n/a
390	n/a	stream.feed_data(b'lineAAA')
391	n/a	data = self.loop.run_until_complete(stream.readuntil(b'AAA'))
392	n/a	self.assertEqual(b'lineAAA', data)
393	n/a	self.assertEqual(b'', stream._buffer)
394	n/a
395	n/a	stream.feed_data(b'lineAAAxxx')
396	n/a	data = self.loop.run_until_complete(stream.readuntil(b'AAA'))
397	n/a	self.assertEqual(b'lineAAA', data)
398	n/a	self.assertEqual(b'xxx', stream._buffer)
399	n/a
400	n/a	def test_readuntil_multi_chunks_1(self):
401	n/a	stream = asyncio.StreamReader(loop=self.loop)
402	n/a
403	n/a	stream.feed_data(b'QWEaa')
404	n/a	stream.feed_data(b'XYaa')
405	n/a	stream.feed_data(b'a')
406	n/a	data = self.loop.run_until_complete(stream.readuntil(b'aaa'))
407	n/a	self.assertEqual(b'QWEaaXYaaa', data)
408	n/a	self.assertEqual(b'', stream._buffer)
409	n/a
410	n/a	stream.feed_data(b'QWEaa')
411	n/a	stream.feed_data(b'XYa')
412	n/a	stream.feed_data(b'aa')
413	n/a	data = self.loop.run_until_complete(stream.readuntil(b'aaa'))
414	n/a	self.assertEqual(b'QWEaaXYaaa', data)
415	n/a	self.assertEqual(b'', stream._buffer)
416	n/a
417	n/a	stream.feed_data(b'aaa')
418	n/a	data = self.loop.run_until_complete(stream.readuntil(b'aaa'))
419	n/a	self.assertEqual(b'aaa', data)
420	n/a	self.assertEqual(b'', stream._buffer)
421	n/a
422	n/a	stream.feed_data(b'Xaaa')
423	n/a	data = self.loop.run_until_complete(stream.readuntil(b'aaa'))
424	n/a	self.assertEqual(b'Xaaa', data)
425	n/a	self.assertEqual(b'', stream._buffer)
426	n/a
427	n/a	stream.feed_data(b'XXX')
428	n/a	stream.feed_data(b'a')
429	n/a	stream.feed_data(b'a')
430	n/a	stream.feed_data(b'a')
431	n/a	data = self.loop.run_until_complete(stream.readuntil(b'aaa'))
432	n/a	self.assertEqual(b'XXXaaa', data)
433	n/a	self.assertEqual(b'', stream._buffer)
434	n/a
435	n/a	def test_readuntil_eof(self):
436	n/a	stream = asyncio.StreamReader(loop=self.loop)
437	n/a	stream.feed_data(b'some dataAA')
438	n/a	stream.feed_eof()
439	n/a
440	n/a	with self.assertRaises(asyncio.IncompleteReadError) as cm:
441	n/a	self.loop.run_until_complete(stream.readuntil(b'AAA'))
442	n/a	self.assertEqual(cm.exception.partial, b'some dataAA')
443	n/a	self.assertIsNone(cm.exception.expected)
444	n/a	self.assertEqual(b'', stream._buffer)
445	n/a
446	n/a	def test_readuntil_limit_found_sep(self):
447	n/a	stream = asyncio.StreamReader(loop=self.loop, limit=3)
448	n/a	stream.feed_data(b'some dataAA')
449	n/a
450	n/a	with self.assertRaisesRegex(asyncio.LimitOverrunError,
451	n/a	'not found') as cm:
452	n/a	self.loop.run_until_complete(stream.readuntil(b'AAA'))
453	n/a
454	n/a	self.assertEqual(b'some dataAA', stream._buffer)
455	n/a
456	n/a	stream.feed_data(b'A')
457	n/a	with self.assertRaisesRegex(asyncio.LimitOverrunError,
458	n/a	'is found') as cm:
459	n/a	self.loop.run_until_complete(stream.readuntil(b'AAA'))
460	n/a
461	n/a	self.assertEqual(b'some dataAAA', stream._buffer)
462	n/a
463	n/a	def test_readexactly_zero_or_less(self):
464	n/a	# Read exact number of bytes (zero or less).
465	n/a	stream = asyncio.StreamReader(loop=self.loop)
466	n/a	stream.feed_data(self.DATA)
467	n/a
468	n/a	data = self.loop.run_until_complete(stream.readexactly(0))
469	n/a	self.assertEqual(b'', data)
470	n/a	self.assertEqual(self.DATA, stream._buffer)
471	n/a
472	n/a	with self.assertRaisesRegex(ValueError, 'less than zero'):
473	n/a	self.loop.run_until_complete(stream.readexactly(-1))
474	n/a	self.assertEqual(self.DATA, stream._buffer)
475	n/a
476	n/a	def test_readexactly(self):
477	n/a	# Read exact number of bytes.
478	n/a	stream = asyncio.StreamReader(loop=self.loop)
479	n/a
480	n/a	n = 2 * len(self.DATA)
481	n/a	read_task = asyncio.Task(stream.readexactly(n), loop=self.loop)
482	n/a
483	n/a	def cb():
484	n/a	stream.feed_data(self.DATA)
485	n/a	stream.feed_data(self.DATA)
486	n/a	stream.feed_data(self.DATA)
487	n/a	self.loop.call_soon(cb)
488	n/a
489	n/a	data = self.loop.run_until_complete(read_task)
490	n/a	self.assertEqual(self.DATA + self.DATA, data)
491	n/a	self.assertEqual(self.DATA, stream._buffer)
492	n/a
493	n/a	def test_readexactly_limit(self):
494	n/a	stream = asyncio.StreamReader(limit=3, loop=self.loop)
495	n/a	stream.feed_data(b'chunk')
496	n/a	data = self.loop.run_until_complete(stream.readexactly(5))
497	n/a	self.assertEqual(b'chunk', data)
498	n/a	self.assertEqual(b'', stream._buffer)
499	n/a
500	n/a	def test_readexactly_eof(self):
501	n/a	# Read exact number of bytes (eof).
502	n/a	stream = asyncio.StreamReader(loop=self.loop)
503	n/a	n = 2 * len(self.DATA)
504	n/a	read_task = asyncio.Task(stream.readexactly(n), loop=self.loop)
505	n/a
506	n/a	def cb():
507	n/a	stream.feed_data(self.DATA)
508	n/a	stream.feed_eof()
509	n/a	self.loop.call_soon(cb)
510	n/a
511	n/a	with self.assertRaises(asyncio.IncompleteReadError) as cm:
512	n/a	self.loop.run_until_complete(read_task)
513	n/a	self.assertEqual(cm.exception.partial, self.DATA)
514	n/a	self.assertEqual(cm.exception.expected, n)
515	n/a	self.assertEqual(str(cm.exception),
516	n/a	'18 bytes read on a total of 36 expected bytes')
517	n/a	self.assertEqual(b'', stream._buffer)
518	n/a
519	n/a	def test_readexactly_exception(self):
520	n/a	stream = asyncio.StreamReader(loop=self.loop)
521	n/a	stream.feed_data(b'line\n')
522	n/a
523	n/a	data = self.loop.run_until_complete(stream.readexactly(2))
524	n/a	self.assertEqual(b'li', data)
525	n/a
526	n/a	stream.set_exception(ValueError())
527	n/a	self.assertRaises(
528	n/a	ValueError, self.loop.run_until_complete, stream.readexactly(2))
529	n/a
530	n/a	def test_exception(self):
531	n/a	stream = asyncio.StreamReader(loop=self.loop)
532	n/a	self.assertIsNone(stream.exception())
533	n/a
534	n/a	exc = ValueError()
535	n/a	stream.set_exception(exc)
536	n/a	self.assertIs(stream.exception(), exc)
537	n/a
538	n/a	def test_exception_waiter(self):
539	n/a	stream = asyncio.StreamReader(loop=self.loop)
540	n/a
541	n/a	@asyncio.coroutine
542	n/a	def set_err():
543	n/a	stream.set_exception(ValueError())
544	n/a
545	n/a	t1 = asyncio.Task(stream.readline(), loop=self.loop)
546	n/a	t2 = asyncio.Task(set_err(), loop=self.loop)
547	n/a
548	n/a	self.loop.run_until_complete(asyncio.wait([t1, t2], loop=self.loop))
549	n/a
550	n/a	self.assertRaises(ValueError, t1.result)
551	n/a
552	n/a	def test_exception_cancel(self):
553	n/a	stream = asyncio.StreamReader(loop=self.loop)
554	n/a
555	n/a	t = asyncio.Task(stream.readline(), loop=self.loop)
556	n/a	test_utils.run_briefly(self.loop)
557	n/a	t.cancel()
558	n/a	test_utils.run_briefly(self.loop)
559	n/a	# The following line fails if set_exception() isn't careful.
560	n/a	stream.set_exception(RuntimeError('message'))
561	n/a	test_utils.run_briefly(self.loop)
562	n/a	self.assertIs(stream._waiter, None)
563	n/a
564	n/a	def test_start_server(self):
565	n/a
566	n/a	class MyServer:
567	n/a
568	n/a	def __init__(self, loop):
569	n/a	self.server = None
570	n/a	self.loop = loop
571	n/a
572	n/a	@asyncio.coroutine
573	n/a	def handle_client(self, client_reader, client_writer):
574	n/a	data = yield from client_reader.readline()
575	n/a	client_writer.write(data)
576	n/a	yield from client_writer.drain()
577	n/a	client_writer.close()
578	n/a
579	n/a	def start(self):
580	n/a	sock = socket.socket()
581	n/a	sock.bind(('127.0.0.1', 0))
582	n/a	self.server = self.loop.run_until_complete(
583	n/a	asyncio.start_server(self.handle_client,
584	n/a	sock=sock,
585	n/a	loop=self.loop))
586	n/a	return sock.getsockname()
587	n/a
588	n/a	def handle_client_callback(self, client_reader, client_writer):
589	n/a	self.loop.create_task(self.handle_client(client_reader,
590	n/a	client_writer))
591	n/a
592	n/a	def start_callback(self):
593	n/a	sock = socket.socket()
594	n/a	sock.bind(('127.0.0.1', 0))
595	n/a	addr = sock.getsockname()
596	n/a	sock.close()
597	n/a	self.server = self.loop.run_until_complete(
598	n/a	asyncio.start_server(self.handle_client_callback,
599	n/a	host=addr[0], port=addr[1],
600	n/a	loop=self.loop))
601	n/a	return addr
602	n/a
603	n/a	def stop(self):
604	n/a	if self.server is not None:
605	n/a	self.server.close()
606	n/a	self.loop.run_until_complete(self.server.wait_closed())
607	n/a	self.server = None
608	n/a
609	n/a	@asyncio.coroutine
610	n/a	def client(addr):
611	n/a	reader, writer = yield from asyncio.open_connection(
612	n/a	*addr, loop=self.loop)
613	n/a	# send a line
614	n/a	writer.write(b"hello world!\n")
615	n/a	# read it back
616	n/a	msgback = yield from reader.readline()
617	n/a	writer.close()
618	n/a	return msgback
619	n/a
620	n/a	# test the server variant with a coroutine as client handler
621	n/a	server = MyServer(self.loop)
622	n/a	addr = server.start()
623	n/a	msg = self.loop.run_until_complete(asyncio.Task(client(addr),
624	n/a	loop=self.loop))
625	n/a	server.stop()
626	n/a	self.assertEqual(msg, b"hello world!\n")
627	n/a
628	n/a	# test the server variant with a callback as client handler
629	n/a	server = MyServer(self.loop)
630	n/a	addr = server.start_callback()
631	n/a	msg = self.loop.run_until_complete(asyncio.Task(client(addr),
632	n/a	loop=self.loop))
633	n/a	server.stop()
634	n/a	self.assertEqual(msg, b"hello world!\n")
635	n/a
636	n/a	@unittest.skipUnless(hasattr(socket, 'AF_UNIX'), 'No UNIX Sockets')
637	n/a	def test_start_unix_server(self):
638	n/a
639	n/a	class MyServer:
640	n/a
641	n/a	def __init__(self, loop, path):
642	n/a	self.server = None
643	n/a	self.loop = loop
644	n/a	self.path = path
645	n/a
646	n/a	@asyncio.coroutine
647	n/a	def handle_client(self, client_reader, client_writer):
648	n/a	data = yield from client_reader.readline()
649	n/a	client_writer.write(data)
650	n/a	yield from client_writer.drain()
651	n/a	client_writer.close()
652	n/a
653	n/a	def start(self):
654	n/a	self.server = self.loop.run_until_complete(
655	n/a	asyncio.start_unix_server(self.handle_client,
656	n/a	path=self.path,
657	n/a	loop=self.loop))
658	n/a
659	n/a	def handle_client_callback(self, client_reader, client_writer):
660	n/a	self.loop.create_task(self.handle_client(client_reader,
661	n/a	client_writer))
662	n/a
663	n/a	def start_callback(self):
664	n/a	start = asyncio.start_unix_server(self.handle_client_callback,
665	n/a	path=self.path,
666	n/a	loop=self.loop)
667	n/a	self.server = self.loop.run_until_complete(start)
668	n/a
669	n/a	def stop(self):
670	n/a	if self.server is not None:
671	n/a	self.server.close()
672	n/a	self.loop.run_until_complete(self.server.wait_closed())
673	n/a	self.server = None
674	n/a
675	n/a	@asyncio.coroutine
676	n/a	def client(path):
677	n/a	reader, writer = yield from asyncio.open_unix_connection(
678	n/a	path, loop=self.loop)
679	n/a	# send a line
680	n/a	writer.write(b"hello world!\n")
681	n/a	# read it back
682	n/a	msgback = yield from reader.readline()
683	n/a	writer.close()
684	n/a	return msgback
685	n/a
686	n/a	# test the server variant with a coroutine as client handler
687	n/a	with test_utils.unix_socket_path() as path:
688	n/a	server = MyServer(self.loop, path)
689	n/a	server.start()
690	n/a	msg = self.loop.run_until_complete(asyncio.Task(client(path),
691	n/a	loop=self.loop))
692	n/a	server.stop()
693	n/a	self.assertEqual(msg, b"hello world!\n")
694	n/a
695	n/a	# test the server variant with a callback as client handler
696	n/a	with test_utils.unix_socket_path() as path:
697	n/a	server = MyServer(self.loop, path)
698	n/a	server.start_callback()
699	n/a	msg = self.loop.run_until_complete(asyncio.Task(client(path),
700	n/a	loop=self.loop))
701	n/a	server.stop()
702	n/a	self.assertEqual(msg, b"hello world!\n")
703	n/a
704	n/a	@unittest.skipIf(sys.platform == 'win32', "Don't have pipes")
705	n/a	def test_read_all_from_pipe_reader(self):
706	n/a	# See asyncio issue 168. This test is derived from the example
707	n/a	# subprocess_attach_read_pipe.py, but we configure the
708	n/a	# StreamReader's limit so that twice it is less than the size
709	n/a	# of the data writter. Also we must explicitly attach a child
710	n/a	# watcher to the event loop.
711	n/a
712	n/a	code = """\
713	n/a	import os, sys
714	n/a	fd = int(sys.argv[1])
715	n/a	os.write(fd, b'data')
716	n/a	os.close(fd)
717	n/a	"""
718	n/a	rfd, wfd = os.pipe()
719	n/a	args = [sys.executable, '-c', code, str(wfd)]
720	n/a
721	n/a	pipe = open(rfd, 'rb', 0)
722	n/a	reader = asyncio.StreamReader(loop=self.loop, limit=1)
723	n/a	protocol = asyncio.StreamReaderProtocol(reader, loop=self.loop)
724	n/a	transport, _ = self.loop.run_until_complete(
725	n/a	self.loop.connect_read_pipe(lambda: protocol, pipe))
726	n/a
727	n/a	watcher = asyncio.SafeChildWatcher()
728	n/a	watcher.attach_loop(self.loop)
729	n/a	try:
730	n/a	asyncio.set_child_watcher(watcher)
731	n/a	create = asyncio.create_subprocess_exec(*args,
732	n/a	pass_fds={wfd},
733	n/a	loop=self.loop)
734	n/a	proc = self.loop.run_until_complete(create)
735	n/a	self.loop.run_until_complete(proc.wait())
736	n/a	finally:
737	n/a	asyncio.set_child_watcher(None)
738	n/a
739	n/a	os.close(wfd)
740	n/a	data = self.loop.run_until_complete(reader.read(-1))
741	n/a	self.assertEqual(data, b'data')
742	n/a
743	n/a	def test_streamreader_constructor(self):
744	n/a	self.addCleanup(asyncio.set_event_loop, None)
745	n/a	asyncio.set_event_loop(self.loop)
746	n/a
747	n/a	# asyncio issue #184: Ensure that StreamReaderProtocol constructor
748	n/a	# retrieves the current loop if the loop parameter is not set
749	n/a	reader = asyncio.StreamReader()
750	n/a	self.assertIs(reader._loop, self.loop)
751	n/a
752	n/a	def test_streamreaderprotocol_constructor(self):
753	n/a	self.addCleanup(asyncio.set_event_loop, None)
754	n/a	asyncio.set_event_loop(self.loop)
755	n/a
756	n/a	# asyncio issue #184: Ensure that StreamReaderProtocol constructor
757	n/a	# retrieves the current loop if the loop parameter is not set
758	n/a	reader = mock.Mock()
759	n/a	protocol = asyncio.StreamReaderProtocol(reader)
760	n/a	self.assertIs(protocol._loop, self.loop)
761	n/a
762	n/a	def test_drain_raises(self):
763	n/a	# See http://bugs.python.org/issue25441
764	n/a
765	n/a	# This test should not use asyncio for the mock server; the
766	n/a	# whole point of the test is to test for a bug in drain()
767	n/a	# where it never gives up the event loop but the socket is
768	n/a	# closed on the server side.
769	n/a
770	n/a	q = queue.Queue()
771	n/a
772	n/a	def server():
773	n/a	# Runs in a separate thread.
774	n/a	sock = socket.socket()
775	n/a	with sock:
776	n/a	sock.bind(('localhost', 0))
777	n/a	sock.listen(1)
778	n/a	addr = sock.getsockname()
779	n/a	q.put(addr)
780	n/a	clt, _ = sock.accept()
781	n/a	clt.close()
782	n/a
783	n/a	@asyncio.coroutine
784	n/a	def client(host, port):
785	n/a	reader, writer = yield from asyncio.open_connection(
786	n/a	host, port, loop=self.loop)
787	n/a
788	n/a	while True:
789	n/a	writer.write(b"foo\n")
790	n/a	yield from writer.drain()
791	n/a
792	n/a	# Start the server thread and wait for it to be listening.
793	n/a	thread = threading.Thread(target=server)
794	n/a	thread.setDaemon(True)
795	n/a	thread.start()
796	n/a	addr = q.get()
797	n/a
798	n/a	# Should not be stuck in an infinite loop.
799	n/a	with self.assertRaises((ConnectionResetError, BrokenPipeError)):
800	n/a	self.loop.run_until_complete(client(*addr))
801	n/a
802	n/a	# Clean up the thread. (Only on success; on failure, it may
803	n/a	# be stuck in accept().)
804	n/a	thread.join()
805	n/a
806	n/a	def test___repr__(self):
807	n/a	stream = asyncio.StreamReader(loop=self.loop)
808	n/a	self.assertEqual("<StreamReader>", repr(stream))
809	n/a
810	n/a	def test___repr__nondefault_limit(self):
811	n/a	stream = asyncio.StreamReader(loop=self.loop, limit=123)
812	n/a	self.assertEqual("<StreamReader l=123>", repr(stream))
813	n/a
814	n/a	def test___repr__eof(self):
815	n/a	stream = asyncio.StreamReader(loop=self.loop)
816	n/a	stream.feed_eof()
817	n/a	self.assertEqual("<StreamReader eof>", repr(stream))
818	n/a
819	n/a	def test___repr__data(self):
820	n/a	stream = asyncio.StreamReader(loop=self.loop)
821	n/a	stream.feed_data(b'data')
822	n/a	self.assertEqual("<StreamReader 4 bytes>", repr(stream))
823	n/a
824	n/a	def test___repr__exception(self):
825	n/a	stream = asyncio.StreamReader(loop=self.loop)
826	n/a	exc = RuntimeError()
827	n/a	stream.set_exception(exc)
828	n/a	self.assertEqual("<StreamReader e=RuntimeError()>", repr(stream))
829	n/a
830	n/a	def test___repr__waiter(self):
831	n/a	stream = asyncio.StreamReader(loop=self.loop)
832	n/a	stream._waiter = asyncio.Future(loop=self.loop)
833	n/a	self.assertRegex(
834	n/a	repr(stream),
835	n/a	r"<StreamReader w=<Future pending[\S ]*>>")
836	n/a	stream._waiter.set_result(None)
837	n/a	self.loop.run_until_complete(stream._waiter)
838	n/a	stream._waiter = None
839	n/a	self.assertEqual("<StreamReader>", repr(stream))
840	n/a
841	n/a	def test___repr__transport(self):
842	n/a	stream = asyncio.StreamReader(loop=self.loop)
843	n/a	stream._transport = mock.Mock()
844	n/a	stream._transport.__repr__ = mock.Mock()
845	n/a	stream._transport.__repr__.return_value = "<Transport>"
846	n/a	self.assertEqual("<StreamReader t=<Transport>>", repr(stream))
847	n/a
848	n/a
849	n/a	if __name__ == '__main__':
850	n/a	unittest.main()