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

#	count	content
1	n/a	"""Tests for events.py."""
2	n/a
3	n/a	import collections.abc
4	n/a	import functools
5	n/a	import gc
6	n/a	import io
7	n/a	import os
8	n/a	import platform
9	n/a	import re
10	n/a	import signal
11	n/a	import socket
12	n/a	try:
13	n/a	import ssl
14	n/a	except ImportError:
15	n/a	ssl = None
16	n/a	import subprocess
17	n/a	import sys
18	n/a	import threading
19	n/a	import time
20	n/a	import errno
21	n/a	import unittest
22	n/a	from unittest import mock
23	n/a	import weakref
24	n/a
25	n/a	if sys.platform != 'win32':
26	n/a	import tty
27	n/a
28	n/a	import asyncio
29	n/a	from asyncio import coroutines
30	n/a	from asyncio import proactor_events
31	n/a	from asyncio import selector_events
32	n/a	from asyncio import sslproto
33	n/a	from asyncio import test_utils
34	n/a	try:
35	n/a	from test import support
36	n/a	except ImportError:
37	n/a	from asyncio import test_support as support
38	n/a
39	n/a
40	n/a	def data_file(filename):
41	n/a	if hasattr(support, 'TEST_HOME_DIR'):
42	n/a	fullname = os.path.join(support.TEST_HOME_DIR, filename)
43	n/a	if os.path.isfile(fullname):
44	n/a	return fullname
45	n/a	fullname = os.path.join(os.path.dirname(__file__), filename)
46	n/a	if os.path.isfile(fullname):
47	n/a	return fullname
48	n/a	raise FileNotFoundError(filename)
49	n/a
50	n/a
51	n/a	def osx_tiger():
52	n/a	"""Return True if the platform is Mac OS 10.4 or older."""
53	n/a	if sys.platform != 'darwin':
54	n/a	return False
55	n/a	version = platform.mac_ver()[0]
56	n/a	version = tuple(map(int, version.split('.')))
57	n/a	return version < (10, 5)
58	n/a
59	n/a
60	n/a	ONLYCERT = data_file('ssl_cert.pem')
61	n/a	ONLYKEY = data_file('ssl_key.pem')
62	n/a	SIGNED_CERTFILE = data_file('keycert3.pem')
63	n/a	SIGNING_CA = data_file('pycacert.pem')
64	n/a	PEERCERT = {'serialNumber': 'B09264B1F2DA21D1',
65	n/a	'version': 1,
66	n/a	'subject': ((('countryName', 'XY'),),
67	n/a	(('localityName', 'Castle Anthrax'),),
68	n/a	(('organizationName', 'Python Software Foundation'),),
69	n/a	(('commonName', 'localhost'),)),
70	n/a	'issuer': ((('countryName', 'XY'),),
71	n/a	(('organizationName', 'Python Software Foundation CA'),),
72	n/a	(('commonName', 'our-ca-server'),)),
73	n/a	'notAfter': 'Nov 13 19:47:07 2022 GMT',
74	n/a	'notBefore': 'Jan 4 19:47:07 2013 GMT'}
75	n/a
76	n/a
77	n/a	class MyBaseProto(asyncio.Protocol):
78	n/a	connected = None
79	n/a	done = None
80	n/a
81	n/a	def __init__(self, loop=None):
82	n/a	self.transport = None
83	n/a	self.state = 'INITIAL'
84	n/a	self.nbytes = 0
85	n/a	if loop is not None:
86	n/a	self.connected = asyncio.Future(loop=loop)
87	n/a	self.done = asyncio.Future(loop=loop)
88	n/a
89	n/a	def connection_made(self, transport):
90	n/a	self.transport = transport
91	n/a	assert self.state == 'INITIAL', self.state
92	n/a	self.state = 'CONNECTED'
93	n/a	if self.connected:
94	n/a	self.connected.set_result(None)
95	n/a
96	n/a	def data_received(self, data):
97	n/a	assert self.state == 'CONNECTED', self.state
98	n/a	self.nbytes += len(data)
99	n/a
100	n/a	def eof_received(self):
101	n/a	assert self.state == 'CONNECTED', self.state
102	n/a	self.state = 'EOF'
103	n/a
104	n/a	def connection_lost(self, exc):
105	n/a	assert self.state in ('CONNECTED', 'EOF'), self.state
106	n/a	self.state = 'CLOSED'
107	n/a	if self.done:
108	n/a	self.done.set_result(None)
109	n/a
110	n/a
111	n/a	class MyProto(MyBaseProto):
112	n/a	def connection_made(self, transport):
113	n/a	super().connection_made(transport)
114	n/a	transport.write(b'GET / HTTP/1.0\r\nHost: example.com\r\n\r\n')
115	n/a
116	n/a
117	n/a	class MyDatagramProto(asyncio.DatagramProtocol):
118	n/a	done = None
119	n/a
120	n/a	def __init__(self, loop=None):
121	n/a	self.state = 'INITIAL'
122	n/a	self.nbytes = 0
123	n/a	if loop is not None:
124	n/a	self.done = asyncio.Future(loop=loop)
125	n/a
126	n/a	def connection_made(self, transport):
127	n/a	self.transport = transport
128	n/a	assert self.state == 'INITIAL', self.state
129	n/a	self.state = 'INITIALIZED'
130	n/a
131	n/a	def datagram_received(self, data, addr):
132	n/a	assert self.state == 'INITIALIZED', self.state
133	n/a	self.nbytes += len(data)
134	n/a
135	n/a	def error_received(self, exc):
136	n/a	assert self.state == 'INITIALIZED', self.state
137	n/a
138	n/a	def connection_lost(self, exc):
139	n/a	assert self.state == 'INITIALIZED', self.state
140	n/a	self.state = 'CLOSED'
141	n/a	if self.done:
142	n/a	self.done.set_result(None)
143	n/a
144	n/a
145	n/a	class MyReadPipeProto(asyncio.Protocol):
146	n/a	done = None
147	n/a
148	n/a	def __init__(self, loop=None):
149	n/a	self.state = ['INITIAL']
150	n/a	self.nbytes = 0
151	n/a	self.transport = None
152	n/a	if loop is not None:
153	n/a	self.done = asyncio.Future(loop=loop)
154	n/a
155	n/a	def connection_made(self, transport):
156	n/a	self.transport = transport
157	n/a	assert self.state == ['INITIAL'], self.state
158	n/a	self.state.append('CONNECTED')
159	n/a
160	n/a	def data_received(self, data):
161	n/a	assert self.state == ['INITIAL', 'CONNECTED'], self.state
162	n/a	self.nbytes += len(data)
163	n/a
164	n/a	def eof_received(self):
165	n/a	assert self.state == ['INITIAL', 'CONNECTED'], self.state
166	n/a	self.state.append('EOF')
167	n/a
168	n/a	def connection_lost(self, exc):
169	n/a	if 'EOF' not in self.state:
170	n/a	self.state.append('EOF') # It is okay if EOF is missed.
171	n/a	assert self.state == ['INITIAL', 'CONNECTED', 'EOF'], self.state
172	n/a	self.state.append('CLOSED')
173	n/a	if self.done:
174	n/a	self.done.set_result(None)
175	n/a
176	n/a
177	n/a	class MyWritePipeProto(asyncio.BaseProtocol):
178	n/a	done = None
179	n/a
180	n/a	def __init__(self, loop=None):
181	n/a	self.state = 'INITIAL'
182	n/a	self.transport = None
183	n/a	if loop is not None:
184	n/a	self.done = asyncio.Future(loop=loop)
185	n/a
186	n/a	def connection_made(self, transport):
187	n/a	self.transport = transport
188	n/a	assert self.state == 'INITIAL', self.state
189	n/a	self.state = 'CONNECTED'
190	n/a
191	n/a	def connection_lost(self, exc):
192	n/a	assert self.state == 'CONNECTED', self.state
193	n/a	self.state = 'CLOSED'
194	n/a	if self.done:
195	n/a	self.done.set_result(None)
196	n/a
197	n/a
198	n/a	class MySubprocessProtocol(asyncio.SubprocessProtocol):
199	n/a
200	n/a	def __init__(self, loop):
201	n/a	self.state = 'INITIAL'
202	n/a	self.transport = None
203	n/a	self.connected = asyncio.Future(loop=loop)
204	n/a	self.completed = asyncio.Future(loop=loop)
205	n/a	self.disconnects = {fd: asyncio.Future(loop=loop) for fd in range(3)}
206	n/a	self.data = {1: b'', 2: b''}
207	n/a	self.returncode = None
208	n/a	self.got_data = {1: asyncio.Event(loop=loop),
209	n/a	2: asyncio.Event(loop=loop)}
210	n/a
211	n/a	def connection_made(self, transport):
212	n/a	self.transport = transport
213	n/a	assert self.state == 'INITIAL', self.state
214	n/a	self.state = 'CONNECTED'
215	n/a	self.connected.set_result(None)
216	n/a
217	n/a	def connection_lost(self, exc):
218	n/a	assert self.state == 'CONNECTED', self.state
219	n/a	self.state = 'CLOSED'
220	n/a	self.completed.set_result(None)
221	n/a
222	n/a	def pipe_data_received(self, fd, data):
223	n/a	assert self.state == 'CONNECTED', self.state
224	n/a	self.data[fd] += data
225	n/a	self.got_data[fd].set()
226	n/a
227	n/a	def pipe_connection_lost(self, fd, exc):
228	n/a	assert self.state == 'CONNECTED', self.state
229	n/a	if exc:
230	n/a	self.disconnects[fd].set_exception(exc)
231	n/a	else:
232	n/a	self.disconnects[fd].set_result(exc)
233	n/a
234	n/a	def process_exited(self):
235	n/a	assert self.state == 'CONNECTED', self.state
236	n/a	self.returncode = self.transport.get_returncode()
237	n/a
238	n/a
239	n/a	class EventLoopTestsMixin:
240	n/a
241	n/a	def setUp(self):
242	n/a	super().setUp()
243	n/a	self.loop = self.create_event_loop()
244	n/a	self.set_event_loop(self.loop)
245	n/a
246	n/a	def tearDown(self):
247	n/a	# just in case if we have transport close callbacks
248	n/a	if not self.loop.is_closed():
249	n/a	test_utils.run_briefly(self.loop)
250	n/a
251	n/a	self.loop.close()
252	n/a	gc.collect()
253	n/a	super().tearDown()
254	n/a
255	n/a	def test_run_until_complete_nesting(self):
256	n/a	@asyncio.coroutine
257	n/a	def coro1():
258	n/a	yield
259	n/a
260	n/a	@asyncio.coroutine
261	n/a	def coro2():
262	n/a	self.assertTrue(self.loop.is_running())
263	n/a	self.loop.run_until_complete(coro1())
264	n/a
265	n/a	self.assertRaises(
266	n/a	RuntimeError, self.loop.run_until_complete, coro2())
267	n/a
268	n/a	# Note: because of the default Windows timing granularity of
269	n/a	# 15.6 msec, we use fairly long sleep times here (~100 msec).
270	n/a
271	n/a	def test_run_until_complete(self):
272	n/a	t0 = self.loop.time()
273	n/a	self.loop.run_until_complete(asyncio.sleep(0.1, loop=self.loop))
274	n/a	t1 = self.loop.time()
275	n/a	self.assertTrue(0.08 <= t1-t0 <= 0.8, t1-t0)
276	n/a
277	n/a	def test_run_until_complete_stopped(self):
278	n/a	@asyncio.coroutine
279	n/a	def cb():
280	n/a	self.loop.stop()
281	n/a	yield from asyncio.sleep(0.1, loop=self.loop)
282	n/a	task = cb()
283	n/a	self.assertRaises(RuntimeError,
284	n/a	self.loop.run_until_complete, task)
285	n/a
286	n/a	def test_call_later(self):
287	n/a	results = []
288	n/a
289	n/a	def callback(arg):
290	n/a	results.append(arg)
291	n/a	self.loop.stop()
292	n/a
293	n/a	self.loop.call_later(0.1, callback, 'hello world')
294	n/a	t0 = time.monotonic()
295	n/a	self.loop.run_forever()
296	n/a	t1 = time.monotonic()
297	n/a	self.assertEqual(results, ['hello world'])
298	n/a	self.assertTrue(0.08 <= t1-t0 <= 0.8, t1-t0)
299	n/a
300	n/a	def test_call_soon(self):
301	n/a	results = []
302	n/a
303	n/a	def callback(arg1, arg2):
304	n/a	results.append((arg1, arg2))
305	n/a	self.loop.stop()
306	n/a
307	n/a	self.loop.call_soon(callback, 'hello', 'world')
308	n/a	self.loop.run_forever()
309	n/a	self.assertEqual(results, [('hello', 'world')])
310	n/a
311	n/a	def test_call_soon_threadsafe(self):
312	n/a	results = []
313	n/a	lock = threading.Lock()
314	n/a
315	n/a	def callback(arg):
316	n/a	results.append(arg)
317	n/a	if len(results) >= 2:
318	n/a	self.loop.stop()
319	n/a
320	n/a	def run_in_thread():
321	n/a	self.loop.call_soon_threadsafe(callback, 'hello')
322	n/a	lock.release()
323	n/a
324	n/a	lock.acquire()
325	n/a	t = threading.Thread(target=run_in_thread)
326	n/a	t.start()
327	n/a
328	n/a	with lock:
329	n/a	self.loop.call_soon(callback, 'world')
330	n/a	self.loop.run_forever()
331	n/a	t.join()
332	n/a	self.assertEqual(results, ['hello', 'world'])
333	n/a
334	n/a	def test_call_soon_threadsafe_same_thread(self):
335	n/a	results = []
336	n/a
337	n/a	def callback(arg):
338	n/a	results.append(arg)
339	n/a	if len(results) >= 2:
340	n/a	self.loop.stop()
341	n/a
342	n/a	self.loop.call_soon_threadsafe(callback, 'hello')
343	n/a	self.loop.call_soon(callback, 'world')
344	n/a	self.loop.run_forever()
345	n/a	self.assertEqual(results, ['hello', 'world'])
346	n/a
347	n/a	def test_run_in_executor(self):
348	n/a	def run(arg):
349	n/a	return (arg, threading.get_ident())
350	n/a	f2 = self.loop.run_in_executor(None, run, 'yo')
351	n/a	res, thread_id = self.loop.run_until_complete(f2)
352	n/a	self.assertEqual(res, 'yo')
353	n/a	self.assertNotEqual(thread_id, threading.get_ident())
354	n/a
355	n/a	def test_reader_callback(self):
356	n/a	r, w = test_utils.socketpair()
357	n/a	r.setblocking(False)
358	n/a	bytes_read = bytearray()
359	n/a
360	n/a	def reader():
361	n/a	try:
362	n/a	data = r.recv(1024)
363	n/a	except BlockingIOError:
364	n/a	# Spurious readiness notifications are possible
365	n/a	# at least on Linux -- see man select.
366	n/a	return
367	n/a	if data:
368	n/a	bytes_read.extend(data)
369	n/a	else:
370	n/a	self.assertTrue(self.loop.remove_reader(r.fileno()))
371	n/a	r.close()
372	n/a
373	n/a	self.loop.add_reader(r.fileno(), reader)
374	n/a	self.loop.call_soon(w.send, b'abc')
375	n/a	test_utils.run_until(self.loop, lambda: len(bytes_read) >= 3)
376	n/a	self.loop.call_soon(w.send, b'def')
377	n/a	test_utils.run_until(self.loop, lambda: len(bytes_read) >= 6)
378	n/a	self.loop.call_soon(w.close)
379	n/a	self.loop.call_soon(self.loop.stop)
380	n/a	self.loop.run_forever()
381	n/a	self.assertEqual(bytes_read, b'abcdef')
382	n/a
383	n/a	def test_writer_callback(self):
384	n/a	r, w = test_utils.socketpair()
385	n/a	w.setblocking(False)
386	n/a
387	n/a	def writer(data):
388	n/a	w.send(data)
389	n/a	self.loop.stop()
390	n/a
391	n/a	data = b'x' * 1024
392	n/a	self.loop.add_writer(w.fileno(), writer, data)
393	n/a	self.loop.run_forever()
394	n/a
395	n/a	self.assertTrue(self.loop.remove_writer(w.fileno()))
396	n/a	self.assertFalse(self.loop.remove_writer(w.fileno()))
397	n/a
398	n/a	w.close()
399	n/a	read = r.recv(len(data) * 2)
400	n/a	r.close()
401	n/a	self.assertEqual(read, data)
402	n/a
403	n/a	def _basetest_sock_client_ops(self, httpd, sock):
404	n/a	if not isinstance(self.loop, proactor_events.BaseProactorEventLoop):
405	n/a	# in debug mode, socket operations must fail
406	n/a	# if the socket is not in blocking mode
407	n/a	self.loop.set_debug(True)
408	n/a	sock.setblocking(True)
409	n/a	with self.assertRaises(ValueError):
410	n/a	self.loop.run_until_complete(
411	n/a	self.loop.sock_connect(sock, httpd.address))
412	n/a	with self.assertRaises(ValueError):
413	n/a	self.loop.run_until_complete(
414	n/a	self.loop.sock_sendall(sock, b'GET / HTTP/1.0\r\n\r\n'))
415	n/a	with self.assertRaises(ValueError):
416	n/a	self.loop.run_until_complete(
417	n/a	self.loop.sock_recv(sock, 1024))
418	n/a	with self.assertRaises(ValueError):
419	n/a	self.loop.run_until_complete(
420	n/a	self.loop.sock_accept(sock))
421	n/a
422	n/a	# test in non-blocking mode
423	n/a	sock.setblocking(False)
424	n/a	self.loop.run_until_complete(
425	n/a	self.loop.sock_connect(sock, httpd.address))
426	n/a	self.loop.run_until_complete(
427	n/a	self.loop.sock_sendall(sock, b'GET / HTTP/1.0\r\n\r\n'))
428	n/a	data = self.loop.run_until_complete(
429	n/a	self.loop.sock_recv(sock, 1024))
430	n/a	# consume data
431	n/a	self.loop.run_until_complete(
432	n/a	self.loop.sock_recv(sock, 1024))
433	n/a	sock.close()
434	n/a	self.assertTrue(data.startswith(b'HTTP/1.0 200 OK'))
435	n/a
436	n/a	def test_sock_client_ops(self):
437	n/a	with test_utils.run_test_server() as httpd:
438	n/a	sock = socket.socket()
439	n/a	self._basetest_sock_client_ops(httpd, sock)
440	n/a
441	n/a	@unittest.skipUnless(hasattr(socket, 'AF_UNIX'), 'No UNIX Sockets')
442	n/a	def test_unix_sock_client_ops(self):
443	n/a	with test_utils.run_test_unix_server() as httpd:
444	n/a	sock = socket.socket(socket.AF_UNIX)
445	n/a	self._basetest_sock_client_ops(httpd, sock)
446	n/a
447	n/a	def test_sock_client_fail(self):
448	n/a	# Make sure that we will get an unused port
449	n/a	address = None
450	n/a	try:
451	n/a	s = socket.socket()
452	n/a	s.bind(('127.0.0.1', 0))
453	n/a	address = s.getsockname()
454	n/a	finally:
455	n/a	s.close()
456	n/a
457	n/a	sock = socket.socket()
458	n/a	sock.setblocking(False)
459	n/a	with self.assertRaises(ConnectionRefusedError):
460	n/a	self.loop.run_until_complete(
461	n/a	self.loop.sock_connect(sock, address))
462	n/a	sock.close()
463	n/a
464	n/a	def test_sock_accept(self):
465	n/a	listener = socket.socket()
466	n/a	listener.setblocking(False)
467	n/a	listener.bind(('127.0.0.1', 0))
468	n/a	listener.listen(1)
469	n/a	client = socket.socket()
470	n/a	client.connect(listener.getsockname())
471	n/a
472	n/a	f = self.loop.sock_accept(listener)
473	n/a	conn, addr = self.loop.run_until_complete(f)
474	n/a	self.assertEqual(conn.gettimeout(), 0)
475	n/a	self.assertEqual(addr, client.getsockname())
476	n/a	self.assertEqual(client.getpeername(), listener.getsockname())
477	n/a	client.close()
478	n/a	conn.close()
479	n/a	listener.close()
480	n/a
481	n/a	@unittest.skipUnless(hasattr(signal, 'SIGKILL'), 'No SIGKILL')
482	n/a	def test_add_signal_handler(self):
483	n/a	caught = 0
484	n/a
485	n/a	def my_handler():
486	n/a	nonlocal caught
487	n/a	caught += 1
488	n/a
489	n/a	# Check error behavior first.
490	n/a	self.assertRaises(
491	n/a	TypeError, self.loop.add_signal_handler, 'boom', my_handler)
492	n/a	self.assertRaises(
493	n/a	TypeError, self.loop.remove_signal_handler, 'boom')
494	n/a	self.assertRaises(
495	n/a	ValueError, self.loop.add_signal_handler, signal.NSIG+1,
496	n/a	my_handler)
497	n/a	self.assertRaises(
498	n/a	ValueError, self.loop.remove_signal_handler, signal.NSIG+1)
499	n/a	self.assertRaises(
500	n/a	ValueError, self.loop.add_signal_handler, 0, my_handler)
501	n/a	self.assertRaises(
502	n/a	ValueError, self.loop.remove_signal_handler, 0)
503	n/a	self.assertRaises(
504	n/a	ValueError, self.loop.add_signal_handler, -1, my_handler)
505	n/a	self.assertRaises(
506	n/a	ValueError, self.loop.remove_signal_handler, -1)
507	n/a	self.assertRaises(
508	n/a	RuntimeError, self.loop.add_signal_handler, signal.SIGKILL,
509	n/a	my_handler)
510	n/a	# Removing SIGKILL doesn't raise, since we don't call signal().
511	n/a	self.assertFalse(self.loop.remove_signal_handler(signal.SIGKILL))
512	n/a	# Now set a handler and handle it.
513	n/a	self.loop.add_signal_handler(signal.SIGINT, my_handler)
514	n/a
515	n/a	os.kill(os.getpid(), signal.SIGINT)
516	n/a	test_utils.run_until(self.loop, lambda: caught)
517	n/a
518	n/a	# Removing it should restore the default handler.
519	n/a	self.assertTrue(self.loop.remove_signal_handler(signal.SIGINT))
520	n/a	self.assertEqual(signal.getsignal(signal.SIGINT),
521	n/a	signal.default_int_handler)
522	n/a	# Removing again returns False.
523	n/a	self.assertFalse(self.loop.remove_signal_handler(signal.SIGINT))
524	n/a
525	n/a	@unittest.skipUnless(hasattr(signal, 'SIGALRM'), 'No SIGALRM')
526	n/a	def test_signal_handling_while_selecting(self):
527	n/a	# Test with a signal actually arriving during a select() call.
528	n/a	caught = 0
529	n/a
530	n/a	def my_handler():
531	n/a	nonlocal caught
532	n/a	caught += 1
533	n/a	self.loop.stop()
534	n/a
535	n/a	self.loop.add_signal_handler(signal.SIGALRM, my_handler)
536	n/a
537	n/a	signal.setitimer(signal.ITIMER_REAL, 0.01, 0) # Send SIGALRM once.
538	n/a	self.loop.run_forever()
539	n/a	self.assertEqual(caught, 1)
540	n/a
541	n/a	@unittest.skipUnless(hasattr(signal, 'SIGALRM'), 'No SIGALRM')
542	n/a	def test_signal_handling_args(self):
543	n/a	some_args = (42,)
544	n/a	caught = 0
545	n/a
546	n/a	def my_handler(*args):
547	n/a	nonlocal caught
548	n/a	caught += 1
549	n/a	self.assertEqual(args, some_args)
550	n/a
551	n/a	self.loop.add_signal_handler(signal.SIGALRM, my_handler, *some_args)
552	n/a
553	n/a	signal.setitimer(signal.ITIMER_REAL, 0.1, 0) # Send SIGALRM once.
554	n/a	self.loop.call_later(0.5, self.loop.stop)
555	n/a	self.loop.run_forever()
556	n/a	self.assertEqual(caught, 1)
557	n/a
558	n/a	def _basetest_create_connection(self, connection_fut, check_sockname=True):
559	n/a	tr, pr = self.loop.run_until_complete(connection_fut)
560	n/a	self.assertIsInstance(tr, asyncio.Transport)
561	n/a	self.assertIsInstance(pr, asyncio.Protocol)
562	n/a	self.assertIs(pr.transport, tr)
563	n/a	if check_sockname:
564	n/a	self.assertIsNotNone(tr.get_extra_info('sockname'))
565	n/a	self.loop.run_until_complete(pr.done)
566	n/a	self.assertGreater(pr.nbytes, 0)
567	n/a	tr.close()
568	n/a
569	n/a	def test_create_connection(self):
570	n/a	with test_utils.run_test_server() as httpd:
571	n/a	conn_fut = self.loop.create_connection(
572	n/a	lambda: MyProto(loop=self.loop), *httpd.address)
573	n/a	self._basetest_create_connection(conn_fut)
574	n/a
575	n/a	@unittest.skipUnless(hasattr(socket, 'AF_UNIX'), 'No UNIX Sockets')
576	n/a	def test_create_unix_connection(self):
577	n/a	# Issue #20682: On Mac OS X Tiger, getsockname() returns a
578	n/a	# zero-length address for UNIX socket.
579	n/a	check_sockname = not osx_tiger()
580	n/a
581	n/a	with test_utils.run_test_unix_server() as httpd:
582	n/a	conn_fut = self.loop.create_unix_connection(
583	n/a	lambda: MyProto(loop=self.loop), httpd.address)
584	n/a	self._basetest_create_connection(conn_fut, check_sockname)
585	n/a
586	n/a	def test_create_connection_sock(self):
587	n/a	with test_utils.run_test_server() as httpd:
588	n/a	sock = None
589	n/a	infos = self.loop.run_until_complete(
590	n/a	self.loop.getaddrinfo(
591	n/a	*httpd.address, type=socket.SOCK_STREAM))
592	n/a	for family, type, proto, cname, address in infos:
593	n/a	try:
594	n/a	sock = socket.socket(family=family, type=type, proto=proto)
595	n/a	sock.setblocking(False)
596	n/a	self.loop.run_until_complete(
597	n/a	self.loop.sock_connect(sock, address))
598	n/a	except:
599	n/a	pass
600	n/a	else:
601	n/a	break
602	n/a	else:
603	n/a	assert False, 'Can not create socket.'
604	n/a
605	n/a	f = self.loop.create_connection(
606	n/a	lambda: MyProto(loop=self.loop), sock=sock)
607	n/a	tr, pr = self.loop.run_until_complete(f)
608	n/a	self.assertIsInstance(tr, asyncio.Transport)
609	n/a	self.assertIsInstance(pr, asyncio.Protocol)
610	n/a	self.loop.run_until_complete(pr.done)
611	n/a	self.assertGreater(pr.nbytes, 0)
612	n/a	tr.close()
613	n/a
614	n/a	def check_ssl_extra_info(self, client, check_sockname=True,
615	n/a	peername=None, peercert={}):
616	n/a	if check_sockname:
617	n/a	self.assertIsNotNone(client.get_extra_info('sockname'))
618	n/a	if peername:
619	n/a	self.assertEqual(peername,
620	n/a	client.get_extra_info('peername'))
621	n/a	else:
622	n/a	self.assertIsNotNone(client.get_extra_info('peername'))
623	n/a	self.assertEqual(peercert,
624	n/a	client.get_extra_info('peercert'))
625	n/a
626	n/a	# test SSL cipher
627	n/a	cipher = client.get_extra_info('cipher')
628	n/a	self.assertIsInstance(cipher, tuple)
629	n/a	self.assertEqual(len(cipher), 3, cipher)
630	n/a	self.assertIsInstance(cipher[0], str)
631	n/a	self.assertIsInstance(cipher[1], str)
632	n/a	self.assertIsInstance(cipher[2], int)
633	n/a
634	n/a	# test SSL object
635	n/a	sslobj = client.get_extra_info('ssl_object')
636	n/a	self.assertIsNotNone(sslobj)
637	n/a	self.assertEqual(sslobj.compression(),
638	n/a	client.get_extra_info('compression'))
639	n/a	self.assertEqual(sslobj.cipher(),
640	n/a	client.get_extra_info('cipher'))
641	n/a	self.assertEqual(sslobj.getpeercert(),
642	n/a	client.get_extra_info('peercert'))
643	n/a	self.assertEqual(sslobj.compression(),
644	n/a	client.get_extra_info('compression'))
645	n/a
646	n/a	def _basetest_create_ssl_connection(self, connection_fut,
647	n/a	check_sockname=True,
648	n/a	peername=None):
649	n/a	tr, pr = self.loop.run_until_complete(connection_fut)
650	n/a	self.assertIsInstance(tr, asyncio.Transport)
651	n/a	self.assertIsInstance(pr, asyncio.Protocol)
652	n/a	self.assertTrue('ssl' in tr.__class__.__name__.lower())
653	n/a	self.check_ssl_extra_info(tr, check_sockname, peername)
654	n/a	self.loop.run_until_complete(pr.done)
655	n/a	self.assertGreater(pr.nbytes, 0)
656	n/a	tr.close()
657	n/a
658	n/a	def _test_create_ssl_connection(self, httpd, create_connection,
659	n/a	check_sockname=True, peername=None):
660	n/a	conn_fut = create_connection(ssl=test_utils.dummy_ssl_context())
661	n/a	self._basetest_create_ssl_connection(conn_fut, check_sockname,
662	n/a	peername)
663	n/a
664	n/a	# ssl.Purpose was introduced in Python 3.4
665	n/a	if hasattr(ssl, 'Purpose'):
666	n/a	def _dummy_ssl_create_context(purpose=ssl.Purpose.SERVER_AUTH, *,
667	n/a	cafile=None, capath=None,
668	n/a	cadata=None):
669	n/a	"""
670	n/a	A ssl.create_default_context() replacement that doesn't enable
671	n/a	cert validation.
672	n/a	"""
673	n/a	self.assertEqual(purpose, ssl.Purpose.SERVER_AUTH)
674	n/a	return test_utils.dummy_ssl_context()
675	n/a
676	n/a	# With ssl=True, ssl.create_default_context() should be called
677	n/a	with mock.patch('ssl.create_default_context',
678	n/a	side_effect=_dummy_ssl_create_context) as m:
679	n/a	conn_fut = create_connection(ssl=True)
680	n/a	self._basetest_create_ssl_connection(conn_fut, check_sockname,
681	n/a	peername)
682	n/a	self.assertEqual(m.call_count, 1)
683	n/a
684	n/a	# With the real ssl.create_default_context(), certificate
685	n/a	# validation will fail
686	n/a	with self.assertRaises(ssl.SSLError) as cm:
687	n/a	conn_fut = create_connection(ssl=True)
688	n/a	# Ignore the "SSL handshake failed" log in debug mode
689	n/a	with test_utils.disable_logger():
690	n/a	self._basetest_create_ssl_connection(conn_fut, check_sockname,
691	n/a	peername)
692	n/a
693	n/a	self.assertEqual(cm.exception.reason, 'CERTIFICATE_VERIFY_FAILED')
694	n/a
695	n/a	@unittest.skipIf(ssl is None, 'No ssl module')
696	n/a	def test_create_ssl_connection(self):
697	n/a	with test_utils.run_test_server(use_ssl=True) as httpd:
698	n/a	create_connection = functools.partial(
699	n/a	self.loop.create_connection,
700	n/a	lambda: MyProto(loop=self.loop),
701	n/a	*httpd.address)
702	n/a	self._test_create_ssl_connection(httpd, create_connection,
703	n/a	peername=httpd.address)
704	n/a
705	n/a	def test_legacy_create_ssl_connection(self):
706	n/a	with test_utils.force_legacy_ssl_support():
707	n/a	self.test_create_ssl_connection()
708	n/a
709	n/a	@unittest.skipIf(ssl is None, 'No ssl module')
710	n/a	@unittest.skipUnless(hasattr(socket, 'AF_UNIX'), 'No UNIX Sockets')
711	n/a	def test_create_ssl_unix_connection(self):
712	n/a	# Issue #20682: On Mac OS X Tiger, getsockname() returns a
713	n/a	# zero-length address for UNIX socket.
714	n/a	check_sockname = not osx_tiger()
715	n/a
716	n/a	with test_utils.run_test_unix_server(use_ssl=True) as httpd:
717	n/a	create_connection = functools.partial(
718	n/a	self.loop.create_unix_connection,
719	n/a	lambda: MyProto(loop=self.loop), httpd.address,
720	n/a	server_hostname='127.0.0.1')
721	n/a
722	n/a	self._test_create_ssl_connection(httpd, create_connection,
723	n/a	check_sockname,
724	n/a	peername=httpd.address)
725	n/a
726	n/a	def test_legacy_create_ssl_unix_connection(self):
727	n/a	with test_utils.force_legacy_ssl_support():
728	n/a	self.test_create_ssl_unix_connection()
729	n/a
730	n/a	def test_create_connection_local_addr(self):
731	n/a	with test_utils.run_test_server() as httpd:
732	n/a	port = support.find_unused_port()
733	n/a	f = self.loop.create_connection(
734	n/a	lambda: MyProto(loop=self.loop),
735	n/a	*httpd.address, local_addr=(httpd.address[0], port))
736	n/a	tr, pr = self.loop.run_until_complete(f)
737	n/a	expected = pr.transport.get_extra_info('sockname')[1]
738	n/a	self.assertEqual(port, expected)
739	n/a	tr.close()
740	n/a
741	n/a	def test_create_connection_local_addr_in_use(self):
742	n/a	with test_utils.run_test_server() as httpd:
743	n/a	f = self.loop.create_connection(
744	n/a	lambda: MyProto(loop=self.loop),
745	n/a	*httpd.address, local_addr=httpd.address)
746	n/a	with self.assertRaises(OSError) as cm:
747	n/a	self.loop.run_until_complete(f)
748	n/a	self.assertEqual(cm.exception.errno, errno.EADDRINUSE)
749	n/a	self.assertIn(str(httpd.address), cm.exception.strerror)
750	n/a
751	n/a	def test_connect_accepted_socket(self, server_ssl=None, client_ssl=None):
752	n/a	loop = self.loop
753	n/a
754	n/a	class MyProto(MyBaseProto):
755	n/a
756	n/a	def connection_lost(self, exc):
757	n/a	super().connection_lost(exc)
758	n/a	loop.call_soon(loop.stop)
759	n/a
760	n/a	def data_received(self, data):
761	n/a	super().data_received(data)
762	n/a	self.transport.write(expected_response)
763	n/a
764	n/a	lsock = socket.socket()
765	n/a	lsock.bind(('127.0.0.1', 0))
766	n/a	lsock.listen(1)
767	n/a	addr = lsock.getsockname()
768	n/a
769	n/a	message = b'test data'
770	n/a	response = None
771	n/a	expected_response = b'roger'
772	n/a
773	n/a	def client():
774	n/a	nonlocal response
775	n/a	try:
776	n/a	csock = socket.socket()
777	n/a	if client_ssl is not None:
778	n/a	csock = client_ssl.wrap_socket(csock)
779	n/a	csock.connect(addr)
780	n/a	csock.sendall(message)
781	n/a	response = csock.recv(99)
782	n/a	csock.close()
783	n/a	except Exception as exc:
784	n/a	print(
785	n/a	"Failure in client thread in test_connect_accepted_socket",
786	n/a	exc)
787	n/a
788	n/a	thread = threading.Thread(target=client, daemon=True)
789	n/a	thread.start()
790	n/a
791	n/a	conn, _ = lsock.accept()
792	n/a	proto = MyProto(loop=loop)
793	n/a	proto.loop = loop
794	n/a	loop.run_until_complete(
795	n/a	loop.connect_accepted_socket(
796	n/a	(lambda: proto), conn, ssl=server_ssl))
797	n/a	loop.run_forever()
798	n/a	proto.transport.close()
799	n/a	lsock.close()
800	n/a
801	n/a	thread.join(1)
802	n/a	self.assertFalse(thread.is_alive())
803	n/a	self.assertEqual(proto.state, 'CLOSED')
804	n/a	self.assertEqual(proto.nbytes, len(message))
805	n/a	self.assertEqual(response, expected_response)
806	n/a
807	n/a	@unittest.skipIf(ssl is None, 'No ssl module')
808	n/a	def test_ssl_connect_accepted_socket(self):
809	n/a	if (sys.platform == 'win32' and
810	n/a	sys.version_info < (3, 5) and
811	n/a	isinstance(self.loop, proactor_events.BaseProactorEventLoop)
812	n/a	):
813	n/a	raise unittest.SkipTest(
814	n/a	'SSL not supported with proactor event loops before Python 3.5'
815	n/a	)
816	n/a
817	n/a	server_context = ssl.SSLContext(ssl.PROTOCOL_SSLv23)
818	n/a	server_context.load_cert_chain(ONLYCERT, ONLYKEY)
819	n/a	if hasattr(server_context, 'check_hostname'):
820	n/a	server_context.check_hostname = False
821	n/a	server_context.verify_mode = ssl.CERT_NONE
822	n/a
823	n/a	client_context = ssl.SSLContext(ssl.PROTOCOL_SSLv23)
824	n/a	if hasattr(server_context, 'check_hostname'):
825	n/a	client_context.check_hostname = False
826	n/a	client_context.verify_mode = ssl.CERT_NONE
827	n/a
828	n/a	self.test_connect_accepted_socket(server_context, client_context)
829	n/a
830	n/a	@mock.patch('asyncio.base_events.socket')
831	n/a	def create_server_multiple_hosts(self, family, hosts, mock_sock):
832	n/a	@asyncio.coroutine
833	n/a	def getaddrinfo(host, port, args, *kw):
834	n/a	if family == socket.AF_INET:
835	n/a	return [(family, socket.SOCK_STREAM, 6, '', (host, port))]
836	n/a	else:
837	n/a	return [(family, socket.SOCK_STREAM, 6, '', (host, port, 0, 0))]
838	n/a
839	n/a	def getaddrinfo_task(args, *kwds):
840	n/a	return asyncio.Task(getaddrinfo(args, *kwds), loop=self.loop)
841	n/a
842	n/a	unique_hosts = set(hosts)
843	n/a
844	n/a	if family == socket.AF_INET:
845	n/a	mock_sock.socket().getsockbyname.side_effect = [
846	n/a	(host, 80) for host in unique_hosts]
847	n/a	else:
848	n/a	mock_sock.socket().getsockbyname.side_effect = [
849	n/a	(host, 80, 0, 0) for host in unique_hosts]
850	n/a	self.loop.getaddrinfo = getaddrinfo_task
851	n/a	self.loop._start_serving = mock.Mock()
852	n/a	self.loop._stop_serving = mock.Mock()
853	n/a	f = self.loop.create_server(lambda: MyProto(self.loop), hosts, 80)
854	n/a	server = self.loop.run_until_complete(f)
855	n/a	self.addCleanup(server.close)
856	n/a	server_hosts = {sock.getsockbyname()[0] for sock in server.sockets}
857	n/a	self.assertEqual(server_hosts, unique_hosts)
858	n/a
859	n/a	def test_create_server_multiple_hosts_ipv4(self):
860	n/a	self.create_server_multiple_hosts(socket.AF_INET,
861	n/a	['1.2.3.4', '5.6.7.8', '1.2.3.4'])
862	n/a
863	n/a	def test_create_server_multiple_hosts_ipv6(self):
864	n/a	self.create_server_multiple_hosts(socket.AF_INET6,
865	n/a	['::1', '::2', '::1'])
866	n/a
867	n/a	def test_create_server(self):
868	n/a	proto = MyProto(self.loop)
869	n/a	f = self.loop.create_server(lambda: proto, '0.0.0.0', 0)
870	n/a	server = self.loop.run_until_complete(f)
871	n/a	self.assertEqual(len(server.sockets), 1)
872	n/a	sock = server.sockets[0]
873	n/a	host, port = sock.getsockname()
874	n/a	self.assertEqual(host, '0.0.0.0')
875	n/a	client = socket.socket()
876	n/a	client.connect(('127.0.0.1', port))
877	n/a	client.sendall(b'xxx')
878	n/a
879	n/a	self.loop.run_until_complete(proto.connected)
880	n/a	self.assertEqual('CONNECTED', proto.state)
881	n/a
882	n/a	test_utils.run_until(self.loop, lambda: proto.nbytes > 0)
883	n/a	self.assertEqual(3, proto.nbytes)
884	n/a
885	n/a	# extra info is available
886	n/a	self.assertIsNotNone(proto.transport.get_extra_info('sockname'))
887	n/a	self.assertEqual('127.0.0.1',
888	n/a	proto.transport.get_extra_info('peername')[0])
889	n/a
890	n/a	# close connection
891	n/a	proto.transport.close()
892	n/a	self.loop.run_until_complete(proto.done)
893	n/a
894	n/a	self.assertEqual('CLOSED', proto.state)
895	n/a
896	n/a	# the client socket must be closed after to avoid ECONNRESET upon
897	n/a	# recv()/send() on the serving socket
898	n/a	client.close()
899	n/a
900	n/a	# close server
901	n/a	server.close()
902	n/a
903	n/a	@unittest.skipUnless(hasattr(socket, 'SO_REUSEPORT'), 'No SO_REUSEPORT')
904	n/a	def test_create_server_reuse_port(self):
905	n/a	proto = MyProto(self.loop)
906	n/a	f = self.loop.create_server(
907	n/a	lambda: proto, '0.0.0.0', 0)
908	n/a	server = self.loop.run_until_complete(f)
909	n/a	self.assertEqual(len(server.sockets), 1)
910	n/a	sock = server.sockets[0]
911	n/a	self.assertFalse(
912	n/a	sock.getsockopt(
913	n/a	socket.SOL_SOCKET, socket.SO_REUSEPORT))
914	n/a	server.close()
915	n/a
916	n/a	test_utils.run_briefly(self.loop)
917	n/a
918	n/a	proto = MyProto(self.loop)
919	n/a	f = self.loop.create_server(
920	n/a	lambda: proto, '0.0.0.0', 0, reuse_port=True)
921	n/a	server = self.loop.run_until_complete(f)
922	n/a	self.assertEqual(len(server.sockets), 1)
923	n/a	sock = server.sockets[0]
924	n/a	self.assertTrue(
925	n/a	sock.getsockopt(
926	n/a	socket.SOL_SOCKET, socket.SO_REUSEPORT))
927	n/a	server.close()
928	n/a
929	n/a	def _make_unix_server(self, factory, **kwargs):
930	n/a	path = test_utils.gen_unix_socket_path()
931	n/a	self.addCleanup(lambda: os.path.exists(path) and os.unlink(path))
932	n/a
933	n/a	f = self.loop.create_unix_server(factory, path, **kwargs)
934	n/a	server = self.loop.run_until_complete(f)
935	n/a
936	n/a	return server, path
937	n/a
938	n/a	@unittest.skipUnless(hasattr(socket, 'AF_UNIX'), 'No UNIX Sockets')
939	n/a	def test_create_unix_server(self):
940	n/a	proto = MyProto(loop=self.loop)
941	n/a	server, path = self._make_unix_server(lambda: proto)
942	n/a	self.assertEqual(len(server.sockets), 1)
943	n/a
944	n/a	client = socket.socket(socket.AF_UNIX)
945	n/a	client.connect(path)
946	n/a	client.sendall(b'xxx')
947	n/a
948	n/a	self.loop.run_until_complete(proto.connected)
949	n/a	self.assertEqual('CONNECTED', proto.state)
950	n/a	test_utils.run_until(self.loop, lambda: proto.nbytes > 0)
951	n/a	self.assertEqual(3, proto.nbytes)
952	n/a
953	n/a	# close connection
954	n/a	proto.transport.close()
955	n/a	self.loop.run_until_complete(proto.done)
956	n/a
957	n/a	self.assertEqual('CLOSED', proto.state)
958	n/a
959	n/a	# the client socket must be closed after to avoid ECONNRESET upon
960	n/a	# recv()/send() on the serving socket
961	n/a	client.close()
962	n/a
963	n/a	# close server
964	n/a	server.close()
965	n/a
966	n/a	@unittest.skipUnless(hasattr(socket, 'AF_UNIX'), 'No UNIX Sockets')
967	n/a	def test_create_unix_server_path_socket_error(self):
968	n/a	proto = MyProto(loop=self.loop)
969	n/a	sock = socket.socket()
970	n/a	with sock:
971	n/a	f = self.loop.create_unix_server(lambda: proto, '/test', sock=sock)
972	n/a	with self.assertRaisesRegex(ValueError,
973	n/a	'path and sock can not be specified '
974	n/a	'at the same time'):
975	n/a	self.loop.run_until_complete(f)
976	n/a
977	n/a	def _create_ssl_context(self, certfile, keyfile=None):
978	n/a	sslcontext = ssl.SSLContext(ssl.PROTOCOL_SSLv23)
979	n/a	sslcontext.options \|= ssl.OP_NO_SSLv2
980	n/a	sslcontext.load_cert_chain(certfile, keyfile)
981	n/a	return sslcontext
982	n/a
983	n/a	def _make_ssl_server(self, factory, certfile, keyfile=None):
984	n/a	sslcontext = self._create_ssl_context(certfile, keyfile)
985	n/a
986	n/a	f = self.loop.create_server(factory, '127.0.0.1', 0, ssl=sslcontext)
987	n/a	server = self.loop.run_until_complete(f)
988	n/a
989	n/a	sock = server.sockets[0]
990	n/a	host, port = sock.getsockname()
991	n/a	self.assertEqual(host, '127.0.0.1')
992	n/a	return server, host, port
993	n/a
994	n/a	def _make_ssl_unix_server(self, factory, certfile, keyfile=None):
995	n/a	sslcontext = self._create_ssl_context(certfile, keyfile)
996	n/a	return self._make_unix_server(factory, ssl=sslcontext)
997	n/a
998	n/a	@unittest.skipIf(ssl is None, 'No ssl module')
999	n/a	def test_create_server_ssl(self):
1000	n/a	proto = MyProto(loop=self.loop)
1001	n/a	server, host, port = self._make_ssl_server(
1002	n/a	lambda: proto, ONLYCERT, ONLYKEY)
1003	n/a
1004	n/a	f_c = self.loop.create_connection(MyBaseProto, host, port,
1005	n/a	ssl=test_utils.dummy_ssl_context())
1006	n/a	client, pr = self.loop.run_until_complete(f_c)
1007	n/a
1008	n/a	client.write(b'xxx')
1009	n/a	self.loop.run_until_complete(proto.connected)
1010	n/a	self.assertEqual('CONNECTED', proto.state)
1011	n/a
1012	n/a	test_utils.run_until(self.loop, lambda: proto.nbytes > 0)
1013	n/a	self.assertEqual(3, proto.nbytes)
1014	n/a
1015	n/a	# extra info is available
1016	n/a	self.check_ssl_extra_info(client, peername=(host, port))
1017	n/a
1018	n/a	# close connection
1019	n/a	proto.transport.close()
1020	n/a	self.loop.run_until_complete(proto.done)
1021	n/a	self.assertEqual('CLOSED', proto.state)
1022	n/a
1023	n/a	# the client socket must be closed after to avoid ECONNRESET upon
1024	n/a	# recv()/send() on the serving socket
1025	n/a	client.close()
1026	n/a
1027	n/a	# stop serving
1028	n/a	server.close()
1029	n/a
1030	n/a	def test_legacy_create_server_ssl(self):
1031	n/a	with test_utils.force_legacy_ssl_support():
1032	n/a	self.test_create_server_ssl()
1033	n/a
1034	n/a	@unittest.skipIf(ssl is None, 'No ssl module')
1035	n/a	@unittest.skipUnless(hasattr(socket, 'AF_UNIX'), 'No UNIX Sockets')
1036	n/a	def test_create_unix_server_ssl(self):
1037	n/a	proto = MyProto(loop=self.loop)
1038	n/a	server, path = self._make_ssl_unix_server(
1039	n/a	lambda: proto, ONLYCERT, ONLYKEY)
1040	n/a
1041	n/a	f_c = self.loop.create_unix_connection(
1042	n/a	MyBaseProto, path, ssl=test_utils.dummy_ssl_context(),
1043	n/a	server_hostname='')
1044	n/a
1045	n/a	client, pr = self.loop.run_until_complete(f_c)
1046	n/a
1047	n/a	client.write(b'xxx')
1048	n/a	self.loop.run_until_complete(proto.connected)
1049	n/a	self.assertEqual('CONNECTED', proto.state)
1050	n/a	test_utils.run_until(self.loop, lambda: proto.nbytes > 0)
1051	n/a	self.assertEqual(3, proto.nbytes)
1052	n/a
1053	n/a	# close connection
1054	n/a	proto.transport.close()
1055	n/a	self.loop.run_until_complete(proto.done)
1056	n/a	self.assertEqual('CLOSED', proto.state)
1057	n/a
1058	n/a	# the client socket must be closed after to avoid ECONNRESET upon
1059	n/a	# recv()/send() on the serving socket
1060	n/a	client.close()
1061	n/a
1062	n/a	# stop serving
1063	n/a	server.close()
1064	n/a
1065	n/a	def test_legacy_create_unix_server_ssl(self):
1066	n/a	with test_utils.force_legacy_ssl_support():
1067	n/a	self.test_create_unix_server_ssl()
1068	n/a
1069	n/a	@unittest.skipIf(ssl is None, 'No ssl module')
1070	n/a	def test_create_server_ssl_verify_failed(self):
1071	n/a	proto = MyProto(loop=self.loop)
1072	n/a	server, host, port = self._make_ssl_server(
1073	n/a	lambda: proto, SIGNED_CERTFILE)
1074	n/a
1075	n/a	sslcontext_client = ssl.SSLContext(ssl.PROTOCOL_SSLv23)
1076	n/a	sslcontext_client.options \|= ssl.OP_NO_SSLv2
1077	n/a	sslcontext_client.verify_mode = ssl.CERT_REQUIRED
1078	n/a	if hasattr(sslcontext_client, 'check_hostname'):
1079	n/a	sslcontext_client.check_hostname = True
1080	n/a
1081	n/a
1082	n/a	# no CA loaded
1083	n/a	f_c = self.loop.create_connection(MyProto, host, port,
1084	n/a	ssl=sslcontext_client)
1085	n/a	with mock.patch.object(self.loop, 'call_exception_handler'):
1086	n/a	with test_utils.disable_logger():
1087	n/a	with self.assertRaisesRegex(ssl.SSLError,
1088	n/a	'(?i)certificate.verify.failed'):
1089	n/a	self.loop.run_until_complete(f_c)
1090	n/a
1091	n/a	# execute the loop to log the connection error
1092	n/a	test_utils.run_briefly(self.loop)
1093	n/a
1094	n/a	# close connection
1095	n/a	self.assertIsNone(proto.transport)
1096	n/a	server.close()
1097	n/a
1098	n/a	def test_legacy_create_server_ssl_verify_failed(self):
1099	n/a	with test_utils.force_legacy_ssl_support():
1100	n/a	self.test_create_server_ssl_verify_failed()
1101	n/a
1102	n/a	@unittest.skipIf(ssl is None, 'No ssl module')
1103	n/a	@unittest.skipUnless(hasattr(socket, 'AF_UNIX'), 'No UNIX Sockets')
1104	n/a	def test_create_unix_server_ssl_verify_failed(self):
1105	n/a	proto = MyProto(loop=self.loop)
1106	n/a	server, path = self._make_ssl_unix_server(
1107	n/a	lambda: proto, SIGNED_CERTFILE)
1108	n/a
1109	n/a	sslcontext_client = ssl.SSLContext(ssl.PROTOCOL_SSLv23)
1110	n/a	sslcontext_client.options \|= ssl.OP_NO_SSLv2
1111	n/a	sslcontext_client.verify_mode = ssl.CERT_REQUIRED
1112	n/a	if hasattr(sslcontext_client, 'check_hostname'):
1113	n/a	sslcontext_client.check_hostname = True
1114	n/a
1115	n/a	# no CA loaded
1116	n/a	f_c = self.loop.create_unix_connection(MyProto, path,
1117	n/a	ssl=sslcontext_client,
1118	n/a	server_hostname='invalid')
1119	n/a	with mock.patch.object(self.loop, 'call_exception_handler'):
1120	n/a	with test_utils.disable_logger():
1121	n/a	with self.assertRaisesRegex(ssl.SSLError,
1122	n/a	'(?i)certificate.verify.failed'):
1123	n/a	self.loop.run_until_complete(f_c)
1124	n/a
1125	n/a	# execute the loop to log the connection error
1126	n/a	test_utils.run_briefly(self.loop)
1127	n/a
1128	n/a	# close connection
1129	n/a	self.assertIsNone(proto.transport)
1130	n/a	server.close()
1131	n/a
1132	n/a
1133	n/a	def test_legacy_create_unix_server_ssl_verify_failed(self):
1134	n/a	with test_utils.force_legacy_ssl_support():
1135	n/a	self.test_create_unix_server_ssl_verify_failed()
1136	n/a
1137	n/a	@unittest.skipIf(ssl is None, 'No ssl module')
1138	n/a	def test_create_server_ssl_match_failed(self):
1139	n/a	proto = MyProto(loop=self.loop)
1140	n/a	server, host, port = self._make_ssl_server(
1141	n/a	lambda: proto, SIGNED_CERTFILE)
1142	n/a
1143	n/a	sslcontext_client = ssl.SSLContext(ssl.PROTOCOL_SSLv23)
1144	n/a	sslcontext_client.options \|= ssl.OP_NO_SSLv2
1145	n/a	sslcontext_client.verify_mode = ssl.CERT_REQUIRED
1146	n/a	sslcontext_client.load_verify_locations(
1147	n/a	cafile=SIGNING_CA)
1148	n/a	if hasattr(sslcontext_client, 'check_hostname'):
1149	n/a	sslcontext_client.check_hostname = True
1150	n/a
1151	n/a	# incorrect server_hostname
1152	n/a	f_c = self.loop.create_connection(MyProto, host, port,
1153	n/a	ssl=sslcontext_client)
1154	n/a	with mock.patch.object(self.loop, 'call_exception_handler'):
1155	n/a	with test_utils.disable_logger():
1156	n/a	with self.assertRaisesRegex(
1157	n/a	ssl.CertificateError,
1158	n/a	"hostname '127.0.0.1' doesn't match 'localhost'"):
1159	n/a	self.loop.run_until_complete(f_c)
1160	n/a
1161	n/a	# close connection
1162	n/a	proto.transport.close()
1163	n/a	server.close()
1164	n/a
1165	n/a	def test_legacy_create_server_ssl_match_failed(self):
1166	n/a	with test_utils.force_legacy_ssl_support():
1167	n/a	self.test_create_server_ssl_match_failed()
1168	n/a
1169	n/a	@unittest.skipIf(ssl is None, 'No ssl module')
1170	n/a	@unittest.skipUnless(hasattr(socket, 'AF_UNIX'), 'No UNIX Sockets')
1171	n/a	def test_create_unix_server_ssl_verified(self):
1172	n/a	proto = MyProto(loop=self.loop)
1173	n/a	server, path = self._make_ssl_unix_server(
1174	n/a	lambda: proto, SIGNED_CERTFILE)
1175	n/a
1176	n/a	sslcontext_client = ssl.SSLContext(ssl.PROTOCOL_SSLv23)
1177	n/a	sslcontext_client.options \|= ssl.OP_NO_SSLv2
1178	n/a	sslcontext_client.verify_mode = ssl.CERT_REQUIRED
1179	n/a	sslcontext_client.load_verify_locations(cafile=SIGNING_CA)
1180	n/a	if hasattr(sslcontext_client, 'check_hostname'):
1181	n/a	sslcontext_client.check_hostname = True
1182	n/a
1183	n/a	# Connection succeeds with correct CA and server hostname.
1184	n/a	f_c = self.loop.create_unix_connection(MyProto, path,
1185	n/a	ssl=sslcontext_client,
1186	n/a	server_hostname='localhost')
1187	n/a	client, pr = self.loop.run_until_complete(f_c)
1188	n/a
1189	n/a	# close connection
1190	n/a	proto.transport.close()
1191	n/a	client.close()
1192	n/a	server.close()
1193	n/a	self.loop.run_until_complete(proto.done)
1194	n/a
1195	n/a	def test_legacy_create_unix_server_ssl_verified(self):
1196	n/a	with test_utils.force_legacy_ssl_support():
1197	n/a	self.test_create_unix_server_ssl_verified()
1198	n/a
1199	n/a	@unittest.skipIf(ssl is None, 'No ssl module')
1200	n/a	def test_create_server_ssl_verified(self):
1201	n/a	proto = MyProto(loop=self.loop)
1202	n/a	server, host, port = self._make_ssl_server(
1203	n/a	lambda: proto, SIGNED_CERTFILE)
1204	n/a
1205	n/a	sslcontext_client = ssl.SSLContext(ssl.PROTOCOL_SSLv23)
1206	n/a	sslcontext_client.options \|= ssl.OP_NO_SSLv2
1207	n/a	sslcontext_client.verify_mode = ssl.CERT_REQUIRED
1208	n/a	sslcontext_client.load_verify_locations(cafile=SIGNING_CA)
1209	n/a	if hasattr(sslcontext_client, 'check_hostname'):
1210	n/a	sslcontext_client.check_hostname = True
1211	n/a
1212	n/a	# Connection succeeds with correct CA and server hostname.
1213	n/a	f_c = self.loop.create_connection(MyProto, host, port,
1214	n/a	ssl=sslcontext_client,
1215	n/a	server_hostname='localhost')
1216	n/a	client, pr = self.loop.run_until_complete(f_c)
1217	n/a
1218	n/a	# extra info is available
1219	n/a	self.check_ssl_extra_info(client,peername=(host, port),
1220	n/a	peercert=PEERCERT)
1221	n/a
1222	n/a	# close connection
1223	n/a	proto.transport.close()
1224	n/a	client.close()
1225	n/a	server.close()
1226	n/a	self.loop.run_until_complete(proto.done)
1227	n/a
1228	n/a	def test_legacy_create_server_ssl_verified(self):
1229	n/a	with test_utils.force_legacy_ssl_support():
1230	n/a	self.test_create_server_ssl_verified()
1231	n/a
1232	n/a	def test_create_server_sock(self):
1233	n/a	proto = asyncio.Future(loop=self.loop)
1234	n/a
1235	n/a	class TestMyProto(MyProto):
1236	n/a	def connection_made(self, transport):
1237	n/a	super().connection_made(transport)
1238	n/a	proto.set_result(self)
1239	n/a
1240	n/a	sock_ob = socket.socket(type=socket.SOCK_STREAM)
1241	n/a	sock_ob.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
1242	n/a	sock_ob.bind(('0.0.0.0', 0))
1243	n/a
1244	n/a	f = self.loop.create_server(TestMyProto, sock=sock_ob)
1245	n/a	server = self.loop.run_until_complete(f)
1246	n/a	sock = server.sockets[0]
1247	n/a	self.assertIs(sock, sock_ob)
1248	n/a
1249	n/a	host, port = sock.getsockname()
1250	n/a	self.assertEqual(host, '0.0.0.0')
1251	n/a	client = socket.socket()
1252	n/a	client.connect(('127.0.0.1', port))
1253	n/a	client.send(b'xxx')
1254	n/a	client.close()
1255	n/a	server.close()
1256	n/a
1257	n/a	def test_create_server_addr_in_use(self):
1258	n/a	sock_ob = socket.socket(type=socket.SOCK_STREAM)
1259	n/a	sock_ob.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
1260	n/a	sock_ob.bind(('0.0.0.0', 0))
1261	n/a
1262	n/a	f = self.loop.create_server(MyProto, sock=sock_ob)
1263	n/a	server = self.loop.run_until_complete(f)
1264	n/a	sock = server.sockets[0]
1265	n/a	host, port = sock.getsockname()
1266	n/a
1267	n/a	f = self.loop.create_server(MyProto, host=host, port=port)
1268	n/a	with self.assertRaises(OSError) as cm:
1269	n/a	self.loop.run_until_complete(f)
1270	n/a	self.assertEqual(cm.exception.errno, errno.EADDRINUSE)
1271	n/a
1272	n/a	server.close()
1273	n/a
1274	n/a	@unittest.skipUnless(support.IPV6_ENABLED, 'IPv6 not supported or enabled')
1275	n/a	def test_create_server_dual_stack(self):
1276	n/a	f_proto = asyncio.Future(loop=self.loop)
1277	n/a
1278	n/a	class TestMyProto(MyProto):
1279	n/a	def connection_made(self, transport):
1280	n/a	super().connection_made(transport)
1281	n/a	f_proto.set_result(self)
1282	n/a
1283	n/a	try_count = 0
1284	n/a	while True:
1285	n/a	try:
1286	n/a	port = support.find_unused_port()
1287	n/a	f = self.loop.create_server(TestMyProto, host=None, port=port)
1288	n/a	server = self.loop.run_until_complete(f)
1289	n/a	except OSError as ex:
1290	n/a	if ex.errno == errno.EADDRINUSE:
1291	n/a	try_count += 1
1292	n/a	self.assertGreaterEqual(5, try_count)
1293	n/a	continue
1294	n/a	else:
1295	n/a	raise
1296	n/a	else:
1297	n/a	break
1298	n/a	client = socket.socket()
1299	n/a	client.connect(('127.0.0.1', port))
1300	n/a	client.send(b'xxx')
1301	n/a	proto = self.loop.run_until_complete(f_proto)
1302	n/a	proto.transport.close()
1303	n/a	client.close()
1304	n/a
1305	n/a	f_proto = asyncio.Future(loop=self.loop)
1306	n/a	client = socket.socket(socket.AF_INET6)
1307	n/a	client.connect(('::1', port))
1308	n/a	client.send(b'xxx')
1309	n/a	proto = self.loop.run_until_complete(f_proto)
1310	n/a	proto.transport.close()
1311	n/a	client.close()
1312	n/a
1313	n/a	server.close()
1314	n/a
1315	n/a	def test_server_close(self):
1316	n/a	f = self.loop.create_server(MyProto, '0.0.0.0', 0)
1317	n/a	server = self.loop.run_until_complete(f)
1318	n/a	sock = server.sockets[0]
1319	n/a	host, port = sock.getsockname()
1320	n/a
1321	n/a	client = socket.socket()
1322	n/a	client.connect(('127.0.0.1', port))
1323	n/a	client.send(b'xxx')
1324	n/a	client.close()
1325	n/a
1326	n/a	server.close()
1327	n/a
1328	n/a	client = socket.socket()
1329	n/a	self.assertRaises(
1330	n/a	ConnectionRefusedError, client.connect, ('127.0.0.1', port))
1331	n/a	client.close()
1332	n/a
1333	n/a	def test_create_datagram_endpoint(self):
1334	n/a	class TestMyDatagramProto(MyDatagramProto):
1335	n/a	def __init__(inner_self):
1336	n/a	super().__init__(loop=self.loop)
1337	n/a
1338	n/a	def datagram_received(self, data, addr):
1339	n/a	super().datagram_received(data, addr)
1340	n/a	self.transport.sendto(b'resp:'+data, addr)
1341	n/a
1342	n/a	coro = self.loop.create_datagram_endpoint(
1343	n/a	TestMyDatagramProto, local_addr=('127.0.0.1', 0))
1344	n/a	s_transport, server = self.loop.run_until_complete(coro)
1345	n/a	host, port = s_transport.get_extra_info('sockname')
1346	n/a
1347	n/a	self.assertIsInstance(s_transport, asyncio.Transport)
1348	n/a	self.assertIsInstance(server, TestMyDatagramProto)
1349	n/a	self.assertEqual('INITIALIZED', server.state)
1350	n/a	self.assertIs(server.transport, s_transport)
1351	n/a
1352	n/a	coro = self.loop.create_datagram_endpoint(
1353	n/a	lambda: MyDatagramProto(loop=self.loop),
1354	n/a	remote_addr=(host, port))
1355	n/a	transport, client = self.loop.run_until_complete(coro)
1356	n/a
1357	n/a	self.assertIsInstance(transport, asyncio.Transport)
1358	n/a	self.assertIsInstance(client, MyDatagramProto)
1359	n/a	self.assertEqual('INITIALIZED', client.state)
1360	n/a	self.assertIs(client.transport, transport)
1361	n/a
1362	n/a	transport.sendto(b'xxx')
1363	n/a	test_utils.run_until(self.loop, lambda: server.nbytes)
1364	n/a	self.assertEqual(3, server.nbytes)
1365	n/a	test_utils.run_until(self.loop, lambda: client.nbytes)
1366	n/a
1367	n/a	# received
1368	n/a	self.assertEqual(8, client.nbytes)
1369	n/a
1370	n/a	# extra info is available
1371	n/a	self.assertIsNotNone(transport.get_extra_info('sockname'))
1372	n/a
1373	n/a	# close connection
1374	n/a	transport.close()
1375	n/a	self.loop.run_until_complete(client.done)
1376	n/a	self.assertEqual('CLOSED', client.state)
1377	n/a	server.transport.close()
1378	n/a
1379	n/a	def test_create_datagram_endpoint_sock(self):
1380	n/a	if (sys.platform == 'win32' and
1381	n/a	isinstance(self.loop, proactor_events.BaseProactorEventLoop)):
1382	n/a	raise unittest.SkipTest(
1383	n/a	'UDP is not supported with proactor event loops')
1384	n/a
1385	n/a	sock = None
1386	n/a	local_address = ('127.0.0.1', 0)
1387	n/a	infos = self.loop.run_until_complete(
1388	n/a	self.loop.getaddrinfo(
1389	n/a	*local_address, type=socket.SOCK_DGRAM))
1390	n/a	for family, type, proto, cname, address in infos:
1391	n/a	try:
1392	n/a	sock = socket.socket(family=family, type=type, proto=proto)
1393	n/a	sock.setblocking(False)
1394	n/a	sock.bind(address)
1395	n/a	except:
1396	n/a	pass
1397	n/a	else:
1398	n/a	break
1399	n/a	else:
1400	n/a	assert False, 'Can not create socket.'
1401	n/a
1402	n/a	f = self.loop.create_datagram_endpoint(
1403	n/a	lambda: MyDatagramProto(loop=self.loop), sock=sock)
1404	n/a	tr, pr = self.loop.run_until_complete(f)
1405	n/a	self.assertIsInstance(tr, asyncio.Transport)
1406	n/a	self.assertIsInstance(pr, MyDatagramProto)
1407	n/a	tr.close()
1408	n/a	self.loop.run_until_complete(pr.done)
1409	n/a
1410	n/a	def test_internal_fds(self):
1411	n/a	loop = self.create_event_loop()
1412	n/a	if not isinstance(loop, selector_events.BaseSelectorEventLoop):
1413	n/a	loop.close()
1414	n/a	self.skipTest('loop is not a BaseSelectorEventLoop')
1415	n/a
1416	n/a	self.assertEqual(1, loop._internal_fds)
1417	n/a	loop.close()
1418	n/a	self.assertEqual(0, loop._internal_fds)
1419	n/a	self.assertIsNone(loop._csock)
1420	n/a	self.assertIsNone(loop._ssock)
1421	n/a
1422	n/a	@unittest.skipUnless(sys.platform != 'win32',
1423	n/a	"Don't support pipes for Windows")
1424	n/a	def test_read_pipe(self):
1425	n/a	proto = MyReadPipeProto(loop=self.loop)
1426	n/a
1427	n/a	rpipe, wpipe = os.pipe()
1428	n/a	pipeobj = io.open(rpipe, 'rb', 1024)
1429	n/a
1430	n/a	@asyncio.coroutine
1431	n/a	def connect():
1432	n/a	t, p = yield from self.loop.connect_read_pipe(
1433	n/a	lambda: proto, pipeobj)
1434	n/a	self.assertIs(p, proto)
1435	n/a	self.assertIs(t, proto.transport)
1436	n/a	self.assertEqual(['INITIAL', 'CONNECTED'], proto.state)
1437	n/a	self.assertEqual(0, proto.nbytes)
1438	n/a
1439	n/a	self.loop.run_until_complete(connect())
1440	n/a
1441	n/a	os.write(wpipe, b'1')
1442	n/a	test_utils.run_until(self.loop, lambda: proto.nbytes >= 1)
1443	n/a	self.assertEqual(1, proto.nbytes)
1444	n/a
1445	n/a	os.write(wpipe, b'2345')
1446	n/a	test_utils.run_until(self.loop, lambda: proto.nbytes >= 5)
1447	n/a	self.assertEqual(['INITIAL', 'CONNECTED'], proto.state)
1448	n/a	self.assertEqual(5, proto.nbytes)
1449	n/a
1450	n/a	os.close(wpipe)
1451	n/a	self.loop.run_until_complete(proto.done)
1452	n/a	self.assertEqual(
1453	n/a	['INITIAL', 'CONNECTED', 'EOF', 'CLOSED'], proto.state)
1454	n/a	# extra info is available
1455	n/a	self.assertIsNotNone(proto.transport.get_extra_info('pipe'))
1456	n/a
1457	n/a	@unittest.skipUnless(sys.platform != 'win32',
1458	n/a	"Don't support pipes for Windows")
1459	n/a	def test_unclosed_pipe_transport(self):
1460	n/a	# This test reproduces the issue #314 on GitHub
1461	n/a	loop = self.create_event_loop()
1462	n/a	read_proto = MyReadPipeProto(loop=loop)
1463	n/a	write_proto = MyWritePipeProto(loop=loop)
1464	n/a
1465	n/a	rpipe, wpipe = os.pipe()
1466	n/a	rpipeobj = io.open(rpipe, 'rb', 1024)
1467	n/a	wpipeobj = io.open(wpipe, 'w', 1024)
1468	n/a
1469	n/a	@asyncio.coroutine
1470	n/a	def connect():
1471	n/a	read_transport, _ = yield from loop.connect_read_pipe(
1472	n/a	lambda: read_proto, rpipeobj)
1473	n/a	write_transport, _ = yield from loop.connect_write_pipe(
1474	n/a	lambda: write_proto, wpipeobj)
1475	n/a	return read_transport, write_transport
1476	n/a
1477	n/a	# Run and close the loop without closing the transports
1478	n/a	read_transport, write_transport = loop.run_until_complete(connect())
1479	n/a	loop.close()
1480	n/a
1481	n/a	# These 'repr' calls used to raise an AttributeError
1482	n/a	# See Issue #314 on GitHub
1483	n/a	self.assertIn('open', repr(read_transport))
1484	n/a	self.assertIn('open', repr(write_transport))
1485	n/a
1486	n/a	# Clean up (avoid ResourceWarning)
1487	n/a	rpipeobj.close()
1488	n/a	wpipeobj.close()
1489	n/a	read_transport._pipe = None
1490	n/a	write_transport._pipe = None
1491	n/a
1492	n/a	@unittest.skipUnless(sys.platform != 'win32',
1493	n/a	"Don't support pipes for Windows")
1494	n/a	# select, poll and kqueue don't support character devices (PTY) on Mac OS X
1495	n/a	# older than 10.6 (Snow Leopard)
1496	n/a	@support.requires_mac_ver(10, 6)
1497	n/a	# Issue #20495: The test hangs on FreeBSD 7.2 but pass on FreeBSD 9
1498	n/a	@support.requires_freebsd_version(8)
1499	n/a	def test_read_pty_output(self):
1500	n/a	proto = MyReadPipeProto(loop=self.loop)
1501	n/a
1502	n/a	master, slave = os.openpty()
1503	n/a	master_read_obj = io.open(master, 'rb', 0)
1504	n/a
1505	n/a	@asyncio.coroutine
1506	n/a	def connect():
1507	n/a	t, p = yield from self.loop.connect_read_pipe(lambda: proto,
1508	n/a	master_read_obj)
1509	n/a	self.assertIs(p, proto)
1510	n/a	self.assertIs(t, proto.transport)
1511	n/a	self.assertEqual(['INITIAL', 'CONNECTED'], proto.state)
1512	n/a	self.assertEqual(0, proto.nbytes)
1513	n/a
1514	n/a	self.loop.run_until_complete(connect())
1515	n/a
1516	n/a	os.write(slave, b'1')
1517	n/a	test_utils.run_until(self.loop, lambda: proto.nbytes)
1518	n/a	self.assertEqual(1, proto.nbytes)
1519	n/a
1520	n/a	os.write(slave, b'2345')
1521	n/a	test_utils.run_until(self.loop, lambda: proto.nbytes >= 5)
1522	n/a	self.assertEqual(['INITIAL', 'CONNECTED'], proto.state)
1523	n/a	self.assertEqual(5, proto.nbytes)
1524	n/a
1525	n/a	os.close(slave)
1526	n/a	self.loop.run_until_complete(proto.done)
1527	n/a	self.assertEqual(
1528	n/a	['INITIAL', 'CONNECTED', 'EOF', 'CLOSED'], proto.state)
1529	n/a	# extra info is available
1530	n/a	self.assertIsNotNone(proto.transport.get_extra_info('pipe'))
1531	n/a
1532	n/a	@unittest.skipUnless(sys.platform != 'win32',
1533	n/a	"Don't support pipes for Windows")
1534	n/a	def test_write_pipe(self):
1535	n/a	rpipe, wpipe = os.pipe()
1536	n/a	pipeobj = io.open(wpipe, 'wb', 1024)
1537	n/a
1538	n/a	proto = MyWritePipeProto(loop=self.loop)
1539	n/a	connect = self.loop.connect_write_pipe(lambda: proto, pipeobj)
1540	n/a	transport, p = self.loop.run_until_complete(connect)
1541	n/a	self.assertIs(p, proto)
1542	n/a	self.assertIs(transport, proto.transport)
1543	n/a	self.assertEqual('CONNECTED', proto.state)
1544	n/a
1545	n/a	transport.write(b'1')
1546	n/a
1547	n/a	data = bytearray()
1548	n/a	def reader(data):
1549	n/a	chunk = os.read(rpipe, 1024)
1550	n/a	data += chunk
1551	n/a	return len(data)
1552	n/a
1553	n/a	test_utils.run_until(self.loop, lambda: reader(data) >= 1)
1554	n/a	self.assertEqual(b'1', data)
1555	n/a
1556	n/a	transport.write(b'2345')
1557	n/a	test_utils.run_until(self.loop, lambda: reader(data) >= 5)
1558	n/a	self.assertEqual(b'12345', data)
1559	n/a	self.assertEqual('CONNECTED', proto.state)
1560	n/a
1561	n/a	os.close(rpipe)
1562	n/a
1563	n/a	# extra info is available
1564	n/a	self.assertIsNotNone(proto.transport.get_extra_info('pipe'))
1565	n/a
1566	n/a	# close connection
1567	n/a	proto.transport.close()
1568	n/a	self.loop.run_until_complete(proto.done)
1569	n/a	self.assertEqual('CLOSED', proto.state)
1570	n/a
1571	n/a	@unittest.skipUnless(sys.platform != 'win32',
1572	n/a	"Don't support pipes for Windows")
1573	n/a	def test_write_pipe_disconnect_on_close(self):
1574	n/a	rsock, wsock = test_utils.socketpair()
1575	n/a	rsock.setblocking(False)
1576	n/a	pipeobj = io.open(wsock.detach(), 'wb', 1024)
1577	n/a
1578	n/a	proto = MyWritePipeProto(loop=self.loop)
1579	n/a	connect = self.loop.connect_write_pipe(lambda: proto, pipeobj)
1580	n/a	transport, p = self.loop.run_until_complete(connect)
1581	n/a	self.assertIs(p, proto)
1582	n/a	self.assertIs(transport, proto.transport)
1583	n/a	self.assertEqual('CONNECTED', proto.state)
1584	n/a
1585	n/a	transport.write(b'1')
1586	n/a	data = self.loop.run_until_complete(self.loop.sock_recv(rsock, 1024))
1587	n/a	self.assertEqual(b'1', data)
1588	n/a
1589	n/a	rsock.close()
1590	n/a
1591	n/a	self.loop.run_until_complete(proto.done)
1592	n/a	self.assertEqual('CLOSED', proto.state)
1593	n/a
1594	n/a	@unittest.skipUnless(sys.platform != 'win32',
1595	n/a	"Don't support pipes for Windows")
1596	n/a	# select, poll and kqueue don't support character devices (PTY) on Mac OS X
1597	n/a	# older than 10.6 (Snow Leopard)
1598	n/a	@support.requires_mac_ver(10, 6)
1599	n/a	def test_write_pty(self):
1600	n/a	master, slave = os.openpty()
1601	n/a	slave_write_obj = io.open(slave, 'wb', 0)
1602	n/a
1603	n/a	proto = MyWritePipeProto(loop=self.loop)
1604	n/a	connect = self.loop.connect_write_pipe(lambda: proto, slave_write_obj)
1605	n/a	transport, p = self.loop.run_until_complete(connect)
1606	n/a	self.assertIs(p, proto)
1607	n/a	self.assertIs(transport, proto.transport)
1608	n/a	self.assertEqual('CONNECTED', proto.state)
1609	n/a
1610	n/a	transport.write(b'1')
1611	n/a
1612	n/a	data = bytearray()
1613	n/a	def reader(data):
1614	n/a	chunk = os.read(master, 1024)
1615	n/a	data += chunk
1616	n/a	return len(data)
1617	n/a
1618	n/a	test_utils.run_until(self.loop, lambda: reader(data) >= 1,
1619	n/a	timeout=10)
1620	n/a	self.assertEqual(b'1', data)
1621	n/a
1622	n/a	transport.write(b'2345')
1623	n/a	test_utils.run_until(self.loop, lambda: reader(data) >= 5,
1624	n/a	timeout=10)
1625	n/a	self.assertEqual(b'12345', data)
1626	n/a	self.assertEqual('CONNECTED', proto.state)
1627	n/a
1628	n/a	os.close(master)
1629	n/a
1630	n/a	# extra info is available
1631	n/a	self.assertIsNotNone(proto.transport.get_extra_info('pipe'))
1632	n/a
1633	n/a	# close connection
1634	n/a	proto.transport.close()
1635	n/a	self.loop.run_until_complete(proto.done)
1636	n/a	self.assertEqual('CLOSED', proto.state)
1637	n/a
1638	n/a	@unittest.skipUnless(sys.platform != 'win32',
1639	n/a	"Don't support pipes for Windows")
1640	n/a	# select, poll and kqueue don't support character devices (PTY) on Mac OS X
1641	n/a	# older than 10.6 (Snow Leopard)
1642	n/a	@support.requires_mac_ver(10, 6)
1643	n/a	def test_bidirectional_pty(self):
1644	n/a	master, read_slave = os.openpty()
1645	n/a	write_slave = os.dup(read_slave)
1646	n/a	tty.setraw(read_slave)
1647	n/a
1648	n/a	slave_read_obj = io.open(read_slave, 'rb', 0)
1649	n/a	read_proto = MyReadPipeProto(loop=self.loop)
1650	n/a	read_connect = self.loop.connect_read_pipe(lambda: read_proto,
1651	n/a	slave_read_obj)
1652	n/a	read_transport, p = self.loop.run_until_complete(read_connect)
1653	n/a	self.assertIs(p, read_proto)
1654	n/a	self.assertIs(read_transport, read_proto.transport)
1655	n/a	self.assertEqual(['INITIAL', 'CONNECTED'], read_proto.state)
1656	n/a	self.assertEqual(0, read_proto.nbytes)
1657	n/a
1658	n/a
1659	n/a	slave_write_obj = io.open(write_slave, 'wb', 0)
1660	n/a	write_proto = MyWritePipeProto(loop=self.loop)
1661	n/a	write_connect = self.loop.connect_write_pipe(lambda: write_proto,
1662	n/a	slave_write_obj)
1663	n/a	write_transport, p = self.loop.run_until_complete(write_connect)
1664	n/a	self.assertIs(p, write_proto)
1665	n/a	self.assertIs(write_transport, write_proto.transport)
1666	n/a	self.assertEqual('CONNECTED', write_proto.state)
1667	n/a
1668	n/a	data = bytearray()
1669	n/a	def reader(data):
1670	n/a	chunk = os.read(master, 1024)
1671	n/a	data += chunk
1672	n/a	return len(data)
1673	n/a
1674	n/a	write_transport.write(b'1')
1675	n/a	test_utils.run_until(self.loop, lambda: reader(data) >= 1, timeout=10)
1676	n/a	self.assertEqual(b'1', data)
1677	n/a	self.assertEqual(['INITIAL', 'CONNECTED'], read_proto.state)
1678	n/a	self.assertEqual('CONNECTED', write_proto.state)
1679	n/a
1680	n/a	os.write(master, b'a')
1681	n/a	test_utils.run_until(self.loop, lambda: read_proto.nbytes >= 1,
1682	n/a	timeout=10)
1683	n/a	self.assertEqual(['INITIAL', 'CONNECTED'], read_proto.state)
1684	n/a	self.assertEqual(1, read_proto.nbytes)
1685	n/a	self.assertEqual('CONNECTED', write_proto.state)
1686	n/a
1687	n/a	write_transport.write(b'2345')
1688	n/a	test_utils.run_until(self.loop, lambda: reader(data) >= 5, timeout=10)
1689	n/a	self.assertEqual(b'12345', data)
1690	n/a	self.assertEqual(['INITIAL', 'CONNECTED'], read_proto.state)
1691	n/a	self.assertEqual('CONNECTED', write_proto.state)
1692	n/a
1693	n/a	os.write(master, b'bcde')
1694	n/a	test_utils.run_until(self.loop, lambda: read_proto.nbytes >= 5,
1695	n/a	timeout=10)
1696	n/a	self.assertEqual(['INITIAL', 'CONNECTED'], read_proto.state)
1697	n/a	self.assertEqual(5, read_proto.nbytes)
1698	n/a	self.assertEqual('CONNECTED', write_proto.state)
1699	n/a
1700	n/a	os.close(master)
1701	n/a
1702	n/a	read_transport.close()
1703	n/a	self.loop.run_until_complete(read_proto.done)
1704	n/a	self.assertEqual(
1705	n/a	['INITIAL', 'CONNECTED', 'EOF', 'CLOSED'], read_proto.state)
1706	n/a
1707	n/a	write_transport.close()
1708	n/a	self.loop.run_until_complete(write_proto.done)
1709	n/a	self.assertEqual('CLOSED', write_proto.state)
1710	n/a
1711	n/a	def test_prompt_cancellation(self):
1712	n/a	r, w = test_utils.socketpair()
1713	n/a	r.setblocking(False)
1714	n/a	f = self.loop.sock_recv(r, 1)
1715	n/a	ov = getattr(f, 'ov', None)
1716	n/a	if ov is not None:
1717	n/a	self.assertTrue(ov.pending)
1718	n/a
1719	n/a	@asyncio.coroutine
1720	n/a	def main():
1721	n/a	try:
1722	n/a	self.loop.call_soon(f.cancel)
1723	n/a	yield from f
1724	n/a	except asyncio.CancelledError:
1725	n/a	res = 'cancelled'
1726	n/a	else:
1727	n/a	res = None
1728	n/a	finally:
1729	n/a	self.loop.stop()
1730	n/a	return res
1731	n/a
1732	n/a	start = time.monotonic()
1733	n/a	t = asyncio.Task(main(), loop=self.loop)
1734	n/a	self.loop.run_forever()
1735	n/a	elapsed = time.monotonic() - start
1736	n/a
1737	n/a	self.assertLess(elapsed, 0.1)
1738	n/a	self.assertEqual(t.result(), 'cancelled')
1739	n/a	self.assertRaises(asyncio.CancelledError, f.result)
1740	n/a	if ov is not None:
1741	n/a	self.assertFalse(ov.pending)
1742	n/a	self.loop._stop_serving(r)
1743	n/a
1744	n/a	r.close()
1745	n/a	w.close()
1746	n/a
1747	n/a	def test_timeout_rounding(self):
1748	n/a	def _run_once():
1749	n/a	self.loop._run_once_counter += 1
1750	n/a	orig_run_once()
1751	n/a
1752	n/a	orig_run_once = self.loop._run_once
1753	n/a	self.loop._run_once_counter = 0
1754	n/a	self.loop._run_once = _run_once
1755	n/a
1756	n/a	@asyncio.coroutine
1757	n/a	def wait():
1758	n/a	loop = self.loop
1759	n/a	yield from asyncio.sleep(1e-2, loop=loop)
1760	n/a	yield from asyncio.sleep(1e-4, loop=loop)
1761	n/a	yield from asyncio.sleep(1e-6, loop=loop)
1762	n/a	yield from asyncio.sleep(1e-8, loop=loop)
1763	n/a	yield from asyncio.sleep(1e-10, loop=loop)
1764	n/a
1765	n/a	self.loop.run_until_complete(wait())
1766	n/a	# The ideal number of call is 12, but on some platforms, the selector
1767	n/a	# may sleep at little bit less than timeout depending on the resolution
1768	n/a	# of the clock used by the kernel. Tolerate a few useless calls on
1769	n/a	# these platforms.
1770	n/a	self.assertLessEqual(self.loop._run_once_counter, 20,
1771	n/a	{'clock_resolution': self.loop._clock_resolution,
1772	n/a	'selector': self.loop._selector.__class__.__name__})
1773	n/a
1774	n/a	def test_remove_fds_after_closing(self):
1775	n/a	loop = self.create_event_loop()
1776	n/a	callback = lambda: None
1777	n/a	r, w = test_utils.socketpair()
1778	n/a	self.addCleanup(r.close)
1779	n/a	self.addCleanup(w.close)
1780	n/a	loop.add_reader(r, callback)
1781	n/a	loop.add_writer(w, callback)
1782	n/a	loop.close()
1783	n/a	self.assertFalse(loop.remove_reader(r))
1784	n/a	self.assertFalse(loop.remove_writer(w))
1785	n/a
1786	n/a	def test_add_fds_after_closing(self):
1787	n/a	loop = self.create_event_loop()
1788	n/a	callback = lambda: None
1789	n/a	r, w = test_utils.socketpair()
1790	n/a	self.addCleanup(r.close)
1791	n/a	self.addCleanup(w.close)
1792	n/a	loop.close()
1793	n/a	with self.assertRaises(RuntimeError):
1794	n/a	loop.add_reader(r, callback)
1795	n/a	with self.assertRaises(RuntimeError):
1796	n/a	loop.add_writer(w, callback)
1797	n/a
1798	n/a	def test_close_running_event_loop(self):
1799	n/a	@asyncio.coroutine
1800	n/a	def close_loop(loop):
1801	n/a	self.loop.close()
1802	n/a
1803	n/a	coro = close_loop(self.loop)
1804	n/a	with self.assertRaises(RuntimeError):
1805	n/a	self.loop.run_until_complete(coro)
1806	n/a
1807	n/a	def test_close(self):
1808	n/a	self.loop.close()
1809	n/a
1810	n/a	@asyncio.coroutine
1811	n/a	def test():
1812	n/a	pass
1813	n/a
1814	n/a	func = lambda: False
1815	n/a	coro = test()
1816	n/a	self.addCleanup(coro.close)
1817	n/a
1818	n/a	# operation blocked when the loop is closed
1819	n/a	with self.assertRaises(RuntimeError):
1820	n/a	self.loop.run_forever()
1821	n/a	with self.assertRaises(RuntimeError):
1822	n/a	fut = asyncio.Future(loop=self.loop)
1823	n/a	self.loop.run_until_complete(fut)
1824	n/a	with self.assertRaises(RuntimeError):
1825	n/a	self.loop.call_soon(func)
1826	n/a	with self.assertRaises(RuntimeError):
1827	n/a	self.loop.call_soon_threadsafe(func)
1828	n/a	with self.assertRaises(RuntimeError):
1829	n/a	self.loop.call_later(1.0, func)
1830	n/a	with self.assertRaises(RuntimeError):
1831	n/a	self.loop.call_at(self.loop.time() + .0, func)
1832	n/a	with self.assertRaises(RuntimeError):
1833	n/a	self.loop.run_in_executor(None, func)
1834	n/a	with self.assertRaises(RuntimeError):
1835	n/a	self.loop.create_task(coro)
1836	n/a	with self.assertRaises(RuntimeError):
1837	n/a	self.loop.add_signal_handler(signal.SIGTERM, func)
1838	n/a
1839	n/a
1840	n/a	class SubprocessTestsMixin:
1841	n/a
1842	n/a	def check_terminated(self, returncode):
1843	n/a	if sys.platform == 'win32':
1844	n/a	self.assertIsInstance(returncode, int)
1845	n/a	# expect 1 but sometimes get 0
1846	n/a	else:
1847	n/a	self.assertEqual(-signal.SIGTERM, returncode)
1848	n/a
1849	n/a	def check_killed(self, returncode):
1850	n/a	if sys.platform == 'win32':
1851	n/a	self.assertIsInstance(returncode, int)
1852	n/a	# expect 1 but sometimes get 0
1853	n/a	else:
1854	n/a	self.assertEqual(-signal.SIGKILL, returncode)
1855	n/a
1856	n/a	def test_subprocess_exec(self):
1857	n/a	prog = os.path.join(os.path.dirname(__file__), 'echo.py')
1858	n/a
1859	n/a	connect = self.loop.subprocess_exec(
1860	n/a	functools.partial(MySubprocessProtocol, self.loop),
1861	n/a	sys.executable, prog)
1862	n/a	transp, proto = self.loop.run_until_complete(connect)
1863	n/a	self.assertIsInstance(proto, MySubprocessProtocol)
1864	n/a	self.loop.run_until_complete(proto.connected)
1865	n/a	self.assertEqual('CONNECTED', proto.state)
1866	n/a
1867	n/a	stdin = transp.get_pipe_transport(0)
1868	n/a	stdin.write(b'Python The Winner')
1869	n/a	self.loop.run_until_complete(proto.got_data[1].wait())
1870	n/a	with test_utils.disable_logger():
1871	n/a	transp.close()
1872	n/a	self.loop.run_until_complete(proto.completed)
1873	n/a	self.check_killed(proto.returncode)
1874	n/a	self.assertEqual(b'Python The Winner', proto.data[1])
1875	n/a
1876	n/a	def test_subprocess_interactive(self):
1877	n/a	prog = os.path.join(os.path.dirname(__file__), 'echo.py')
1878	n/a
1879	n/a	connect = self.loop.subprocess_exec(
1880	n/a	functools.partial(MySubprocessProtocol, self.loop),
1881	n/a	sys.executable, prog)
1882	n/a	transp, proto = self.loop.run_until_complete(connect)
1883	n/a	self.assertIsInstance(proto, MySubprocessProtocol)
1884	n/a	self.loop.run_until_complete(proto.connected)
1885	n/a	self.assertEqual('CONNECTED', proto.state)
1886	n/a
1887	n/a	stdin = transp.get_pipe_transport(0)
1888	n/a	stdin.write(b'Python ')
1889	n/a	self.loop.run_until_complete(proto.got_data[1].wait())
1890	n/a	proto.got_data[1].clear()
1891	n/a	self.assertEqual(b'Python ', proto.data[1])
1892	n/a
1893	n/a	stdin.write(b'The Winner')
1894	n/a	self.loop.run_until_complete(proto.got_data[1].wait())
1895	n/a	self.assertEqual(b'Python The Winner', proto.data[1])
1896	n/a
1897	n/a	with test_utils.disable_logger():
1898	n/a	transp.close()
1899	n/a	self.loop.run_until_complete(proto.completed)
1900	n/a	self.check_killed(proto.returncode)
1901	n/a
1902	n/a	def test_subprocess_shell(self):
1903	n/a	connect = self.loop.subprocess_shell(
1904	n/a	functools.partial(MySubprocessProtocol, self.loop),
1905	n/a	'echo Python')
1906	n/a	transp, proto = self.loop.run_until_complete(connect)
1907	n/a	self.assertIsInstance(proto, MySubprocessProtocol)
1908	n/a	self.loop.run_until_complete(proto.connected)
1909	n/a
1910	n/a	transp.get_pipe_transport(0).close()
1911	n/a	self.loop.run_until_complete(proto.completed)
1912	n/a	self.assertEqual(0, proto.returncode)
1913	n/a	self.assertTrue(all(f.done() for f in proto.disconnects.values()))
1914	n/a	self.assertEqual(proto.data[1].rstrip(b'\r\n'), b'Python')
1915	n/a	self.assertEqual(proto.data[2], b'')
1916	n/a	transp.close()
1917	n/a
1918	n/a	def test_subprocess_exitcode(self):
1919	n/a	connect = self.loop.subprocess_shell(
1920	n/a	functools.partial(MySubprocessProtocol, self.loop),
1921	n/a	'exit 7', stdin=None, stdout=None, stderr=None)
1922	n/a	transp, proto = self.loop.run_until_complete(connect)
1923	n/a	self.assertIsInstance(proto, MySubprocessProtocol)
1924	n/a	self.loop.run_until_complete(proto.completed)
1925	n/a	self.assertEqual(7, proto.returncode)
1926	n/a	transp.close()
1927	n/a
1928	n/a	def test_subprocess_close_after_finish(self):
1929	n/a	connect = self.loop.subprocess_shell(
1930	n/a	functools.partial(MySubprocessProtocol, self.loop),
1931	n/a	'exit 7', stdin=None, stdout=None, stderr=None)
1932	n/a	transp, proto = self.loop.run_until_complete(connect)
1933	n/a	self.assertIsInstance(proto, MySubprocessProtocol)
1934	n/a	self.assertIsNone(transp.get_pipe_transport(0))
1935	n/a	self.assertIsNone(transp.get_pipe_transport(1))
1936	n/a	self.assertIsNone(transp.get_pipe_transport(2))
1937	n/a	self.loop.run_until_complete(proto.completed)
1938	n/a	self.assertEqual(7, proto.returncode)
1939	n/a	self.assertIsNone(transp.close())
1940	n/a
1941	n/a	def test_subprocess_kill(self):
1942	n/a	prog = os.path.join(os.path.dirname(__file__), 'echo.py')
1943	n/a
1944	n/a	connect = self.loop.subprocess_exec(
1945	n/a	functools.partial(MySubprocessProtocol, self.loop),
1946	n/a	sys.executable, prog)
1947	n/a	transp, proto = self.loop.run_until_complete(connect)
1948	n/a	self.assertIsInstance(proto, MySubprocessProtocol)
1949	n/a	self.loop.run_until_complete(proto.connected)
1950	n/a
1951	n/a	transp.kill()
1952	n/a	self.loop.run_until_complete(proto.completed)
1953	n/a	self.check_killed(proto.returncode)
1954	n/a	transp.close()
1955	n/a
1956	n/a	def test_subprocess_terminate(self):
1957	n/a	prog = os.path.join(os.path.dirname(__file__), 'echo.py')
1958	n/a
1959	n/a	connect = self.loop.subprocess_exec(
1960	n/a	functools.partial(MySubprocessProtocol, self.loop),
1961	n/a	sys.executable, prog)
1962	n/a	transp, proto = self.loop.run_until_complete(connect)
1963	n/a	self.assertIsInstance(proto, MySubprocessProtocol)
1964	n/a	self.loop.run_until_complete(proto.connected)
1965	n/a
1966	n/a	transp.terminate()
1967	n/a	self.loop.run_until_complete(proto.completed)
1968	n/a	self.check_terminated(proto.returncode)
1969	n/a	transp.close()
1970	n/a
1971	n/a	@unittest.skipIf(sys.platform == 'win32', "Don't have SIGHUP")
1972	n/a	def test_subprocess_send_signal(self):
1973	n/a	prog = os.path.join(os.path.dirname(__file__), 'echo.py')
1974	n/a
1975	n/a	connect = self.loop.subprocess_exec(
1976	n/a	functools.partial(MySubprocessProtocol, self.loop),
1977	n/a	sys.executable, prog)
1978	n/a	transp, proto = self.loop.run_until_complete(connect)
1979	n/a	self.assertIsInstance(proto, MySubprocessProtocol)
1980	n/a	self.loop.run_until_complete(proto.connected)
1981	n/a
1982	n/a	transp.send_signal(signal.SIGHUP)
1983	n/a	self.loop.run_until_complete(proto.completed)
1984	n/a	self.assertEqual(-signal.SIGHUP, proto.returncode)
1985	n/a	transp.close()
1986	n/a
1987	n/a	def test_subprocess_stderr(self):
1988	n/a	prog = os.path.join(os.path.dirname(__file__), 'echo2.py')
1989	n/a
1990	n/a	connect = self.loop.subprocess_exec(
1991	n/a	functools.partial(MySubprocessProtocol, self.loop),
1992	n/a	sys.executable, prog)
1993	n/a	transp, proto = self.loop.run_until_complete(connect)
1994	n/a	self.assertIsInstance(proto, MySubprocessProtocol)
1995	n/a	self.loop.run_until_complete(proto.connected)
1996	n/a
1997	n/a	stdin = transp.get_pipe_transport(0)
1998	n/a	stdin.write(b'test')
1999	n/a
2000	n/a	self.loop.run_until_complete(proto.completed)
2001	n/a
2002	n/a	transp.close()
2003	n/a	self.assertEqual(b'OUT:test', proto.data[1])
2004	n/a	self.assertTrue(proto.data[2].startswith(b'ERR:test'), proto.data[2])
2005	n/a	self.assertEqual(0, proto.returncode)
2006	n/a
2007	n/a	def test_subprocess_stderr_redirect_to_stdout(self):
2008	n/a	prog = os.path.join(os.path.dirname(__file__), 'echo2.py')
2009	n/a
2010	n/a	connect = self.loop.subprocess_exec(
2011	n/a	functools.partial(MySubprocessProtocol, self.loop),
2012	n/a	sys.executable, prog, stderr=subprocess.STDOUT)
2013	n/a	transp, proto = self.loop.run_until_complete(connect)
2014	n/a	self.assertIsInstance(proto, MySubprocessProtocol)
2015	n/a	self.loop.run_until_complete(proto.connected)
2016	n/a
2017	n/a	stdin = transp.get_pipe_transport(0)
2018	n/a	self.assertIsNotNone(transp.get_pipe_transport(1))
2019	n/a	self.assertIsNone(transp.get_pipe_transport(2))
2020	n/a
2021	n/a	stdin.write(b'test')
2022	n/a	self.loop.run_until_complete(proto.completed)
2023	n/a	self.assertTrue(proto.data[1].startswith(b'OUT:testERR:test'),
2024	n/a	proto.data[1])
2025	n/a	self.assertEqual(b'', proto.data[2])
2026	n/a
2027	n/a	transp.close()
2028	n/a	self.assertEqual(0, proto.returncode)
2029	n/a
2030	n/a	def test_subprocess_close_client_stream(self):
2031	n/a	prog = os.path.join(os.path.dirname(__file__), 'echo3.py')
2032	n/a
2033	n/a	connect = self.loop.subprocess_exec(
2034	n/a	functools.partial(MySubprocessProtocol, self.loop),
2035	n/a	sys.executable, prog)
2036	n/a	transp, proto = self.loop.run_until_complete(connect)
2037	n/a	self.assertIsInstance(proto, MySubprocessProtocol)
2038	n/a	self.loop.run_until_complete(proto.connected)
2039	n/a
2040	n/a	stdin = transp.get_pipe_transport(0)
2041	n/a	stdout = transp.get_pipe_transport(1)
2042	n/a	stdin.write(b'test')
2043	n/a	self.loop.run_until_complete(proto.got_data[1].wait())
2044	n/a	self.assertEqual(b'OUT:test', proto.data[1])
2045	n/a
2046	n/a	stdout.close()
2047	n/a	self.loop.run_until_complete(proto.disconnects[1])
2048	n/a	stdin.write(b'xxx')
2049	n/a	self.loop.run_until_complete(proto.got_data[2].wait())
2050	n/a	if sys.platform != 'win32':
2051	n/a	self.assertEqual(b'ERR:BrokenPipeError', proto.data[2])
2052	n/a	else:
2053	n/a	# After closing the read-end of a pipe, writing to the
2054	n/a	# write-end using os.write() fails with errno==EINVAL and
2055	n/a	# GetLastError()==ERROR_INVALID_NAME on Windows!?! (Using
2056	n/a	# WriteFile() we get ERROR_BROKEN_PIPE as expected.)
2057	n/a	self.assertEqual(b'ERR:OSError', proto.data[2])
2058	n/a	with test_utils.disable_logger():
2059	n/a	transp.close()
2060	n/a	self.loop.run_until_complete(proto.completed)
2061	n/a	self.check_killed(proto.returncode)
2062	n/a
2063	n/a	def test_subprocess_wait_no_same_group(self):
2064	n/a	# start the new process in a new session
2065	n/a	connect = self.loop.subprocess_shell(
2066	n/a	functools.partial(MySubprocessProtocol, self.loop),
2067	n/a	'exit 7', stdin=None, stdout=None, stderr=None,
2068	n/a	start_new_session=True)
2069	n/a	_, proto = yield self.loop.run_until_complete(connect)
2070	n/a	self.assertIsInstance(proto, MySubprocessProtocol)
2071	n/a	self.loop.run_until_complete(proto.completed)
2072	n/a	self.assertEqual(7, proto.returncode)
2073	n/a
2074	n/a	def test_subprocess_exec_invalid_args(self):
2075	n/a	@asyncio.coroutine
2076	n/a	def connect(**kwds):
2077	n/a	yield from self.loop.subprocess_exec(
2078	n/a	asyncio.SubprocessProtocol,
2079	n/a	'pwd', **kwds)
2080	n/a
2081	n/a	with self.assertRaises(ValueError):
2082	n/a	self.loop.run_until_complete(connect(universal_newlines=True))
2083	n/a	with self.assertRaises(ValueError):
2084	n/a	self.loop.run_until_complete(connect(bufsize=4096))
2085	n/a	with self.assertRaises(ValueError):
2086	n/a	self.loop.run_until_complete(connect(shell=True))
2087	n/a
2088	n/a	def test_subprocess_shell_invalid_args(self):
2089	n/a	@asyncio.coroutine
2090	n/a	def connect(cmd=None, **kwds):
2091	n/a	if not cmd:
2092	n/a	cmd = 'pwd'
2093	n/a	yield from self.loop.subprocess_shell(
2094	n/a	asyncio.SubprocessProtocol,
2095	n/a	cmd, **kwds)
2096	n/a
2097	n/a	with self.assertRaises(ValueError):
2098	n/a	self.loop.run_until_complete(connect(['ls', '-l']))
2099	n/a	with self.assertRaises(ValueError):
2100	n/a	self.loop.run_until_complete(connect(universal_newlines=True))
2101	n/a	with self.assertRaises(ValueError):
2102	n/a	self.loop.run_until_complete(connect(bufsize=4096))
2103	n/a	with self.assertRaises(ValueError):
2104	n/a	self.loop.run_until_complete(connect(shell=False))
2105	n/a
2106	n/a
2107	n/a	if sys.platform == 'win32':
2108	n/a
2109	n/a	class SelectEventLoopTests(EventLoopTestsMixin, test_utils.TestCase):
2110	n/a
2111	n/a	def create_event_loop(self):
2112	n/a	return asyncio.SelectorEventLoop()
2113	n/a
2114	n/a	class ProactorEventLoopTests(EventLoopTestsMixin,
2115	n/a	SubprocessTestsMixin,
2116	n/a	test_utils.TestCase):
2117	n/a
2118	n/a	def create_event_loop(self):
2119	n/a	return asyncio.ProactorEventLoop()
2120	n/a
2121	n/a	if not sslproto._is_sslproto_available():
2122	n/a	def test_create_ssl_connection(self):
2123	n/a	raise unittest.SkipTest("need python 3.5 (ssl.MemoryBIO)")
2124	n/a
2125	n/a	def test_create_server_ssl(self):
2126	n/a	raise unittest.SkipTest("need python 3.5 (ssl.MemoryBIO)")
2127	n/a
2128	n/a	def test_create_server_ssl_verify_failed(self):
2129	n/a	raise unittest.SkipTest("need python 3.5 (ssl.MemoryBIO)")
2130	n/a
2131	n/a	def test_create_server_ssl_match_failed(self):
2132	n/a	raise unittest.SkipTest("need python 3.5 (ssl.MemoryBIO)")
2133	n/a
2134	n/a	def test_create_server_ssl_verified(self):
2135	n/a	raise unittest.SkipTest("need python 3.5 (ssl.MemoryBIO)")
2136	n/a
2137	n/a	def test_legacy_create_ssl_connection(self):
2138	n/a	raise unittest.SkipTest("IocpEventLoop incompatible with legacy SSL")
2139	n/a
2140	n/a	def test_legacy_create_server_ssl(self):
2141	n/a	raise unittest.SkipTest("IocpEventLoop incompatible with legacy SSL")
2142	n/a
2143	n/a	def test_legacy_create_server_ssl_verify_failed(self):
2144	n/a	raise unittest.SkipTest("IocpEventLoop incompatible with legacy SSL")
2145	n/a
2146	n/a	def test_legacy_create_server_ssl_match_failed(self):
2147	n/a	raise unittest.SkipTest("IocpEventLoop incompatible with legacy SSL")
2148	n/a
2149	n/a	def test_legacy_create_server_ssl_verified(self):
2150	n/a	raise unittest.SkipTest("IocpEventLoop incompatible with legacy SSL")
2151	n/a
2152	n/a	def test_reader_callback(self):
2153	n/a	raise unittest.SkipTest("IocpEventLoop does not have add_reader()")
2154	n/a
2155	n/a	def test_reader_callback_cancel(self):
2156	n/a	raise unittest.SkipTest("IocpEventLoop does not have add_reader()")
2157	n/a
2158	n/a	def test_writer_callback(self):
2159	n/a	raise unittest.SkipTest("IocpEventLoop does not have add_writer()")
2160	n/a
2161	n/a	def test_writer_callback_cancel(self):
2162	n/a	raise unittest.SkipTest("IocpEventLoop does not have add_writer()")
2163	n/a
2164	n/a	def test_create_datagram_endpoint(self):
2165	n/a	raise unittest.SkipTest(
2166	n/a	"IocpEventLoop does not have create_datagram_endpoint()")
2167	n/a
2168	n/a	def test_remove_fds_after_closing(self):
2169	n/a	raise unittest.SkipTest("IocpEventLoop does not have add_reader()")
2170	n/a	else:
2171	n/a	from asyncio import selectors
2172	n/a
2173	n/a	class UnixEventLoopTestsMixin(EventLoopTestsMixin):
2174	n/a	def setUp(self):
2175	n/a	super().setUp()
2176	n/a	watcher = asyncio.SafeChildWatcher()
2177	n/a	watcher.attach_loop(self.loop)
2178	n/a	asyncio.set_child_watcher(watcher)
2179	n/a
2180	n/a	def tearDown(self):
2181	n/a	asyncio.set_child_watcher(None)
2182	n/a	super().tearDown()
2183	n/a
2184	n/a	if hasattr(selectors, 'KqueueSelector'):
2185	n/a	class KqueueEventLoopTests(UnixEventLoopTestsMixin,
2186	n/a	SubprocessTestsMixin,
2187	n/a	test_utils.TestCase):
2188	n/a
2189	n/a	def create_event_loop(self):
2190	n/a	return asyncio.SelectorEventLoop(
2191	n/a	selectors.KqueueSelector())
2192	n/a
2193	n/a	# kqueue doesn't support character devices (PTY) on Mac OS X older
2194	n/a	# than 10.9 (Maverick)
2195	n/a	@support.requires_mac_ver(10, 9)
2196	n/a	# Issue #20667: KqueueEventLoopTests.test_read_pty_output()
2197	n/a	# hangs on OpenBSD 5.5
2198	n/a	@unittest.skipIf(sys.platform.startswith('openbsd'),
2199	n/a	'test hangs on OpenBSD')
2200	n/a	def test_read_pty_output(self):
2201	n/a	super().test_read_pty_output()
2202	n/a
2203	n/a	# kqueue doesn't support character devices (PTY) on Mac OS X older
2204	n/a	# than 10.9 (Maverick)
2205	n/a	@support.requires_mac_ver(10, 9)
2206	n/a	def test_write_pty(self):
2207	n/a	super().test_write_pty()
2208	n/a
2209	n/a	if hasattr(selectors, 'EpollSelector'):
2210	n/a	class EPollEventLoopTests(UnixEventLoopTestsMixin,
2211	n/a	SubprocessTestsMixin,
2212	n/a	test_utils.TestCase):
2213	n/a
2214	n/a	def create_event_loop(self):
2215	n/a	return asyncio.SelectorEventLoop(selectors.EpollSelector())
2216	n/a
2217	n/a	if hasattr(selectors, 'PollSelector'):
2218	n/a	class PollEventLoopTests(UnixEventLoopTestsMixin,
2219	n/a	SubprocessTestsMixin,
2220	n/a	test_utils.TestCase):
2221	n/a
2222	n/a	def create_event_loop(self):
2223	n/a	return asyncio.SelectorEventLoop(selectors.PollSelector())
2224	n/a
2225	n/a	# Should always exist.
2226	n/a	class SelectEventLoopTests(UnixEventLoopTestsMixin,
2227	n/a	SubprocessTestsMixin,
2228	n/a	test_utils.TestCase):
2229	n/a
2230	n/a	def create_event_loop(self):
2231	n/a	return asyncio.SelectorEventLoop(selectors.SelectSelector())
2232	n/a
2233	n/a
2234	n/a	def noop(args, *kwargs):
2235	n/a	pass
2236	n/a
2237	n/a
2238	n/a	class HandleTests(test_utils.TestCase):
2239	n/a
2240	n/a	def setUp(self):
2241	n/a	super().setUp()
2242	n/a	self.loop = mock.Mock()
2243	n/a	self.loop.get_debug.return_value = True
2244	n/a
2245	n/a	def test_handle(self):
2246	n/a	def callback(*args):
2247	n/a	return args
2248	n/a
2249	n/a	args = ()
2250	n/a	h = asyncio.Handle(callback, args, self.loop)
2251	n/a	self.assertIs(h._callback, callback)
2252	n/a	self.assertIs(h._args, args)
2253	n/a	self.assertFalse(h._cancelled)
2254	n/a
2255	n/a	h.cancel()
2256	n/a	self.assertTrue(h._cancelled)
2257	n/a
2258	n/a	def test_callback_with_exception(self):
2259	n/a	def callback():
2260	n/a	raise ValueError()
2261	n/a
2262	n/a	self.loop = mock.Mock()
2263	n/a	self.loop.call_exception_handler = mock.Mock()
2264	n/a
2265	n/a	h = asyncio.Handle(callback, (), self.loop)
2266	n/a	h._run()
2267	n/a
2268	n/a	self.loop.call_exception_handler.assert_called_with({
2269	n/a	'message': test_utils.MockPattern('Exception in callback.*'),
2270	n/a	'exception': mock.ANY,
2271	n/a	'handle': h,
2272	n/a	'source_traceback': h._source_traceback,
2273	n/a	})
2274	n/a
2275	n/a	def test_handle_weakref(self):
2276	n/a	wd = weakref.WeakValueDictionary()
2277	n/a	h = asyncio.Handle(lambda: None, (), self.loop)
2278	n/a	wd['h'] = h # Would fail without __weakref__ slot.
2279	n/a
2280	n/a	def test_handle_repr(self):
2281	n/a	self.loop.get_debug.return_value = False
2282	n/a
2283	n/a	# simple function
2284	n/a	h = asyncio.Handle(noop, (1, 2), self.loop)
2285	n/a	filename, lineno = test_utils.get_function_source(noop)
2286	n/a	self.assertEqual(repr(h),
2287	n/a	'<Handle noop(1, 2) at %s:%s>'
2288	n/a	% (filename, lineno))
2289	n/a
2290	n/a	# cancelled handle
2291	n/a	h.cancel()
2292	n/a	self.assertEqual(repr(h),
2293	n/a	'<Handle cancelled>')
2294	n/a
2295	n/a	# decorated function
2296	n/a	cb = asyncio.coroutine(noop)
2297	n/a	h = asyncio.Handle(cb, (), self.loop)
2298	n/a	self.assertEqual(repr(h),
2299	n/a	'<Handle noop() at %s:%s>'
2300	n/a	% (filename, lineno))
2301	n/a
2302	n/a	# partial function
2303	n/a	cb = functools.partial(noop, 1, 2)
2304	n/a	h = asyncio.Handle(cb, (3,), self.loop)
2305	n/a	regex = (r'^<Handle noop$1, 2$$3$ at %s:%s>$'
2306	n/a	% (re.escape(filename), lineno))
2307	n/a	self.assertRegex(repr(h), regex)
2308	n/a
2309	n/a	# partial function with keyword args
2310	n/a	cb = functools.partial(noop, x=1)
2311	n/a	h = asyncio.Handle(cb, (2, 3), self.loop)
2312	n/a	regex = (r'^<Handle noop$x=1$$2, 3$ at %s:%s>$'
2313	n/a	% (re.escape(filename), lineno))
2314	n/a	self.assertRegex(repr(h), regex)
2315	n/a
2316	n/a	# partial method
2317	n/a	if sys.version_info >= (3, 4):
2318	n/a	method = HandleTests.test_handle_repr
2319	n/a	cb = functools.partialmethod(method)
2320	n/a	filename, lineno = test_utils.get_function_source(method)
2321	n/a	h = asyncio.Handle(cb, (), self.loop)
2322	n/a
2323	n/a	cb_regex = r'<function HandleTests.test_handle_repr .*>'
2324	n/a	cb_regex = (r'functools.partialmethod$%s, , $' % cb_regex)
2325	n/a	regex = (r'^<Handle %s at %s:%s>$'
2326	n/a	% (cb_regex, re.escape(filename), lineno))
2327	n/a	self.assertRegex(repr(h), regex)
2328	n/a
2329	n/a	def test_handle_repr_debug(self):
2330	n/a	self.loop.get_debug.return_value = True
2331	n/a
2332	n/a	# simple function
2333	n/a	create_filename = __file__
2334	n/a	create_lineno = sys._getframe().f_lineno + 1
2335	n/a	h = asyncio.Handle(noop, (1, 2), self.loop)
2336	n/a	filename, lineno = test_utils.get_function_source(noop)
2337	n/a	self.assertEqual(repr(h),
2338	n/a	'<Handle noop(1, 2) at %s:%s created at %s:%s>'
2339	n/a	% (filename, lineno, create_filename, create_lineno))
2340	n/a
2341	n/a	# cancelled handle
2342	n/a	h.cancel()
2343	n/a	self.assertEqual(
2344	n/a	repr(h),
2345	n/a	'<Handle cancelled noop(1, 2) at %s:%s created at %s:%s>'
2346	n/a	% (filename, lineno, create_filename, create_lineno))
2347	n/a
2348	n/a	# double cancellation won't overwrite _repr
2349	n/a	h.cancel()
2350	n/a	self.assertEqual(
2351	n/a	repr(h),
2352	n/a	'<Handle cancelled noop(1, 2) at %s:%s created at %s:%s>'
2353	n/a	% (filename, lineno, create_filename, create_lineno))
2354	n/a
2355	n/a	def test_handle_source_traceback(self):
2356	n/a	loop = asyncio.get_event_loop_policy().new_event_loop()
2357	n/a	loop.set_debug(True)
2358	n/a	self.set_event_loop(loop)
2359	n/a
2360	n/a	def check_source_traceback(h):
2361	n/a	lineno = sys._getframe(1).f_lineno - 1
2362	n/a	self.assertIsInstance(h._source_traceback, list)
2363	n/a	self.assertEqual(h._source_traceback[-1][:3],
2364	n/a	(__file__,
2365	n/a	lineno,
2366	n/a	'test_handle_source_traceback'))
2367	n/a
2368	n/a	# call_soon
2369	n/a	h = loop.call_soon(noop)
2370	n/a	check_source_traceback(h)
2371	n/a
2372	n/a	# call_soon_threadsafe
2373	n/a	h = loop.call_soon_threadsafe(noop)
2374	n/a	check_source_traceback(h)
2375	n/a
2376	n/a	# call_later
2377	n/a	h = loop.call_later(0, noop)
2378	n/a	check_source_traceback(h)
2379	n/a
2380	n/a	# call_at
2381	n/a	h = loop.call_later(0, noop)
2382	n/a	check_source_traceback(h)
2383	n/a
2384	n/a	@unittest.skipUnless(hasattr(collections.abc, 'Coroutine'),
2385	n/a	'No collections.abc.Coroutine')
2386	n/a	def test_coroutine_like_object_debug_formatting(self):
2387	n/a	# Test that asyncio can format coroutines that are instances of
2388	n/a	# collections.abc.Coroutine, but lack cr_core or gi_code attributes
2389	n/a	# (such as ones compiled with Cython).
2390	n/a
2391	n/a	class Coro:
2392	n/a	def send(self, v):
2393	n/a	pass
2394	n/a
2395	n/a	def throw(self, *exc):
2396	n/a	pass
2397	n/a
2398	n/a	def close(self):
2399	n/a	pass
2400	n/a
2401	n/a	def __await__(self):
2402	n/a	pass
2403	n/a
2404	n/a	coro = Coro()
2405	n/a	coro.__name__ = 'AAA'
2406	n/a	self.assertTrue(asyncio.iscoroutine(coro))
2407	n/a	self.assertEqual(coroutines._format_coroutine(coro), 'AAA()')
2408	n/a
2409	n/a	coro.__qualname__ = 'BBB'
2410	n/a	self.assertEqual(coroutines._format_coroutine(coro), 'BBB()')
2411	n/a
2412	n/a	coro.cr_running = True
2413	n/a	self.assertEqual(coroutines._format_coroutine(coro), 'BBB() running')
2414	n/a
2415	n/a	coro = Coro()
2416	n/a	# Some coroutines might not have '__name__', such as
2417	n/a	# built-in async_gen.asend().
2418	n/a	self.assertEqual(coroutines._format_coroutine(coro), 'Coro()')
2419	n/a
2420	n/a
2421	n/a	class TimerTests(unittest.TestCase):
2422	n/a
2423	n/a	def setUp(self):
2424	n/a	super().setUp()
2425	n/a	self.loop = mock.Mock()
2426	n/a
2427	n/a	def test_hash(self):
2428	n/a	when = time.monotonic()
2429	n/a	h = asyncio.TimerHandle(when, lambda: False, (),
2430	n/a	mock.Mock())
2431	n/a	self.assertEqual(hash(h), hash(when))
2432	n/a
2433	n/a	def test_timer(self):
2434	n/a	def callback(*args):
2435	n/a	return args
2436	n/a
2437	n/a	args = (1, 2, 3)
2438	n/a	when = time.monotonic()
2439	n/a	h = asyncio.TimerHandle(when, callback, args, mock.Mock())
2440	n/a	self.assertIs(h._callback, callback)
2441	n/a	self.assertIs(h._args, args)
2442	n/a	self.assertFalse(h._cancelled)
2443	n/a
2444	n/a	# cancel
2445	n/a	h.cancel()
2446	n/a	self.assertTrue(h._cancelled)
2447	n/a	self.assertIsNone(h._callback)
2448	n/a	self.assertIsNone(h._args)
2449	n/a
2450	n/a	# when cannot be None
2451	n/a	self.assertRaises(AssertionError,
2452	n/a	asyncio.TimerHandle, None, callback, args,
2453	n/a	self.loop)
2454	n/a
2455	n/a	def test_timer_repr(self):
2456	n/a	self.loop.get_debug.return_value = False
2457	n/a
2458	n/a	# simple function
2459	n/a	h = asyncio.TimerHandle(123, noop, (), self.loop)
2460	n/a	src = test_utils.get_function_source(noop)
2461	n/a	self.assertEqual(repr(h),
2462	n/a	'<TimerHandle when=123 noop() at %s:%s>' % src)
2463	n/a
2464	n/a	# cancelled handle
2465	n/a	h.cancel()
2466	n/a	self.assertEqual(repr(h),
2467	n/a	'<TimerHandle cancelled when=123>')
2468	n/a
2469	n/a	def test_timer_repr_debug(self):
2470	n/a	self.loop.get_debug.return_value = True
2471	n/a
2472	n/a	# simple function
2473	n/a	create_filename = __file__
2474	n/a	create_lineno = sys._getframe().f_lineno + 1
2475	n/a	h = asyncio.TimerHandle(123, noop, (), self.loop)
2476	n/a	filename, lineno = test_utils.get_function_source(noop)
2477	n/a	self.assertEqual(repr(h),
2478	n/a	'<TimerHandle when=123 noop() '
2479	n/a	'at %s:%s created at %s:%s>'
2480	n/a	% (filename, lineno, create_filename, create_lineno))
2481	n/a
2482	n/a	# cancelled handle
2483	n/a	h.cancel()
2484	n/a	self.assertEqual(repr(h),
2485	n/a	'<TimerHandle cancelled when=123 noop() '
2486	n/a	'at %s:%s created at %s:%s>'
2487	n/a	% (filename, lineno, create_filename, create_lineno))
2488	n/a
2489	n/a
2490	n/a	def test_timer_comparison(self):
2491	n/a	def callback(*args):
2492	n/a	return args
2493	n/a
2494	n/a	when = time.monotonic()
2495	n/a
2496	n/a	h1 = asyncio.TimerHandle(when, callback, (), self.loop)
2497	n/a	h2 = asyncio.TimerHandle(when, callback, (), self.loop)
2498	n/a	# TODO: Use assertLess etc.
2499	n/a	self.assertFalse(h1 < h2)
2500	n/a	self.assertFalse(h2 < h1)
2501	n/a	self.assertTrue(h1 <= h2)
2502	n/a	self.assertTrue(h2 <= h1)
2503	n/a	self.assertFalse(h1 > h2)
2504	n/a	self.assertFalse(h2 > h1)
2505	n/a	self.assertTrue(h1 >= h2)
2506	n/a	self.assertTrue(h2 >= h1)
2507	n/a	self.assertTrue(h1 == h2)
2508	n/a	self.assertFalse(h1 != h2)
2509	n/a
2510	n/a	h2.cancel()
2511	n/a	self.assertFalse(h1 == h2)
2512	n/a
2513	n/a	h1 = asyncio.TimerHandle(when, callback, (), self.loop)
2514	n/a	h2 = asyncio.TimerHandle(when + 10.0, callback, (), self.loop)
2515	n/a	self.assertTrue(h1 < h2)
2516	n/a	self.assertFalse(h2 < h1)
2517	n/a	self.assertTrue(h1 <= h2)
2518	n/a	self.assertFalse(h2 <= h1)
2519	n/a	self.assertFalse(h1 > h2)
2520	n/a	self.assertTrue(h2 > h1)
2521	n/a	self.assertFalse(h1 >= h2)
2522	n/a	self.assertTrue(h2 >= h1)
2523	n/a	self.assertFalse(h1 == h2)
2524	n/a	self.assertTrue(h1 != h2)
2525	n/a
2526	n/a	h3 = asyncio.Handle(callback, (), self.loop)
2527	n/a	self.assertIs(NotImplemented, h1.__eq__(h3))
2528	n/a	self.assertIs(NotImplemented, h1.__ne__(h3))
2529	n/a
2530	n/a
2531	n/a	class AbstractEventLoopTests(unittest.TestCase):
2532	n/a
2533	n/a	def test_not_implemented(self):
2534	n/a	f = mock.Mock()
2535	n/a	loop = asyncio.AbstractEventLoop()
2536	n/a	self.assertRaises(
2537	n/a	NotImplementedError, loop.run_forever)
2538	n/a	self.assertRaises(
2539	n/a	NotImplementedError, loop.run_until_complete, None)
2540	n/a	self.assertRaises(
2541	n/a	NotImplementedError, loop.stop)
2542	n/a	self.assertRaises(
2543	n/a	NotImplementedError, loop.is_running)
2544	n/a	self.assertRaises(
2545	n/a	NotImplementedError, loop.is_closed)
2546	n/a	self.assertRaises(
2547	n/a	NotImplementedError, loop.close)
2548	n/a	self.assertRaises(
2549	n/a	NotImplementedError, loop.create_task, None)
2550	n/a	self.assertRaises(
2551	n/a	NotImplementedError, loop.call_later, None, None)
2552	n/a	self.assertRaises(
2553	n/a	NotImplementedError, loop.call_at, f, f)
2554	n/a	self.assertRaises(
2555	n/a	NotImplementedError, loop.call_soon, None)
2556	n/a	self.assertRaises(
2557	n/a	NotImplementedError, loop.time)
2558	n/a	self.assertRaises(
2559	n/a	NotImplementedError, loop.call_soon_threadsafe, None)
2560	n/a	self.assertRaises(
2561	n/a	NotImplementedError, loop.run_in_executor, f, f)
2562	n/a	self.assertRaises(
2563	n/a	NotImplementedError, loop.set_default_executor, f)
2564	n/a	self.assertRaises(
2565	n/a	NotImplementedError, loop.getaddrinfo, 'localhost', 8080)
2566	n/a	self.assertRaises(
2567	n/a	NotImplementedError, loop.getnameinfo, ('localhost', 8080))
2568	n/a	self.assertRaises(
2569	n/a	NotImplementedError, loop.create_connection, f)
2570	n/a	self.assertRaises(
2571	n/a	NotImplementedError, loop.create_server, f)
2572	n/a	self.assertRaises(
2573	n/a	NotImplementedError, loop.create_datagram_endpoint, f)
2574	n/a	self.assertRaises(
2575	n/a	NotImplementedError, loop.add_reader, 1, f)
2576	n/a	self.assertRaises(
2577	n/a	NotImplementedError, loop.remove_reader, 1)
2578	n/a	self.assertRaises(
2579	n/a	NotImplementedError, loop.add_writer, 1, f)
2580	n/a	self.assertRaises(
2581	n/a	NotImplementedError, loop.remove_writer, 1)
2582	n/a	self.assertRaises(
2583	n/a	NotImplementedError, loop.sock_recv, f, 10)
2584	n/a	self.assertRaises(
2585	n/a	NotImplementedError, loop.sock_sendall, f, 10)
2586	n/a	self.assertRaises(
2587	n/a	NotImplementedError, loop.sock_connect, f, f)
2588	n/a	self.assertRaises(
2589	n/a	NotImplementedError, loop.sock_accept, f)
2590	n/a	self.assertRaises(
2591	n/a	NotImplementedError, loop.add_signal_handler, 1, f)
2592	n/a	self.assertRaises(
2593	n/a	NotImplementedError, loop.remove_signal_handler, 1)
2594	n/a	self.assertRaises(
2595	n/a	NotImplementedError, loop.remove_signal_handler, 1)
2596	n/a	self.assertRaises(
2597	n/a	NotImplementedError, loop.connect_read_pipe, f,
2598	n/a	mock.sentinel.pipe)
2599	n/a	self.assertRaises(
2600	n/a	NotImplementedError, loop.connect_write_pipe, f,
2601	n/a	mock.sentinel.pipe)
2602	n/a	self.assertRaises(
2603	n/a	NotImplementedError, loop.subprocess_shell, f,
2604	n/a	mock.sentinel)
2605	n/a	self.assertRaises(
2606	n/a	NotImplementedError, loop.subprocess_exec, f)
2607	n/a	self.assertRaises(
2608	n/a	NotImplementedError, loop.set_exception_handler, f)
2609	n/a	self.assertRaises(
2610	n/a	NotImplementedError, loop.default_exception_handler, f)
2611	n/a	self.assertRaises(
2612	n/a	NotImplementedError, loop.call_exception_handler, f)
2613	n/a	self.assertRaises(
2614	n/a	NotImplementedError, loop.get_debug)
2615	n/a	self.assertRaises(
2616	n/a	NotImplementedError, loop.set_debug, f)
2617	n/a
2618	n/a
2619	n/a	class ProtocolsAbsTests(unittest.TestCase):
2620	n/a
2621	n/a	def test_empty(self):
2622	n/a	f = mock.Mock()
2623	n/a	p = asyncio.Protocol()
2624	n/a	self.assertIsNone(p.connection_made(f))
2625	n/a	self.assertIsNone(p.connection_lost(f))
2626	n/a	self.assertIsNone(p.data_received(f))
2627	n/a	self.assertIsNone(p.eof_received())
2628	n/a
2629	n/a	dp = asyncio.DatagramProtocol()
2630	n/a	self.assertIsNone(dp.connection_made(f))
2631	n/a	self.assertIsNone(dp.connection_lost(f))
2632	n/a	self.assertIsNone(dp.error_received(f))
2633	n/a	self.assertIsNone(dp.datagram_received(f, f))
2634	n/a
2635	n/a	sp = asyncio.SubprocessProtocol()
2636	n/a	self.assertIsNone(sp.connection_made(f))
2637	n/a	self.assertIsNone(sp.connection_lost(f))
2638	n/a	self.assertIsNone(sp.pipe_data_received(1, f))
2639	n/a	self.assertIsNone(sp.pipe_connection_lost(1, f))
2640	n/a	self.assertIsNone(sp.process_exited())
2641	n/a
2642	n/a
2643	n/a	class PolicyTests(unittest.TestCase):
2644	n/a
2645	n/a	def test_event_loop_policy(self):
2646	n/a	policy = asyncio.AbstractEventLoopPolicy()
2647	n/a	self.assertRaises(NotImplementedError, policy.get_event_loop)
2648	n/a	self.assertRaises(NotImplementedError, policy.set_event_loop, object())
2649	n/a	self.assertRaises(NotImplementedError, policy.new_event_loop)
2650	n/a	self.assertRaises(NotImplementedError, policy.get_child_watcher)
2651	n/a	self.assertRaises(NotImplementedError, policy.set_child_watcher,
2652	n/a	object())
2653	n/a
2654	n/a	def test_get_event_loop(self):
2655	n/a	policy = asyncio.DefaultEventLoopPolicy()
2656	n/a	self.assertIsNone(policy._local._loop)
2657	n/a
2658	n/a	loop = policy.get_event_loop()
2659	n/a	self.assertIsInstance(loop, asyncio.AbstractEventLoop)
2660	n/a
2661	n/a	self.assertIs(policy._local._loop, loop)
2662	n/a	self.assertIs(loop, policy.get_event_loop())
2663	n/a	loop.close()
2664	n/a
2665	n/a	def test_get_event_loop_calls_set_event_loop(self):
2666	n/a	policy = asyncio.DefaultEventLoopPolicy()
2667	n/a
2668	n/a	with mock.patch.object(
2669	n/a	policy, "set_event_loop",
2670	n/a	wraps=policy.set_event_loop) as m_set_event_loop:
2671	n/a
2672	n/a	loop = policy.get_event_loop()
2673	n/a
2674	n/a	# policy._local._loop must be set through .set_event_loop()
2675	n/a	# (the unix DefaultEventLoopPolicy needs this call to attach
2676	n/a	# the child watcher correctly)
2677	n/a	m_set_event_loop.assert_called_with(loop)
2678	n/a
2679	n/a	loop.close()
2680	n/a
2681	n/a	def test_get_event_loop_after_set_none(self):
2682	n/a	policy = asyncio.DefaultEventLoopPolicy()
2683	n/a	policy.set_event_loop(None)
2684	n/a	self.assertRaises(RuntimeError, policy.get_event_loop)
2685	n/a
2686	n/a	@mock.patch('asyncio.events.threading.current_thread')
2687	n/a	def test_get_event_loop_thread(self, m_current_thread):
2688	n/a
2689	n/a	def f():
2690	n/a	policy = asyncio.DefaultEventLoopPolicy()
2691	n/a	self.assertRaises(RuntimeError, policy.get_event_loop)
2692	n/a
2693	n/a	th = threading.Thread(target=f)
2694	n/a	th.start()
2695	n/a	th.join()
2696	n/a
2697	n/a	def test_new_event_loop(self):
2698	n/a	policy = asyncio.DefaultEventLoopPolicy()
2699	n/a
2700	n/a	loop = policy.new_event_loop()
2701	n/a	self.assertIsInstance(loop, asyncio.AbstractEventLoop)
2702	n/a	loop.close()
2703	n/a
2704	n/a	def test_set_event_loop(self):
2705	n/a	policy = asyncio.DefaultEventLoopPolicy()
2706	n/a	old_loop = policy.get_event_loop()
2707	n/a
2708	n/a	self.assertRaises(AssertionError, policy.set_event_loop, object())
2709	n/a
2710	n/a	loop = policy.new_event_loop()
2711	n/a	policy.set_event_loop(loop)
2712	n/a	self.assertIs(loop, policy.get_event_loop())
2713	n/a	self.assertIsNot(old_loop, policy.get_event_loop())
2714	n/a	loop.close()
2715	n/a	old_loop.close()
2716	n/a
2717	n/a	def test_get_event_loop_policy(self):
2718	n/a	policy = asyncio.get_event_loop_policy()
2719	n/a	self.assertIsInstance(policy, asyncio.AbstractEventLoopPolicy)
2720	n/a	self.assertIs(policy, asyncio.get_event_loop_policy())
2721	n/a
2722	n/a	def test_set_event_loop_policy(self):
2723	n/a	self.assertRaises(
2724	n/a	AssertionError, asyncio.set_event_loop_policy, object())
2725	n/a
2726	n/a	old_policy = asyncio.get_event_loop_policy()
2727	n/a
2728	n/a	policy = asyncio.DefaultEventLoopPolicy()
2729	n/a	asyncio.set_event_loop_policy(policy)
2730	n/a	self.assertIs(policy, asyncio.get_event_loop_policy())
2731	n/a	self.assertIsNot(policy, old_policy)
2732	n/a
2733	n/a	def test_get_event_loop_returns_running_loop(self):
2734	n/a	class Policy(asyncio.DefaultEventLoopPolicy):
2735	n/a	def get_event_loop(self):
2736	n/a	raise NotImplementedError
2737	n/a
2738	n/a	loop = None
2739	n/a
2740	n/a	old_policy = asyncio.get_event_loop_policy()
2741	n/a	try:
2742	n/a	asyncio.set_event_loop_policy(Policy())
2743	n/a	loop = asyncio.new_event_loop()
2744	n/a	self.assertIs(asyncio._get_running_loop(), None)
2745	n/a
2746	n/a	async def func():
2747	n/a	self.assertIs(asyncio.get_event_loop(), loop)
2748	n/a	self.assertIs(asyncio._get_running_loop(), loop)
2749	n/a
2750	n/a	loop.run_until_complete(func())
2751	n/a	finally:
2752	n/a	asyncio.set_event_loop_policy(old_policy)
2753	n/a	if loop is not None:
2754	n/a	loop.close()
2755	n/a
2756	n/a	self.assertIs(asyncio._get_running_loop(), None)
2757	n/a
2758	n/a
2759	n/a	if __name__ == '__main__':
2760	n/a	unittest.main()