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

#	count	content
1	n/a	"""Tests for base_events.py"""
2	n/a
3	n/a	import errno
4	n/a	import logging
5	n/a	import math
6	n/a	import os
7	n/a	import socket
8	n/a	import sys
9	n/a	import threading
10	n/a	import time
11	n/a	import unittest
12	n/a	from unittest import mock
13	n/a
14	n/a	import asyncio
15	n/a	from asyncio import base_events
16	n/a	from asyncio import constants
17	n/a	from asyncio import test_utils
18	n/a	try:
19	n/a	from test import support
20	n/a	except ImportError:
21	n/a	from asyncio import test_support as support
22	n/a	try:
23	n/a	from test.support.script_helper import assert_python_ok
24	n/a	except ImportError:
25	n/a	try:
26	n/a	from test.script_helper import assert_python_ok
27	n/a	except ImportError:
28	n/a	from asyncio.test_support import assert_python_ok
29	n/a
30	n/a
31	n/a	MOCK_ANY = mock.ANY
32	n/a	PY34 = sys.version_info >= (3, 4)
33	n/a
34	n/a
35	n/a	def mock_socket_module():
36	n/a	m_socket = mock.MagicMock(spec=socket)
37	n/a	for name in (
38	n/a	'AF_INET', 'AF_INET6', 'AF_UNSPEC', 'IPPROTO_TCP', 'IPPROTO_UDP',
39	n/a	'SOCK_STREAM', 'SOCK_DGRAM', 'SOL_SOCKET', 'SO_REUSEADDR', 'inet_pton'
40	n/a	):
41	n/a	if hasattr(socket, name):
42	n/a	setattr(m_socket, name, getattr(socket, name))
43	n/a	else:
44	n/a	delattr(m_socket, name)
45	n/a
46	n/a	m_socket.socket = mock.MagicMock()
47	n/a	m_socket.socket.return_value = test_utils.mock_nonblocking_socket()
48	n/a	m_socket.getaddrinfo._is_coroutine = False
49	n/a
50	n/a	return m_socket
51	n/a
52	n/a
53	n/a	def patch_socket(f):
54	n/a	return mock.patch('asyncio.base_events.socket',
55	n/a	new_callable=mock_socket_module)(f)
56	n/a
57	n/a
58	n/a	class BaseEventTests(test_utils.TestCase):
59	n/a
60	n/a	def test_ipaddr_info(self):
61	n/a	UNSPEC = socket.AF_UNSPEC
62	n/a	INET = socket.AF_INET
63	n/a	INET6 = socket.AF_INET6
64	n/a	STREAM = socket.SOCK_STREAM
65	n/a	DGRAM = socket.SOCK_DGRAM
66	n/a	TCP = socket.IPPROTO_TCP
67	n/a	UDP = socket.IPPROTO_UDP
68	n/a
69	n/a	self.assertEqual(
70	n/a	(INET, STREAM, TCP, '', ('1.2.3.4', 1)),
71	n/a	base_events._ipaddr_info('1.2.3.4', 1, INET, STREAM, TCP))
72	n/a
73	n/a	self.assertEqual(
74	n/a	(INET, STREAM, TCP, '', ('1.2.3.4', 1)),
75	n/a	base_events._ipaddr_info(b'1.2.3.4', 1, INET, STREAM, TCP))
76	n/a
77	n/a	self.assertEqual(
78	n/a	(INET, STREAM, TCP, '', ('1.2.3.4', 1)),
79	n/a	base_events._ipaddr_info('1.2.3.4', 1, UNSPEC, STREAM, TCP))
80	n/a
81	n/a	self.assertEqual(
82	n/a	(INET, DGRAM, UDP, '', ('1.2.3.4', 1)),
83	n/a	base_events._ipaddr_info('1.2.3.4', 1, UNSPEC, DGRAM, UDP))
84	n/a
85	n/a	# Socket type STREAM implies TCP protocol.
86	n/a	self.assertEqual(
87	n/a	(INET, STREAM, TCP, '', ('1.2.3.4', 1)),
88	n/a	base_events._ipaddr_info('1.2.3.4', 1, UNSPEC, STREAM, 0))
89	n/a
90	n/a	# Socket type DGRAM implies UDP protocol.
91	n/a	self.assertEqual(
92	n/a	(INET, DGRAM, UDP, '', ('1.2.3.4', 1)),
93	n/a	base_events._ipaddr_info('1.2.3.4', 1, UNSPEC, DGRAM, 0))
94	n/a
95	n/a	# No socket type.
96	n/a	self.assertIsNone(
97	n/a	base_events._ipaddr_info('1.2.3.4', 1, UNSPEC, 0, 0))
98	n/a
99	n/a	# IPv4 address with family IPv6.
100	n/a	self.assertIsNone(
101	n/a	base_events._ipaddr_info('1.2.3.4', 1, INET6, STREAM, TCP))
102	n/a
103	n/a	self.assertEqual(
104	n/a	(INET6, STREAM, TCP, '', ('::3', 1)),
105	n/a	base_events._ipaddr_info('::3', 1, INET6, STREAM, TCP))
106	n/a
107	n/a	self.assertEqual(
108	n/a	(INET6, STREAM, TCP, '', ('::3', 1)),
109	n/a	base_events._ipaddr_info('::3', 1, UNSPEC, STREAM, TCP))
110	n/a
111	n/a	# IPv6 address with family IPv4.
112	n/a	self.assertIsNone(
113	n/a	base_events._ipaddr_info('::3', 1, INET, STREAM, TCP))
114	n/a
115	n/a	# IPv6 address with zone index.
116	n/a	self.assertIsNone(
117	n/a	base_events._ipaddr_info('::3%lo0', 1, INET6, STREAM, TCP))
118	n/a
119	n/a	if hasattr(socket, 'SOCK_NONBLOCK'):
120	n/a	self.assertEqual(
121	n/a	None,
122	n/a	base_events._ipaddr_info(
123	n/a	'1.2.3.4', 1, INET, STREAM \| socket.SOCK_NONBLOCK, TCP))
124	n/a
125	n/a
126	n/a	def test_port_parameter_types(self):
127	n/a	# Test obscure kinds of arguments for "port".
128	n/a	INET = socket.AF_INET
129	n/a	STREAM = socket.SOCK_STREAM
130	n/a	TCP = socket.IPPROTO_TCP
131	n/a
132	n/a	self.assertEqual(
133	n/a	(INET, STREAM, TCP, '', ('1.2.3.4', 0)),
134	n/a	base_events._ipaddr_info('1.2.3.4', None, INET, STREAM, TCP))
135	n/a
136	n/a	self.assertEqual(
137	n/a	(INET, STREAM, TCP, '', ('1.2.3.4', 0)),
138	n/a	base_events._ipaddr_info('1.2.3.4', b'', INET, STREAM, TCP))
139	n/a
140	n/a	self.assertEqual(
141	n/a	(INET, STREAM, TCP, '', ('1.2.3.4', 0)),
142	n/a	base_events._ipaddr_info('1.2.3.4', '', INET, STREAM, TCP))
143	n/a
144	n/a	self.assertEqual(
145	n/a	(INET, STREAM, TCP, '', ('1.2.3.4', 1)),
146	n/a	base_events._ipaddr_info('1.2.3.4', '1', INET, STREAM, TCP))
147	n/a
148	n/a	self.assertEqual(
149	n/a	(INET, STREAM, TCP, '', ('1.2.3.4', 1)),
150	n/a	base_events._ipaddr_info('1.2.3.4', b'1', INET, STREAM, TCP))
151	n/a
152	n/a	@patch_socket
153	n/a	def test_ipaddr_info_no_inet_pton(self, m_socket):
154	n/a	del m_socket.inet_pton
155	n/a	self.assertIsNone(base_events._ipaddr_info('1.2.3.4', 1,
156	n/a	socket.AF_INET,
157	n/a	socket.SOCK_STREAM,
158	n/a	socket.IPPROTO_TCP))
159	n/a
160	n/a
161	n/a	class BaseEventLoopTests(test_utils.TestCase):
162	n/a
163	n/a	def setUp(self):
164	n/a	super().setUp()
165	n/a	self.loop = base_events.BaseEventLoop()
166	n/a	self.loop._selector = mock.Mock()
167	n/a	self.loop._selector.select.return_value = ()
168	n/a	self.set_event_loop(self.loop)
169	n/a
170	n/a	def test_not_implemented(self):
171	n/a	m = mock.Mock()
172	n/a	self.assertRaises(
173	n/a	NotImplementedError,
174	n/a	self.loop._make_socket_transport, m, m)
175	n/a	self.assertRaises(
176	n/a	NotImplementedError,
177	n/a	self.loop._make_ssl_transport, m, m, m, m)
178	n/a	self.assertRaises(
179	n/a	NotImplementedError,
180	n/a	self.loop._make_datagram_transport, m, m)
181	n/a	self.assertRaises(
182	n/a	NotImplementedError, self.loop._process_events, [])
183	n/a	self.assertRaises(
184	n/a	NotImplementedError, self.loop._write_to_self)
185	n/a	self.assertRaises(
186	n/a	NotImplementedError,
187	n/a	self.loop._make_read_pipe_transport, m, m)
188	n/a	self.assertRaises(
189	n/a	NotImplementedError,
190	n/a	self.loop._make_write_pipe_transport, m, m)
191	n/a	gen = self.loop._make_subprocess_transport(m, m, m, m, m, m, m)
192	n/a	with self.assertRaises(NotImplementedError):
193	n/a	gen.send(None)
194	n/a
195	n/a	def test_close(self):
196	n/a	self.assertFalse(self.loop.is_closed())
197	n/a	self.loop.close()
198	n/a	self.assertTrue(self.loop.is_closed())
199	n/a
200	n/a	# it should be possible to call close() more than once
201	n/a	self.loop.close()
202	n/a	self.loop.close()
203	n/a
204	n/a	# operation blocked when the loop is closed
205	n/a	f = asyncio.Future(loop=self.loop)
206	n/a	self.assertRaises(RuntimeError, self.loop.run_forever)
207	n/a	self.assertRaises(RuntimeError, self.loop.run_until_complete, f)
208	n/a
209	n/a	def test__add_callback_handle(self):
210	n/a	h = asyncio.Handle(lambda: False, (), self.loop)
211	n/a
212	n/a	self.loop._add_callback(h)
213	n/a	self.assertFalse(self.loop._scheduled)
214	n/a	self.assertIn(h, self.loop._ready)
215	n/a
216	n/a	def test__add_callback_cancelled_handle(self):
217	n/a	h = asyncio.Handle(lambda: False, (), self.loop)
218	n/a	h.cancel()
219	n/a
220	n/a	self.loop._add_callback(h)
221	n/a	self.assertFalse(self.loop._scheduled)
222	n/a	self.assertFalse(self.loop._ready)
223	n/a
224	n/a	def test_set_default_executor(self):
225	n/a	executor = mock.Mock()
226	n/a	self.loop.set_default_executor(executor)
227	n/a	self.assertIs(executor, self.loop._default_executor)
228	n/a
229	n/a	def test_getnameinfo(self):
230	n/a	sockaddr = mock.Mock()
231	n/a	self.loop.run_in_executor = mock.Mock()
232	n/a	self.loop.getnameinfo(sockaddr)
233	n/a	self.assertEqual(
234	n/a	(None, socket.getnameinfo, sockaddr, 0),
235	n/a	self.loop.run_in_executor.call_args[0])
236	n/a
237	n/a	def test_call_soon(self):
238	n/a	def cb():
239	n/a	pass
240	n/a
241	n/a	h = self.loop.call_soon(cb)
242	n/a	self.assertEqual(h._callback, cb)
243	n/a	self.assertIsInstance(h, asyncio.Handle)
244	n/a	self.assertIn(h, self.loop._ready)
245	n/a
246	n/a	def test_call_soon_non_callable(self):
247	n/a	self.loop.set_debug(True)
248	n/a	with self.assertRaisesRegex(TypeError, 'a callable object'):
249	n/a	self.loop.call_soon(1)
250	n/a
251	n/a	def test_call_later(self):
252	n/a	def cb():
253	n/a	pass
254	n/a
255	n/a	h = self.loop.call_later(10.0, cb)
256	n/a	self.assertIsInstance(h, asyncio.TimerHandle)
257	n/a	self.assertIn(h, self.loop._scheduled)
258	n/a	self.assertNotIn(h, self.loop._ready)
259	n/a
260	n/a	def test_call_later_negative_delays(self):
261	n/a	calls = []
262	n/a
263	n/a	def cb(arg):
264	n/a	calls.append(arg)
265	n/a
266	n/a	self.loop._process_events = mock.Mock()
267	n/a	self.loop.call_later(-1, cb, 'a')
268	n/a	self.loop.call_later(-2, cb, 'b')
269	n/a	test_utils.run_briefly(self.loop)
270	n/a	self.assertEqual(calls, ['b', 'a'])
271	n/a
272	n/a	def test_time_and_call_at(self):
273	n/a	def cb():
274	n/a	self.loop.stop()
275	n/a
276	n/a	self.loop._process_events = mock.Mock()
277	n/a	delay = 0.1
278	n/a
279	n/a	when = self.loop.time() + delay
280	n/a	self.loop.call_at(when, cb)
281	n/a	t0 = self.loop.time()
282	n/a	self.loop.run_forever()
283	n/a	dt = self.loop.time() - t0
284	n/a
285	n/a	# 50 ms: maximum granularity of the event loop
286	n/a	self.assertGreaterEqual(dt, delay - 0.050, dt)
287	n/a	# tolerate a difference of +800 ms because some Python buildbots
288	n/a	# are really slow
289	n/a	self.assertLessEqual(dt, 0.9, dt)
290	n/a
291	n/a	def check_thread(self, loop, debug):
292	n/a	def cb():
293	n/a	pass
294	n/a
295	n/a	loop.set_debug(debug)
296	n/a	if debug:
297	n/a	msg = ("Non-thread-safe operation invoked on an event loop other "
298	n/a	"than the current one")
299	n/a	with self.assertRaisesRegex(RuntimeError, msg):
300	n/a	loop.call_soon(cb)
301	n/a	with self.assertRaisesRegex(RuntimeError, msg):
302	n/a	loop.call_later(60, cb)
303	n/a	with self.assertRaisesRegex(RuntimeError, msg):
304	n/a	loop.call_at(loop.time() + 60, cb)
305	n/a	else:
306	n/a	loop.call_soon(cb)
307	n/a	loop.call_later(60, cb)
308	n/a	loop.call_at(loop.time() + 60, cb)
309	n/a
310	n/a	def test_check_thread(self):
311	n/a	def check_in_thread(loop, event, debug, create_loop, fut):
312	n/a	# wait until the event loop is running
313	n/a	event.wait()
314	n/a
315	n/a	try:
316	n/a	if create_loop:
317	n/a	loop2 = base_events.BaseEventLoop()
318	n/a	try:
319	n/a	asyncio.set_event_loop(loop2)
320	n/a	self.check_thread(loop, debug)
321	n/a	finally:
322	n/a	asyncio.set_event_loop(None)
323	n/a	loop2.close()
324	n/a	else:
325	n/a	self.check_thread(loop, debug)
326	n/a	except Exception as exc:
327	n/a	loop.call_soon_threadsafe(fut.set_exception, exc)
328	n/a	else:
329	n/a	loop.call_soon_threadsafe(fut.set_result, None)
330	n/a
331	n/a	def test_thread(loop, debug, create_loop=False):
332	n/a	event = threading.Event()
333	n/a	fut = asyncio.Future(loop=loop)
334	n/a	loop.call_soon(event.set)
335	n/a	args = (loop, event, debug, create_loop, fut)
336	n/a	thread = threading.Thread(target=check_in_thread, args=args)
337	n/a	thread.start()
338	n/a	loop.run_until_complete(fut)
339	n/a	thread.join()
340	n/a
341	n/a	self.loop._process_events = mock.Mock()
342	n/a	self.loop._write_to_self = mock.Mock()
343	n/a
344	n/a	# raise RuntimeError if the thread has no event loop
345	n/a	test_thread(self.loop, True)
346	n/a
347	n/a	# check disabled if debug mode is disabled
348	n/a	test_thread(self.loop, False)
349	n/a
350	n/a	# raise RuntimeError if the event loop of the thread is not the called
351	n/a	# event loop
352	n/a	test_thread(self.loop, True, create_loop=True)
353	n/a
354	n/a	# check disabled if debug mode is disabled
355	n/a	test_thread(self.loop, False, create_loop=True)
356	n/a
357	n/a	def test_run_once_in_executor_plain(self):
358	n/a	def cb():
359	n/a	pass
360	n/a	f = asyncio.Future(loop=self.loop)
361	n/a	executor = mock.Mock()
362	n/a	executor.submit.return_value = f
363	n/a
364	n/a	self.loop.set_default_executor(executor)
365	n/a
366	n/a	res = self.loop.run_in_executor(None, cb)
367	n/a	self.assertIs(f, res)
368	n/a
369	n/a	executor = mock.Mock()
370	n/a	executor.submit.return_value = f
371	n/a	res = self.loop.run_in_executor(executor, cb)
372	n/a	self.assertIs(f, res)
373	n/a	self.assertTrue(executor.submit.called)
374	n/a
375	n/a	f.cancel() # Don't complain about abandoned Future.
376	n/a
377	n/a	def test__run_once(self):
378	n/a	h1 = asyncio.TimerHandle(time.monotonic() + 5.0, lambda: True, (),
379	n/a	self.loop)
380	n/a	h2 = asyncio.TimerHandle(time.monotonic() + 10.0, lambda: True, (),
381	n/a	self.loop)
382	n/a
383	n/a	h1.cancel()
384	n/a
385	n/a	self.loop._process_events = mock.Mock()
386	n/a	self.loop._scheduled.append(h1)
387	n/a	self.loop._scheduled.append(h2)
388	n/a	self.loop._run_once()
389	n/a
390	n/a	t = self.loop._selector.select.call_args[0][0]
391	n/a	self.assertTrue(9.5 < t < 10.5, t)
392	n/a	self.assertEqual([h2], self.loop._scheduled)
393	n/a	self.assertTrue(self.loop._process_events.called)
394	n/a
395	n/a	def test_set_debug(self):
396	n/a	self.loop.set_debug(True)
397	n/a	self.assertTrue(self.loop.get_debug())
398	n/a	self.loop.set_debug(False)
399	n/a	self.assertFalse(self.loop.get_debug())
400	n/a
401	n/a	@mock.patch('asyncio.base_events.logger')
402	n/a	def test__run_once_logging(self, m_logger):
403	n/a	def slow_select(timeout):
404	n/a	# Sleep a bit longer than a second to avoid timer resolution
405	n/a	# issues.
406	n/a	time.sleep(1.1)
407	n/a	return []
408	n/a
409	n/a	# logging needs debug flag
410	n/a	self.loop.set_debug(True)
411	n/a
412	n/a	# Log to INFO level if timeout > 1.0 sec.
413	n/a	self.loop._selector.select = slow_select
414	n/a	self.loop._process_events = mock.Mock()
415	n/a	self.loop._run_once()
416	n/a	self.assertEqual(logging.INFO, m_logger.log.call_args[0][0])
417	n/a
418	n/a	def fast_select(timeout):
419	n/a	time.sleep(0.001)
420	n/a	return []
421	n/a
422	n/a	self.loop._selector.select = fast_select
423	n/a	self.loop._run_once()
424	n/a	self.assertEqual(logging.DEBUG, m_logger.log.call_args[0][0])
425	n/a
426	n/a	def test__run_once_schedule_handle(self):
427	n/a	handle = None
428	n/a	processed = False
429	n/a
430	n/a	def cb(loop):
431	n/a	nonlocal processed, handle
432	n/a	processed = True
433	n/a	handle = loop.call_soon(lambda: True)
434	n/a
435	n/a	h = asyncio.TimerHandle(time.monotonic() - 1, cb, (self.loop,),
436	n/a	self.loop)
437	n/a
438	n/a	self.loop._process_events = mock.Mock()
439	n/a	self.loop._scheduled.append(h)
440	n/a	self.loop._run_once()
441	n/a
442	n/a	self.assertTrue(processed)
443	n/a	self.assertEqual([handle], list(self.loop._ready))
444	n/a
445	n/a	def test__run_once_cancelled_event_cleanup(self):
446	n/a	self.loop._process_events = mock.Mock()
447	n/a
448	n/a	self.assertTrue(
449	n/a	0 < base_events._MIN_CANCELLED_TIMER_HANDLES_FRACTION < 1.0)
450	n/a
451	n/a	def cb():
452	n/a	pass
453	n/a
454	n/a	# Set up one "blocking" event that will not be cancelled to
455	n/a	# ensure later cancelled events do not make it to the head
456	n/a	# of the queue and get cleaned.
457	n/a	not_cancelled_count = 1
458	n/a	self.loop.call_later(3000, cb)
459	n/a
460	n/a	# Add less than threshold (base_events._MIN_SCHEDULED_TIMER_HANDLES)
461	n/a	# cancelled handles, ensure they aren't removed
462	n/a
463	n/a	cancelled_count = 2
464	n/a	for x in range(2):
465	n/a	h = self.loop.call_later(3600, cb)
466	n/a	h.cancel()
467	n/a
468	n/a	# Add some cancelled events that will be at head and removed
469	n/a	cancelled_count += 2
470	n/a	for x in range(2):
471	n/a	h = self.loop.call_later(100, cb)
472	n/a	h.cancel()
473	n/a
474	n/a	# This test is invalid if _MIN_SCHEDULED_TIMER_HANDLES is too low
475	n/a	self.assertLessEqual(cancelled_count + not_cancelled_count,
476	n/a	base_events._MIN_SCHEDULED_TIMER_HANDLES)
477	n/a
478	n/a	self.assertEqual(self.loop._timer_cancelled_count, cancelled_count)
479	n/a
480	n/a	self.loop._run_once()
481	n/a
482	n/a	cancelled_count -= 2
483	n/a
484	n/a	self.assertEqual(self.loop._timer_cancelled_count, cancelled_count)
485	n/a
486	n/a	self.assertEqual(len(self.loop._scheduled),
487	n/a	cancelled_count + not_cancelled_count)
488	n/a
489	n/a	# Need enough events to pass _MIN_CANCELLED_TIMER_HANDLES_FRACTION
490	n/a	# so that deletion of cancelled events will occur on next _run_once
491	n/a	add_cancel_count = int(math.ceil(
492	n/a	base_events._MIN_SCHEDULED_TIMER_HANDLES *
493	n/a	base_events._MIN_CANCELLED_TIMER_HANDLES_FRACTION)) + 1
494	n/a
495	n/a	add_not_cancel_count = max(base_events._MIN_SCHEDULED_TIMER_HANDLES -
496	n/a	add_cancel_count, 0)
497	n/a
498	n/a	# Add some events that will not be cancelled
499	n/a	not_cancelled_count += add_not_cancel_count
500	n/a	for x in range(add_not_cancel_count):
501	n/a	self.loop.call_later(3600, cb)
502	n/a
503	n/a	# Add enough cancelled events
504	n/a	cancelled_count += add_cancel_count
505	n/a	for x in range(add_cancel_count):
506	n/a	h = self.loop.call_later(3600, cb)
507	n/a	h.cancel()
508	n/a
509	n/a	# Ensure all handles are still scheduled
510	n/a	self.assertEqual(len(self.loop._scheduled),
511	n/a	cancelled_count + not_cancelled_count)
512	n/a
513	n/a	self.loop._run_once()
514	n/a
515	n/a	# Ensure cancelled events were removed
516	n/a	self.assertEqual(len(self.loop._scheduled), not_cancelled_count)
517	n/a
518	n/a	# Ensure only uncancelled events remain scheduled
519	n/a	self.assertTrue(all([not x._cancelled for x in self.loop._scheduled]))
520	n/a
521	n/a	def test_run_until_complete_type_error(self):
522	n/a	self.assertRaises(TypeError,
523	n/a	self.loop.run_until_complete, 'blah')
524	n/a
525	n/a	def test_run_until_complete_loop(self):
526	n/a	task = asyncio.Future(loop=self.loop)
527	n/a	other_loop = self.new_test_loop()
528	n/a	self.addCleanup(other_loop.close)
529	n/a	self.assertRaises(ValueError,
530	n/a	other_loop.run_until_complete, task)
531	n/a
532	n/a	def test_subprocess_exec_invalid_args(self):
533	n/a	args = [sys.executable, '-c', 'pass']
534	n/a
535	n/a	# missing program parameter (empty args)
536	n/a	self.assertRaises(TypeError,
537	n/a	self.loop.run_until_complete, self.loop.subprocess_exec,
538	n/a	asyncio.SubprocessProtocol)
539	n/a
540	n/a	# expected multiple arguments, not a list
541	n/a	self.assertRaises(TypeError,
542	n/a	self.loop.run_until_complete, self.loop.subprocess_exec,
543	n/a	asyncio.SubprocessProtocol, args)
544	n/a
545	n/a	# program arguments must be strings, not int
546	n/a	self.assertRaises(TypeError,
547	n/a	self.loop.run_until_complete, self.loop.subprocess_exec,
548	n/a	asyncio.SubprocessProtocol, sys.executable, 123)
549	n/a
550	n/a	# universal_newlines, shell, bufsize must not be set
551	n/a	self.assertRaises(TypeError,
552	n/a	self.loop.run_until_complete, self.loop.subprocess_exec,
553	n/a	asyncio.SubprocessProtocol, *args, universal_newlines=True)
554	n/a	self.assertRaises(TypeError,
555	n/a	self.loop.run_until_complete, self.loop.subprocess_exec,
556	n/a	asyncio.SubprocessProtocol, *args, shell=True)
557	n/a	self.assertRaises(TypeError,
558	n/a	self.loop.run_until_complete, self.loop.subprocess_exec,
559	n/a	asyncio.SubprocessProtocol, *args, bufsize=4096)
560	n/a
561	n/a	def test_subprocess_shell_invalid_args(self):
562	n/a	# expected a string, not an int or a list
563	n/a	self.assertRaises(TypeError,
564	n/a	self.loop.run_until_complete, self.loop.subprocess_shell,
565	n/a	asyncio.SubprocessProtocol, 123)
566	n/a	self.assertRaises(TypeError,
567	n/a	self.loop.run_until_complete, self.loop.subprocess_shell,
568	n/a	asyncio.SubprocessProtocol, [sys.executable, '-c', 'pass'])
569	n/a
570	n/a	# universal_newlines, shell, bufsize must not be set
571	n/a	self.assertRaises(TypeError,
572	n/a	self.loop.run_until_complete, self.loop.subprocess_shell,
573	n/a	asyncio.SubprocessProtocol, 'exit 0', universal_newlines=True)
574	n/a	self.assertRaises(TypeError,
575	n/a	self.loop.run_until_complete, self.loop.subprocess_shell,
576	n/a	asyncio.SubprocessProtocol, 'exit 0', shell=True)
577	n/a	self.assertRaises(TypeError,
578	n/a	self.loop.run_until_complete, self.loop.subprocess_shell,
579	n/a	asyncio.SubprocessProtocol, 'exit 0', bufsize=4096)
580	n/a
581	n/a	def test_default_exc_handler_callback(self):
582	n/a	self.loop._process_events = mock.Mock()
583	n/a
584	n/a	def zero_error(fut):
585	n/a	fut.set_result(True)
586	n/a	1/0
587	n/a
588	n/a	# Test call_soon (events.Handle)
589	n/a	with mock.patch('asyncio.base_events.logger') as log:
590	n/a	fut = asyncio.Future(loop=self.loop)
591	n/a	self.loop.call_soon(zero_error, fut)
592	n/a	fut.add_done_callback(lambda fut: self.loop.stop())
593	n/a	self.loop.run_forever()
594	n/a	log.error.assert_called_with(
595	n/a	test_utils.MockPattern('Exception in callback.*zero'),
596	n/a	exc_info=(ZeroDivisionError, MOCK_ANY, MOCK_ANY))
597	n/a
598	n/a	# Test call_later (events.TimerHandle)
599	n/a	with mock.patch('asyncio.base_events.logger') as log:
600	n/a	fut = asyncio.Future(loop=self.loop)
601	n/a	self.loop.call_later(0.01, zero_error, fut)
602	n/a	fut.add_done_callback(lambda fut: self.loop.stop())
603	n/a	self.loop.run_forever()
604	n/a	log.error.assert_called_with(
605	n/a	test_utils.MockPattern('Exception in callback.*zero'),
606	n/a	exc_info=(ZeroDivisionError, MOCK_ANY, MOCK_ANY))
607	n/a
608	n/a	def test_default_exc_handler_coro(self):
609	n/a	self.loop._process_events = mock.Mock()
610	n/a
611	n/a	@asyncio.coroutine
612	n/a	def zero_error_coro():
613	n/a	yield from asyncio.sleep(0.01, loop=self.loop)
614	n/a	1/0
615	n/a
616	n/a	# Test Future.__del__
617	n/a	with mock.patch('asyncio.base_events.logger') as log:
618	n/a	fut = asyncio.ensure_future(zero_error_coro(), loop=self.loop)
619	n/a	fut.add_done_callback(lambda *args: self.loop.stop())
620	n/a	self.loop.run_forever()
621	n/a	fut = None # Trigger Future.__del__ or futures._TracebackLogger
622	n/a	support.gc_collect()
623	n/a	if PY34:
624	n/a	# Future.__del__ in Python 3.4 logs error with
625	n/a	# an actual exception context
626	n/a	log.error.assert_called_with(
627	n/a	test_utils.MockPattern('.*exception was never retrieved'),
628	n/a	exc_info=(ZeroDivisionError, MOCK_ANY, MOCK_ANY))
629	n/a	else:
630	n/a	# futures._TracebackLogger logs only textual traceback
631	n/a	log.error.assert_called_with(
632	n/a	test_utils.MockPattern(
633	n/a	'.exception was never retrieved.ZeroDiv'),
634	n/a	exc_info=False)
635	n/a
636	n/a	def test_set_exc_handler_invalid(self):
637	n/a	with self.assertRaisesRegex(TypeError, 'A callable object or None'):
638	n/a	self.loop.set_exception_handler('spam')
639	n/a
640	n/a	def test_set_exc_handler_custom(self):
641	n/a	def zero_error():
642	n/a	1/0
643	n/a
644	n/a	def run_loop():
645	n/a	handle = self.loop.call_soon(zero_error)
646	n/a	self.loop._run_once()
647	n/a	return handle
648	n/a
649	n/a	self.loop.set_debug(True)
650	n/a	self.loop._process_events = mock.Mock()
651	n/a
652	n/a	self.assertIsNone(self.loop.get_exception_handler())
653	n/a	mock_handler = mock.Mock()
654	n/a	self.loop.set_exception_handler(mock_handler)
655	n/a	self.assertIs(self.loop.get_exception_handler(), mock_handler)
656	n/a	handle = run_loop()
657	n/a	mock_handler.assert_called_with(self.loop, {
658	n/a	'exception': MOCK_ANY,
659	n/a	'message': test_utils.MockPattern(
660	n/a	'Exception in callback.*zero_error'),
661	n/a	'handle': handle,
662	n/a	'source_traceback': handle._source_traceback,
663	n/a	})
664	n/a	mock_handler.reset_mock()
665	n/a
666	n/a	self.loop.set_exception_handler(None)
667	n/a	with mock.patch('asyncio.base_events.logger') as log:
668	n/a	run_loop()
669	n/a	log.error.assert_called_with(
670	n/a	test_utils.MockPattern(
671	n/a	'Exception in callback.*zero'),
672	n/a	exc_info=(ZeroDivisionError, MOCK_ANY, MOCK_ANY))
673	n/a
674	n/a	assert not mock_handler.called
675	n/a
676	n/a	def test_set_exc_handler_broken(self):
677	n/a	def run_loop():
678	n/a	def zero_error():
679	n/a	1/0
680	n/a	self.loop.call_soon(zero_error)
681	n/a	self.loop._run_once()
682	n/a
683	n/a	def handler(loop, context):
684	n/a	raise AttributeError('spam')
685	n/a
686	n/a	self.loop._process_events = mock.Mock()
687	n/a
688	n/a	self.loop.set_exception_handler(handler)
689	n/a
690	n/a	with mock.patch('asyncio.base_events.logger') as log:
691	n/a	run_loop()
692	n/a	log.error.assert_called_with(
693	n/a	test_utils.MockPattern(
694	n/a	'Unhandled error in exception handler'),
695	n/a	exc_info=(AttributeError, MOCK_ANY, MOCK_ANY))
696	n/a
697	n/a	def test_default_exc_handler_broken(self):
698	n/a	_context = None
699	n/a
700	n/a	class Loop(base_events.BaseEventLoop):
701	n/a
702	n/a	_selector = mock.Mock()
703	n/a	_process_events = mock.Mock()
704	n/a
705	n/a	def default_exception_handler(self, context):
706	n/a	nonlocal _context
707	n/a	_context = context
708	n/a	# Simulates custom buggy "default_exception_handler"
709	n/a	raise ValueError('spam')
710	n/a
711	n/a	loop = Loop()
712	n/a	self.addCleanup(loop.close)
713	n/a	asyncio.set_event_loop(loop)
714	n/a
715	n/a	def run_loop():
716	n/a	def zero_error():
717	n/a	1/0
718	n/a	loop.call_soon(zero_error)
719	n/a	loop._run_once()
720	n/a
721	n/a	with mock.patch('asyncio.base_events.logger') as log:
722	n/a	run_loop()
723	n/a	log.error.assert_called_with(
724	n/a	'Exception in default exception handler',
725	n/a	exc_info=True)
726	n/a
727	n/a	def custom_handler(loop, context):
728	n/a	raise ValueError('ham')
729	n/a
730	n/a	_context = None
731	n/a	loop.set_exception_handler(custom_handler)
732	n/a	with mock.patch('asyncio.base_events.logger') as log:
733	n/a	run_loop()
734	n/a	log.error.assert_called_with(
735	n/a	test_utils.MockPattern('Exception in default exception.*'
736	n/a	'while handling.*in custom'),
737	n/a	exc_info=True)
738	n/a
739	n/a	# Check that original context was passed to default
740	n/a	# exception handler.
741	n/a	self.assertIn('context', _context)
742	n/a	self.assertIs(type(_context['context']['exception']),
743	n/a	ZeroDivisionError)
744	n/a
745	n/a	def test_set_task_factory_invalid(self):
746	n/a	with self.assertRaisesRegex(
747	n/a	TypeError, 'task factory must be a callable or None'):
748	n/a
749	n/a	self.loop.set_task_factory(1)
750	n/a
751	n/a	self.assertIsNone(self.loop.get_task_factory())
752	n/a
753	n/a	def test_set_task_factory(self):
754	n/a	self.loop._process_events = mock.Mock()
755	n/a
756	n/a	class MyTask(asyncio.Task):
757	n/a	pass
758	n/a
759	n/a	@asyncio.coroutine
760	n/a	def coro():
761	n/a	pass
762	n/a
763	n/a	factory = lambda loop, coro: MyTask(coro, loop=loop)
764	n/a
765	n/a	self.assertIsNone(self.loop.get_task_factory())
766	n/a	self.loop.set_task_factory(factory)
767	n/a	self.assertIs(self.loop.get_task_factory(), factory)
768	n/a
769	n/a	task = self.loop.create_task(coro())
770	n/a	self.assertTrue(isinstance(task, MyTask))
771	n/a	self.loop.run_until_complete(task)
772	n/a
773	n/a	self.loop.set_task_factory(None)
774	n/a	self.assertIsNone(self.loop.get_task_factory())
775	n/a
776	n/a	task = self.loop.create_task(coro())
777	n/a	self.assertTrue(isinstance(task, asyncio.Task))
778	n/a	self.assertFalse(isinstance(task, MyTask))
779	n/a	self.loop.run_until_complete(task)
780	n/a
781	n/a	def test_env_var_debug(self):
782	n/a	code = '\n'.join((
783	n/a	'import asyncio',
784	n/a	'loop = asyncio.get_event_loop()',
785	n/a	'print(loop.get_debug())'))
786	n/a
787	n/a	# Test with -E to not fail if the unit test was run with
788	n/a	# PYTHONASYNCIODEBUG set to a non-empty string
789	n/a	sts, stdout, stderr = assert_python_ok('-E', '-c', code)
790	n/a	self.assertEqual(stdout.rstrip(), b'False')
791	n/a
792	n/a	sts, stdout, stderr = assert_python_ok('-c', code,
793	n/a	PYTHONASYNCIODEBUG='')
794	n/a	self.assertEqual(stdout.rstrip(), b'False')
795	n/a
796	n/a	sts, stdout, stderr = assert_python_ok('-c', code,
797	n/a	PYTHONASYNCIODEBUG='1')
798	n/a	self.assertEqual(stdout.rstrip(), b'True')
799	n/a
800	n/a	sts, stdout, stderr = assert_python_ok('-E', '-c', code,
801	n/a	PYTHONASYNCIODEBUG='1')
802	n/a	self.assertEqual(stdout.rstrip(), b'False')
803	n/a
804	n/a	def test_create_task(self):
805	n/a	class MyTask(asyncio.Task):
806	n/a	pass
807	n/a
808	n/a	@asyncio.coroutine
809	n/a	def test():
810	n/a	pass
811	n/a
812	n/a	class EventLoop(base_events.BaseEventLoop):
813	n/a	def create_task(self, coro):
814	n/a	return MyTask(coro, loop=loop)
815	n/a
816	n/a	loop = EventLoop()
817	n/a	self.set_event_loop(loop)
818	n/a
819	n/a	coro = test()
820	n/a	task = asyncio.ensure_future(coro, loop=loop)
821	n/a	self.assertIsInstance(task, MyTask)
822	n/a
823	n/a	# make warnings quiet
824	n/a	task._log_destroy_pending = False
825	n/a	coro.close()
826	n/a
827	n/a	def test_run_forever_keyboard_interrupt(self):
828	n/a	# Python issue #22601: ensure that the temporary task created by
829	n/a	# run_forever() consumes the KeyboardInterrupt and so don't log
830	n/a	# a warning
831	n/a	@asyncio.coroutine
832	n/a	def raise_keyboard_interrupt():
833	n/a	raise KeyboardInterrupt
834	n/a
835	n/a	self.loop._process_events = mock.Mock()
836	n/a	self.loop.call_exception_handler = mock.Mock()
837	n/a
838	n/a	try:
839	n/a	self.loop.run_until_complete(raise_keyboard_interrupt())
840	n/a	except KeyboardInterrupt:
841	n/a	pass
842	n/a	self.loop.close()
843	n/a	support.gc_collect()
844	n/a
845	n/a	self.assertFalse(self.loop.call_exception_handler.called)
846	n/a
847	n/a	def test_run_until_complete_baseexception(self):
848	n/a	# Python issue #22429: run_until_complete() must not schedule a pending
849	n/a	# call to stop() if the future raised a BaseException
850	n/a	@asyncio.coroutine
851	n/a	def raise_keyboard_interrupt():
852	n/a	raise KeyboardInterrupt
853	n/a
854	n/a	self.loop._process_events = mock.Mock()
855	n/a
856	n/a	try:
857	n/a	self.loop.run_until_complete(raise_keyboard_interrupt())
858	n/a	except KeyboardInterrupt:
859	n/a	pass
860	n/a
861	n/a	def func():
862	n/a	self.loop.stop()
863	n/a	func.called = True
864	n/a	func.called = False
865	n/a	try:
866	n/a	self.loop.call_soon(func)
867	n/a	self.loop.run_forever()
868	n/a	except KeyboardInterrupt:
869	n/a	pass
870	n/a	self.assertTrue(func.called)
871	n/a
872	n/a	def test_single_selecter_event_callback_after_stopping(self):
873	n/a	# Python issue #25593: A stopped event loop may cause event callbacks
874	n/a	# to run more than once.
875	n/a	event_sentinel = object()
876	n/a	callcount = 0
877	n/a	doer = None
878	n/a
879	n/a	def proc_events(event_list):
880	n/a	nonlocal doer
881	n/a	if event_sentinel in event_list:
882	n/a	doer = self.loop.call_soon(do_event)
883	n/a
884	n/a	def do_event():
885	n/a	nonlocal callcount
886	n/a	callcount += 1
887	n/a	self.loop.call_soon(clear_selector)
888	n/a
889	n/a	def clear_selector():
890	n/a	doer.cancel()
891	n/a	self.loop._selector.select.return_value = ()
892	n/a
893	n/a	self.loop._process_events = proc_events
894	n/a	self.loop._selector.select.return_value = (event_sentinel,)
895	n/a
896	n/a	for i in range(1, 3):
897	n/a	with self.subTest('Loop %d/2' % i):
898	n/a	self.loop.call_soon(self.loop.stop)
899	n/a	self.loop.run_forever()
900	n/a	self.assertEqual(callcount, 1)
901	n/a
902	n/a	def test_run_once(self):
903	n/a	# Simple test for test_utils.run_once(). It may seem strange
904	n/a	# to have a test for this (the function isn't even used!) but
905	n/a	# it's a de-factor standard API for library tests. This tests
906	n/a	# the idiom: loop.call_soon(loop.stop); loop.run_forever().
907	n/a	count = 0
908	n/a
909	n/a	def callback():
910	n/a	nonlocal count
911	n/a	count += 1
912	n/a
913	n/a	self.loop._process_events = mock.Mock()
914	n/a	self.loop.call_soon(callback)
915	n/a	test_utils.run_once(self.loop)
916	n/a	self.assertEqual(count, 1)
917	n/a
918	n/a	def test_run_forever_pre_stopped(self):
919	n/a	# Test that the old idiom for pre-stopping the loop works.
920	n/a	self.loop._process_events = mock.Mock()
921	n/a	self.loop.stop()
922	n/a	self.loop.run_forever()
923	n/a	self.loop._selector.select.assert_called_once_with(0)
924	n/a
925	n/a
926	n/a	class MyProto(asyncio.Protocol):
927	n/a	done = None
928	n/a
929	n/a	def __init__(self, create_future=False):
930	n/a	self.state = 'INITIAL'
931	n/a	self.nbytes = 0
932	n/a	if create_future:
933	n/a	self.done = asyncio.Future()
934	n/a
935	n/a	def connection_made(self, transport):
936	n/a	self.transport = transport
937	n/a	assert self.state == 'INITIAL', self.state
938	n/a	self.state = 'CONNECTED'
939	n/a	transport.write(b'GET / HTTP/1.0\r\nHost: example.com\r\n\r\n')
940	n/a
941	n/a	def data_received(self, data):
942	n/a	assert self.state == 'CONNECTED', self.state
943	n/a	self.nbytes += len(data)
944	n/a
945	n/a	def eof_received(self):
946	n/a	assert self.state == 'CONNECTED', self.state
947	n/a	self.state = 'EOF'
948	n/a
949	n/a	def connection_lost(self, exc):
950	n/a	assert self.state in ('CONNECTED', 'EOF'), self.state
951	n/a	self.state = 'CLOSED'
952	n/a	if self.done:
953	n/a	self.done.set_result(None)
954	n/a
955	n/a
956	n/a	class MyDatagramProto(asyncio.DatagramProtocol):
957	n/a	done = None
958	n/a
959	n/a	def __init__(self, create_future=False, loop=None):
960	n/a	self.state = 'INITIAL'
961	n/a	self.nbytes = 0
962	n/a	if create_future:
963	n/a	self.done = asyncio.Future(loop=loop)
964	n/a
965	n/a	def connection_made(self, transport):
966	n/a	self.transport = transport
967	n/a	assert self.state == 'INITIAL', self.state
968	n/a	self.state = 'INITIALIZED'
969	n/a
970	n/a	def datagram_received(self, data, addr):
971	n/a	assert self.state == 'INITIALIZED', self.state
972	n/a	self.nbytes += len(data)
973	n/a
974	n/a	def error_received(self, exc):
975	n/a	assert self.state == 'INITIALIZED', self.state
976	n/a
977	n/a	def connection_lost(self, exc):
978	n/a	assert self.state == 'INITIALIZED', self.state
979	n/a	self.state = 'CLOSED'
980	n/a	if self.done:
981	n/a	self.done.set_result(None)
982	n/a
983	n/a
984	n/a	class BaseEventLoopWithSelectorTests(test_utils.TestCase):
985	n/a
986	n/a	def setUp(self):
987	n/a	super().setUp()
988	n/a	self.loop = asyncio.new_event_loop()
989	n/a	self.set_event_loop(self.loop)
990	n/a
991	n/a	@patch_socket
992	n/a	def test_create_connection_multiple_errors(self, m_socket):
993	n/a
994	n/a	class MyProto(asyncio.Protocol):
995	n/a	pass
996	n/a
997	n/a	@asyncio.coroutine
998	n/a	def getaddrinfo(args, *kw):
999	n/a	yield from []
1000	n/a	return [(2, 1, 6, '', ('107.6.106.82', 80)),
1001	n/a	(2, 1, 6, '', ('107.6.106.82', 80))]
1002	n/a
1003	n/a	def getaddrinfo_task(args, *kwds):
1004	n/a	return asyncio.Task(getaddrinfo(args, *kwds), loop=self.loop)
1005	n/a
1006	n/a	idx = -1
1007	n/a	errors = ['err1', 'err2']
1008	n/a
1009	n/a	def _socket(args, *kw):
1010	n/a	nonlocal idx, errors
1011	n/a	idx += 1
1012	n/a	raise OSError(errors[idx])
1013	n/a
1014	n/a	m_socket.socket = _socket
1015	n/a
1016	n/a	self.loop.getaddrinfo = getaddrinfo_task
1017	n/a
1018	n/a	coro = self.loop.create_connection(MyProto, 'example.com', 80)
1019	n/a	with self.assertRaises(OSError) as cm:
1020	n/a	self.loop.run_until_complete(coro)
1021	n/a
1022	n/a	self.assertEqual(str(cm.exception), 'Multiple exceptions: err1, err2')
1023	n/a
1024	n/a	@patch_socket
1025	n/a	def test_create_connection_timeout(self, m_socket):
1026	n/a	# Ensure that the socket is closed on timeout
1027	n/a	sock = mock.Mock()
1028	n/a	m_socket.socket.return_value = sock
1029	n/a
1030	n/a	def getaddrinfo(args, *kw):
1031	n/a	fut = asyncio.Future(loop=self.loop)
1032	n/a	addr = (socket.AF_INET, socket.SOCK_STREAM, 0, '',
1033	n/a	('127.0.0.1', 80))
1034	n/a	fut.set_result([addr])
1035	n/a	return fut
1036	n/a	self.loop.getaddrinfo = getaddrinfo
1037	n/a
1038	n/a	with mock.patch.object(self.loop, 'sock_connect',
1039	n/a	side_effect=asyncio.TimeoutError):
1040	n/a	coro = self.loop.create_connection(MyProto, '127.0.0.1', 80)
1041	n/a	with self.assertRaises(asyncio.TimeoutError):
1042	n/a	self.loop.run_until_complete(coro)
1043	n/a	self.assertTrue(sock.close.called)
1044	n/a
1045	n/a	def test_create_connection_host_port_sock(self):
1046	n/a	coro = self.loop.create_connection(
1047	n/a	MyProto, 'example.com', 80, sock=object())
1048	n/a	self.assertRaises(ValueError, self.loop.run_until_complete, coro)
1049	n/a
1050	n/a	def test_create_connection_wrong_sock(self):
1051	n/a	sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
1052	n/a	with sock:
1053	n/a	coro = self.loop.create_connection(MyProto, sock=sock)
1054	n/a	with self.assertRaisesRegex(ValueError,
1055	n/a	'A Stream Socket was expected'):
1056	n/a	self.loop.run_until_complete(coro)
1057	n/a
1058	n/a	def test_create_server_wrong_sock(self):
1059	n/a	sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
1060	n/a	with sock:
1061	n/a	coro = self.loop.create_server(MyProto, sock=sock)
1062	n/a	with self.assertRaisesRegex(ValueError,
1063	n/a	'A Stream Socket was expected'):
1064	n/a	self.loop.run_until_complete(coro)
1065	n/a
1066	n/a	@unittest.skipUnless(hasattr(socket, 'SOCK_NONBLOCK'),
1067	n/a	'no socket.SOCK_NONBLOCK (linux only)')
1068	n/a	def test_create_server_stream_bittype(self):
1069	n/a	sock = socket.socket(
1070	n/a	socket.AF_INET, socket.SOCK_STREAM \| socket.SOCK_NONBLOCK)
1071	n/a	with sock:
1072	n/a	coro = self.loop.create_server(lambda: None, sock=sock)
1073	n/a	srv = self.loop.run_until_complete(coro)
1074	n/a	srv.close()
1075	n/a	self.loop.run_until_complete(srv.wait_closed())
1076	n/a
1077	n/a	def test_create_datagram_endpoint_wrong_sock(self):
1078	n/a	sock = socket.socket(socket.AF_INET)
1079	n/a	with sock:
1080	n/a	coro = self.loop.create_datagram_endpoint(MyProto, sock=sock)
1081	n/a	with self.assertRaisesRegex(ValueError,
1082	n/a	'A UDP Socket was expected'):
1083	n/a	self.loop.run_until_complete(coro)
1084	n/a
1085	n/a	def test_create_connection_no_host_port_sock(self):
1086	n/a	coro = self.loop.create_connection(MyProto)
1087	n/a	self.assertRaises(ValueError, self.loop.run_until_complete, coro)
1088	n/a
1089	n/a	def test_create_connection_no_getaddrinfo(self):
1090	n/a	@asyncio.coroutine
1091	n/a	def getaddrinfo(args, *kw):
1092	n/a	yield from []
1093	n/a
1094	n/a	def getaddrinfo_task(args, *kwds):
1095	n/a	return asyncio.Task(getaddrinfo(args, *kwds), loop=self.loop)
1096	n/a
1097	n/a	self.loop.getaddrinfo = getaddrinfo_task
1098	n/a	coro = self.loop.create_connection(MyProto, 'example.com', 80)
1099	n/a	self.assertRaises(
1100	n/a	OSError, self.loop.run_until_complete, coro)
1101	n/a
1102	n/a	def test_create_connection_connect_err(self):
1103	n/a	@asyncio.coroutine
1104	n/a	def getaddrinfo(args, *kw):
1105	n/a	yield from []
1106	n/a	return [(2, 1, 6, '', ('107.6.106.82', 80))]
1107	n/a
1108	n/a	def getaddrinfo_task(args, *kwds):
1109	n/a	return asyncio.Task(getaddrinfo(args, *kwds), loop=self.loop)
1110	n/a
1111	n/a	self.loop.getaddrinfo = getaddrinfo_task
1112	n/a	self.loop.sock_connect = mock.Mock()
1113	n/a	self.loop.sock_connect.side_effect = OSError
1114	n/a
1115	n/a	coro = self.loop.create_connection(MyProto, 'example.com', 80)
1116	n/a	self.assertRaises(
1117	n/a	OSError, self.loop.run_until_complete, coro)
1118	n/a
1119	n/a	def test_create_connection_multiple(self):
1120	n/a	@asyncio.coroutine
1121	n/a	def getaddrinfo(args, *kw):
1122	n/a	return [(2, 1, 6, '', ('0.0.0.1', 80)),
1123	n/a	(2, 1, 6, '', ('0.0.0.2', 80))]
1124	n/a
1125	n/a	def getaddrinfo_task(args, *kwds):
1126	n/a	return asyncio.Task(getaddrinfo(args, *kwds), loop=self.loop)
1127	n/a
1128	n/a	self.loop.getaddrinfo = getaddrinfo_task
1129	n/a	self.loop.sock_connect = mock.Mock()
1130	n/a	self.loop.sock_connect.side_effect = OSError
1131	n/a
1132	n/a	coro = self.loop.create_connection(
1133	n/a	MyProto, 'example.com', 80, family=socket.AF_INET)
1134	n/a	with self.assertRaises(OSError):
1135	n/a	self.loop.run_until_complete(coro)
1136	n/a
1137	n/a	@patch_socket
1138	n/a	def test_create_connection_multiple_errors_local_addr(self, m_socket):
1139	n/a
1140	n/a	def bind(addr):
1141	n/a	if addr[0] == '0.0.0.1':
1142	n/a	err = OSError('Err')
1143	n/a	err.strerror = 'Err'
1144	n/a	raise err
1145	n/a
1146	n/a	m_socket.socket.return_value.bind = bind
1147	n/a
1148	n/a	@asyncio.coroutine
1149	n/a	def getaddrinfo(args, *kw):
1150	n/a	return [(2, 1, 6, '', ('0.0.0.1', 80)),
1151	n/a	(2, 1, 6, '', ('0.0.0.2', 80))]
1152	n/a
1153	n/a	def getaddrinfo_task(args, *kwds):
1154	n/a	return asyncio.Task(getaddrinfo(args, *kwds), loop=self.loop)
1155	n/a
1156	n/a	self.loop.getaddrinfo = getaddrinfo_task
1157	n/a	self.loop.sock_connect = mock.Mock()
1158	n/a	self.loop.sock_connect.side_effect = OSError('Err2')
1159	n/a
1160	n/a	coro = self.loop.create_connection(
1161	n/a	MyProto, 'example.com', 80, family=socket.AF_INET,
1162	n/a	local_addr=(None, 8080))
1163	n/a	with self.assertRaises(OSError) as cm:
1164	n/a	self.loop.run_until_complete(coro)
1165	n/a
1166	n/a	self.assertTrue(str(cm.exception).startswith('Multiple exceptions: '))
1167	n/a	self.assertTrue(m_socket.socket.return_value.close.called)
1168	n/a
1169	n/a	def _test_create_connection_ip_addr(self, m_socket, allow_inet_pton):
1170	n/a	# Test the fallback code, even if this system has inet_pton.
1171	n/a	if not allow_inet_pton:
1172	n/a	del m_socket.inet_pton
1173	n/a
1174	n/a	m_socket.getaddrinfo = socket.getaddrinfo
1175	n/a	sock = m_socket.socket.return_value
1176	n/a
1177	n/a	self.loop._add_reader = mock.Mock()
1178	n/a	self.loop._add_reader._is_coroutine = False
1179	n/a	self.loop._add_writer = mock.Mock()
1180	n/a	self.loop._add_writer._is_coroutine = False
1181	n/a
1182	n/a	coro = self.loop.create_connection(asyncio.Protocol, '1.2.3.4', 80)
1183	n/a	t, p = self.loop.run_until_complete(coro)
1184	n/a	try:
1185	n/a	sock.connect.assert_called_with(('1.2.3.4', 80))
1186	n/a	_, kwargs = m_socket.socket.call_args
1187	n/a	self.assertEqual(kwargs['family'], m_socket.AF_INET)
1188	n/a	self.assertEqual(kwargs['type'], m_socket.SOCK_STREAM)
1189	n/a	finally:
1190	n/a	t.close()
1191	n/a	test_utils.run_briefly(self.loop) # allow transport to close
1192	n/a
1193	n/a	sock.family = socket.AF_INET6
1194	n/a	coro = self.loop.create_connection(asyncio.Protocol, '::1', 80)
1195	n/a	t, p = self.loop.run_until_complete(coro)
1196	n/a	try:
1197	n/a	# Without inet_pton we use getaddrinfo, which transforms ('::1', 80)
1198	n/a	# to ('::1', 80, 0, 0). The last 0s are flow info, scope id.
1199	n/a	[address] = sock.connect.call_args[0]
1200	n/a	host, port = address[:2]
1201	n/a	self.assertRegex(host, r'::(0\.)*1')
1202	n/a	self.assertEqual(port, 80)
1203	n/a	_, kwargs = m_socket.socket.call_args
1204	n/a	self.assertEqual(kwargs['family'], m_socket.AF_INET6)
1205	n/a	self.assertEqual(kwargs['type'], m_socket.SOCK_STREAM)
1206	n/a	finally:
1207	n/a	t.close()
1208	n/a	test_utils.run_briefly(self.loop) # allow transport to close
1209	n/a
1210	n/a	@patch_socket
1211	n/a	def test_create_connection_ip_addr(self, m_socket):
1212	n/a	self._test_create_connection_ip_addr(m_socket, True)
1213	n/a
1214	n/a	@patch_socket
1215	n/a	def test_create_connection_no_inet_pton(self, m_socket):
1216	n/a	self._test_create_connection_ip_addr(m_socket, False)
1217	n/a
1218	n/a	@patch_socket
1219	n/a	def test_create_connection_service_name(self, m_socket):
1220	n/a	m_socket.getaddrinfo = socket.getaddrinfo
1221	n/a	sock = m_socket.socket.return_value
1222	n/a
1223	n/a	self.loop._add_reader = mock.Mock()
1224	n/a	self.loop._add_reader._is_coroutine = False
1225	n/a	self.loop._add_writer = mock.Mock()
1226	n/a	self.loop._add_writer._is_coroutine = False
1227	n/a
1228	n/a	for service, port in ('http', 80), (b'http', 80):
1229	n/a	coro = self.loop.create_connection(asyncio.Protocol,
1230	n/a	'127.0.0.1', service)
1231	n/a
1232	n/a	t, p = self.loop.run_until_complete(coro)
1233	n/a	try:
1234	n/a	sock.connect.assert_called_with(('127.0.0.1', port))
1235	n/a	_, kwargs = m_socket.socket.call_args
1236	n/a	self.assertEqual(kwargs['family'], m_socket.AF_INET)
1237	n/a	self.assertEqual(kwargs['type'], m_socket.SOCK_STREAM)
1238	n/a	finally:
1239	n/a	t.close()
1240	n/a	test_utils.run_briefly(self.loop) # allow transport to close
1241	n/a
1242	n/a	for service in 'nonsense', b'nonsense':
1243	n/a	coro = self.loop.create_connection(asyncio.Protocol,
1244	n/a	'127.0.0.1', service)
1245	n/a
1246	n/a	with self.assertRaises(OSError):
1247	n/a	self.loop.run_until_complete(coro)
1248	n/a
1249	n/a	def test_create_connection_no_local_addr(self):
1250	n/a	@asyncio.coroutine
1251	n/a	def getaddrinfo(host, args, *kw):
1252	n/a	if host == 'example.com':
1253	n/a	return [(2, 1, 6, '', ('107.6.106.82', 80)),
1254	n/a	(2, 1, 6, '', ('107.6.106.82', 80))]
1255	n/a	else:
1256	n/a	return []
1257	n/a
1258	n/a	def getaddrinfo_task(args, *kwds):
1259	n/a	return asyncio.Task(getaddrinfo(args, *kwds), loop=self.loop)
1260	n/a	self.loop.getaddrinfo = getaddrinfo_task
1261	n/a
1262	n/a	coro = self.loop.create_connection(
1263	n/a	MyProto, 'example.com', 80, family=socket.AF_INET,
1264	n/a	local_addr=(None, 8080))
1265	n/a	self.assertRaises(
1266	n/a	OSError, self.loop.run_until_complete, coro)
1267	n/a
1268	n/a	@patch_socket
1269	n/a	def test_create_connection_bluetooth(self, m_socket):
1270	n/a	# See http://bugs.python.org/issue27136, fallback to getaddrinfo when
1271	n/a	# we can't recognize an address is resolved, e.g. a Bluetooth address.
1272	n/a	addr = ('00:01:02:03:04:05', 1)
1273	n/a
1274	n/a	def getaddrinfo(host, port, args, *kw):
1275	n/a	assert (host, port) == addr
1276	n/a	return [(999, 1, 999, '', (addr, 1))]
1277	n/a
1278	n/a	m_socket.getaddrinfo = getaddrinfo
1279	n/a	sock = m_socket.socket()
1280	n/a	coro = self.loop.sock_connect(sock, addr)
1281	n/a	self.loop.run_until_complete(coro)
1282	n/a
1283	n/a	def test_create_connection_ssl_server_hostname_default(self):
1284	n/a	self.loop.getaddrinfo = mock.Mock()
1285	n/a
1286	n/a	def mock_getaddrinfo(args, *kwds):
1287	n/a	f = asyncio.Future(loop=self.loop)
1288	n/a	f.set_result([(socket.AF_INET, socket.SOCK_STREAM,
1289	n/a	socket.SOL_TCP, '', ('1.2.3.4', 80))])
1290	n/a	return f
1291	n/a
1292	n/a	self.loop.getaddrinfo.side_effect = mock_getaddrinfo
1293	n/a	self.loop.sock_connect = mock.Mock()
1294	n/a	self.loop.sock_connect.return_value = ()
1295	n/a	self.loop._make_ssl_transport = mock.Mock()
1296	n/a
1297	n/a	class _SelectorTransportMock:
1298	n/a	_sock = None
1299	n/a
1300	n/a	def get_extra_info(self, key):
1301	n/a	return mock.Mock()
1302	n/a
1303	n/a	def close(self):
1304	n/a	self._sock.close()
1305	n/a
1306	n/a	def mock_make_ssl_transport(sock, protocol, sslcontext, waiter,
1307	n/a	**kwds):
1308	n/a	waiter.set_result(None)
1309	n/a	transport = _SelectorTransportMock()
1310	n/a	transport._sock = sock
1311	n/a	return transport
1312	n/a
1313	n/a	self.loop._make_ssl_transport.side_effect = mock_make_ssl_transport
1314	n/a	ANY = mock.ANY
1315	n/a	# First try the default server_hostname.
1316	n/a	self.loop._make_ssl_transport.reset_mock()
1317	n/a	coro = self.loop.create_connection(MyProto, 'python.org', 80, ssl=True)
1318	n/a	transport, _ = self.loop.run_until_complete(coro)
1319	n/a	transport.close()
1320	n/a	self.loop._make_ssl_transport.assert_called_with(
1321	n/a	ANY, ANY, ANY, ANY,
1322	n/a	server_side=False,
1323	n/a	server_hostname='python.org')
1324	n/a	# Next try an explicit server_hostname.
1325	n/a	self.loop._make_ssl_transport.reset_mock()
1326	n/a	coro = self.loop.create_connection(MyProto, 'python.org', 80, ssl=True,
1327	n/a	server_hostname='perl.com')
1328	n/a	transport, _ = self.loop.run_until_complete(coro)
1329	n/a	transport.close()
1330	n/a	self.loop._make_ssl_transport.assert_called_with(
1331	n/a	ANY, ANY, ANY, ANY,
1332	n/a	server_side=False,
1333	n/a	server_hostname='perl.com')
1334	n/a	# Finally try an explicit empty server_hostname.
1335	n/a	self.loop._make_ssl_transport.reset_mock()
1336	n/a	coro = self.loop.create_connection(MyProto, 'python.org', 80, ssl=True,
1337	n/a	server_hostname='')
1338	n/a	transport, _ = self.loop.run_until_complete(coro)
1339	n/a	transport.close()
1340	n/a	self.loop._make_ssl_transport.assert_called_with(ANY, ANY, ANY, ANY,
1341	n/a	server_side=False,
1342	n/a	server_hostname='')
1343	n/a
1344	n/a	def test_create_connection_no_ssl_server_hostname_errors(self):
1345	n/a	# When not using ssl, server_hostname must be None.
1346	n/a	coro = self.loop.create_connection(MyProto, 'python.org', 80,
1347	n/a	server_hostname='')
1348	n/a	self.assertRaises(ValueError, self.loop.run_until_complete, coro)
1349	n/a	coro = self.loop.create_connection(MyProto, 'python.org', 80,
1350	n/a	server_hostname='python.org')
1351	n/a	self.assertRaises(ValueError, self.loop.run_until_complete, coro)
1352	n/a
1353	n/a	def test_create_connection_ssl_server_hostname_errors(self):
1354	n/a	# When using ssl, server_hostname may be None if host is non-empty.
1355	n/a	coro = self.loop.create_connection(MyProto, '', 80, ssl=True)
1356	n/a	self.assertRaises(ValueError, self.loop.run_until_complete, coro)
1357	n/a	coro = self.loop.create_connection(MyProto, None, 80, ssl=True)
1358	n/a	self.assertRaises(ValueError, self.loop.run_until_complete, coro)
1359	n/a	sock = socket.socket()
1360	n/a	coro = self.loop.create_connection(MyProto, None, None,
1361	n/a	ssl=True, sock=sock)
1362	n/a	self.addCleanup(sock.close)
1363	n/a	self.assertRaises(ValueError, self.loop.run_until_complete, coro)
1364	n/a
1365	n/a	def test_create_server_empty_host(self):
1366	n/a	# if host is empty string use None instead
1367	n/a	host = object()
1368	n/a
1369	n/a	@asyncio.coroutine
1370	n/a	def getaddrinfo(args, *kw):
1371	n/a	nonlocal host
1372	n/a	host = args[0]
1373	n/a	yield from []
1374	n/a
1375	n/a	def getaddrinfo_task(args, *kwds):
1376	n/a	return asyncio.Task(getaddrinfo(args, *kwds), loop=self.loop)
1377	n/a
1378	n/a	self.loop.getaddrinfo = getaddrinfo_task
1379	n/a	fut = self.loop.create_server(MyProto, '', 0)
1380	n/a	self.assertRaises(OSError, self.loop.run_until_complete, fut)
1381	n/a	self.assertIsNone(host)
1382	n/a
1383	n/a	def test_create_server_host_port_sock(self):
1384	n/a	fut = self.loop.create_server(
1385	n/a	MyProto, '0.0.0.0', 0, sock=object())
1386	n/a	self.assertRaises(ValueError, self.loop.run_until_complete, fut)
1387	n/a
1388	n/a	def test_create_server_no_host_port_sock(self):
1389	n/a	fut = self.loop.create_server(MyProto)
1390	n/a	self.assertRaises(ValueError, self.loop.run_until_complete, fut)
1391	n/a
1392	n/a	def test_create_server_no_getaddrinfo(self):
1393	n/a	getaddrinfo = self.loop.getaddrinfo = mock.Mock()
1394	n/a	getaddrinfo.return_value = []
1395	n/a
1396	n/a	f = self.loop.create_server(MyProto, 'python.org', 0)
1397	n/a	self.assertRaises(OSError, self.loop.run_until_complete, f)
1398	n/a
1399	n/a	@patch_socket
1400	n/a	def test_create_server_nosoreuseport(self, m_socket):
1401	n/a	m_socket.getaddrinfo = socket.getaddrinfo
1402	n/a	del m_socket.SO_REUSEPORT
1403	n/a	m_socket.socket.return_value = mock.Mock()
1404	n/a
1405	n/a	f = self.loop.create_server(
1406	n/a	MyProto, '0.0.0.0', 0, reuse_port=True)
1407	n/a
1408	n/a	self.assertRaises(ValueError, self.loop.run_until_complete, f)
1409	n/a
1410	n/a	@patch_socket
1411	n/a	def test_create_server_soreuseport_only_defined(self, m_socket):
1412	n/a	m_socket.getaddrinfo = socket.getaddrinfo
1413	n/a	m_socket.socket.return_value = mock.Mock()
1414	n/a	m_socket.SO_REUSEPORT = -1
1415	n/a
1416	n/a	f = self.loop.create_server(
1417	n/a	MyProto, '0.0.0.0', 0, reuse_port=True)
1418	n/a
1419	n/a	self.assertRaises(ValueError, self.loop.run_until_complete, f)
1420	n/a
1421	n/a	@patch_socket
1422	n/a	def test_create_server_cant_bind(self, m_socket):
1423	n/a
1424	n/a	class Err(OSError):
1425	n/a	strerror = 'error'
1426	n/a
1427	n/a	m_socket.getaddrinfo.return_value = [
1428	n/a	(2, 1, 6, '', ('127.0.0.1', 10100))]
1429	n/a	m_socket.getaddrinfo._is_coroutine = False
1430	n/a	m_sock = m_socket.socket.return_value = mock.Mock()
1431	n/a	m_sock.bind.side_effect = Err
1432	n/a
1433	n/a	fut = self.loop.create_server(MyProto, '0.0.0.0', 0)
1434	n/a	self.assertRaises(OSError, self.loop.run_until_complete, fut)
1435	n/a	self.assertTrue(m_sock.close.called)
1436	n/a
1437	n/a	@patch_socket
1438	n/a	def test_create_datagram_endpoint_no_addrinfo(self, m_socket):
1439	n/a	m_socket.getaddrinfo.return_value = []
1440	n/a	m_socket.getaddrinfo._is_coroutine = False
1441	n/a
1442	n/a	coro = self.loop.create_datagram_endpoint(
1443	n/a	MyDatagramProto, local_addr=('localhost', 0))
1444	n/a	self.assertRaises(
1445	n/a	OSError, self.loop.run_until_complete, coro)
1446	n/a
1447	n/a	def test_create_datagram_endpoint_addr_error(self):
1448	n/a	coro = self.loop.create_datagram_endpoint(
1449	n/a	MyDatagramProto, local_addr='localhost')
1450	n/a	self.assertRaises(
1451	n/a	AssertionError, self.loop.run_until_complete, coro)
1452	n/a	coro = self.loop.create_datagram_endpoint(
1453	n/a	MyDatagramProto, local_addr=('localhost', 1, 2, 3))
1454	n/a	self.assertRaises(
1455	n/a	AssertionError, self.loop.run_until_complete, coro)
1456	n/a
1457	n/a	def test_create_datagram_endpoint_connect_err(self):
1458	n/a	self.loop.sock_connect = mock.Mock()
1459	n/a	self.loop.sock_connect.side_effect = OSError
1460	n/a
1461	n/a	coro = self.loop.create_datagram_endpoint(
1462	n/a	asyncio.DatagramProtocol, remote_addr=('127.0.0.1', 0))
1463	n/a	self.assertRaises(
1464	n/a	OSError, self.loop.run_until_complete, coro)
1465	n/a
1466	n/a	@patch_socket
1467	n/a	def test_create_datagram_endpoint_socket_err(self, m_socket):
1468	n/a	m_socket.getaddrinfo = socket.getaddrinfo
1469	n/a	m_socket.socket.side_effect = OSError
1470	n/a
1471	n/a	coro = self.loop.create_datagram_endpoint(
1472	n/a	asyncio.DatagramProtocol, family=socket.AF_INET)
1473	n/a	self.assertRaises(
1474	n/a	OSError, self.loop.run_until_complete, coro)
1475	n/a
1476	n/a	coro = self.loop.create_datagram_endpoint(
1477	n/a	asyncio.DatagramProtocol, local_addr=('127.0.0.1', 0))
1478	n/a	self.assertRaises(
1479	n/a	OSError, self.loop.run_until_complete, coro)
1480	n/a
1481	n/a	@unittest.skipUnless(support.IPV6_ENABLED, 'IPv6 not supported or enabled')
1482	n/a	def test_create_datagram_endpoint_no_matching_family(self):
1483	n/a	coro = self.loop.create_datagram_endpoint(
1484	n/a	asyncio.DatagramProtocol,
1485	n/a	remote_addr=('127.0.0.1', 0), local_addr=('::1', 0))
1486	n/a	self.assertRaises(
1487	n/a	ValueError, self.loop.run_until_complete, coro)
1488	n/a
1489	n/a	@patch_socket
1490	n/a	def test_create_datagram_endpoint_setblk_err(self, m_socket):
1491	n/a	m_socket.socket.return_value.setblocking.side_effect = OSError
1492	n/a
1493	n/a	coro = self.loop.create_datagram_endpoint(
1494	n/a	asyncio.DatagramProtocol, family=socket.AF_INET)
1495	n/a	self.assertRaises(
1496	n/a	OSError, self.loop.run_until_complete, coro)
1497	n/a	self.assertTrue(
1498	n/a	m_socket.socket.return_value.close.called)
1499	n/a
1500	n/a	def test_create_datagram_endpoint_noaddr_nofamily(self):
1501	n/a	coro = self.loop.create_datagram_endpoint(
1502	n/a	asyncio.DatagramProtocol)
1503	n/a	self.assertRaises(ValueError, self.loop.run_until_complete, coro)
1504	n/a
1505	n/a	@patch_socket
1506	n/a	def test_create_datagram_endpoint_cant_bind(self, m_socket):
1507	n/a	class Err(OSError):
1508	n/a	pass
1509	n/a
1510	n/a	m_socket.getaddrinfo = socket.getaddrinfo
1511	n/a	m_sock = m_socket.socket.return_value = mock.Mock()
1512	n/a	m_sock.bind.side_effect = Err
1513	n/a
1514	n/a	fut = self.loop.create_datagram_endpoint(
1515	n/a	MyDatagramProto,
1516	n/a	local_addr=('127.0.0.1', 0), family=socket.AF_INET)
1517	n/a	self.assertRaises(Err, self.loop.run_until_complete, fut)
1518	n/a	self.assertTrue(m_sock.close.called)
1519	n/a
1520	n/a	def test_create_datagram_endpoint_sock(self):
1521	n/a	sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
1522	n/a	sock.bind(('127.0.0.1', 0))
1523	n/a	fut = self.loop.create_datagram_endpoint(
1524	n/a	lambda: MyDatagramProto(create_future=True, loop=self.loop),
1525	n/a	sock=sock)
1526	n/a	transport, protocol = self.loop.run_until_complete(fut)
1527	n/a	transport.close()
1528	n/a	self.loop.run_until_complete(protocol.done)
1529	n/a	self.assertEqual('CLOSED', protocol.state)
1530	n/a
1531	n/a	def test_create_datagram_endpoint_sock_sockopts(self):
1532	n/a	class FakeSock:
1533	n/a	type = socket.SOCK_DGRAM
1534	n/a
1535	n/a	fut = self.loop.create_datagram_endpoint(
1536	n/a	MyDatagramProto, local_addr=('127.0.0.1', 0), sock=FakeSock())
1537	n/a	self.assertRaises(ValueError, self.loop.run_until_complete, fut)
1538	n/a
1539	n/a	fut = self.loop.create_datagram_endpoint(
1540	n/a	MyDatagramProto, remote_addr=('127.0.0.1', 0), sock=FakeSock())
1541	n/a	self.assertRaises(ValueError, self.loop.run_until_complete, fut)
1542	n/a
1543	n/a	fut = self.loop.create_datagram_endpoint(
1544	n/a	MyDatagramProto, family=1, sock=FakeSock())
1545	n/a	self.assertRaises(ValueError, self.loop.run_until_complete, fut)
1546	n/a
1547	n/a	fut = self.loop.create_datagram_endpoint(
1548	n/a	MyDatagramProto, proto=1, sock=FakeSock())
1549	n/a	self.assertRaises(ValueError, self.loop.run_until_complete, fut)
1550	n/a
1551	n/a	fut = self.loop.create_datagram_endpoint(
1552	n/a	MyDatagramProto, flags=1, sock=FakeSock())
1553	n/a	self.assertRaises(ValueError, self.loop.run_until_complete, fut)
1554	n/a
1555	n/a	fut = self.loop.create_datagram_endpoint(
1556	n/a	MyDatagramProto, reuse_address=True, sock=FakeSock())
1557	n/a	self.assertRaises(ValueError, self.loop.run_until_complete, fut)
1558	n/a
1559	n/a	fut = self.loop.create_datagram_endpoint(
1560	n/a	MyDatagramProto, reuse_port=True, sock=FakeSock())
1561	n/a	self.assertRaises(ValueError, self.loop.run_until_complete, fut)
1562	n/a
1563	n/a	fut = self.loop.create_datagram_endpoint(
1564	n/a	MyDatagramProto, allow_broadcast=True, sock=FakeSock())
1565	n/a	self.assertRaises(ValueError, self.loop.run_until_complete, fut)
1566	n/a
1567	n/a	def test_create_datagram_endpoint_sockopts(self):
1568	n/a	# Socket options should not be applied unless asked for.
1569	n/a	# SO_REUSEADDR defaults to on for UNIX.
1570	n/a	# SO_REUSEPORT is not available on all platforms.
1571	n/a
1572	n/a	coro = self.loop.create_datagram_endpoint(
1573	n/a	lambda: MyDatagramProto(create_future=True, loop=self.loop),
1574	n/a	local_addr=('127.0.0.1', 0))
1575	n/a	transport, protocol = self.loop.run_until_complete(coro)
1576	n/a	sock = transport.get_extra_info('socket')
1577	n/a
1578	n/a	reuse_address_default_on = (
1579	n/a	os.name == 'posix' and sys.platform != 'cygwin')
1580	n/a	reuseport_supported = hasattr(socket, 'SO_REUSEPORT')
1581	n/a
1582	n/a	if reuse_address_default_on:
1583	n/a	self.assertTrue(
1584	n/a	sock.getsockopt(
1585	n/a	socket.SOL_SOCKET, socket.SO_REUSEADDR))
1586	n/a	else:
1587	n/a	self.assertFalse(
1588	n/a	sock.getsockopt(
1589	n/a	socket.SOL_SOCKET, socket.SO_REUSEADDR))
1590	n/a	if reuseport_supported:
1591	n/a	self.assertFalse(
1592	n/a	sock.getsockopt(
1593	n/a	socket.SOL_SOCKET, socket.SO_REUSEPORT))
1594	n/a	self.assertFalse(
1595	n/a	sock.getsockopt(
1596	n/a	socket.SOL_SOCKET, socket.SO_BROADCAST))
1597	n/a
1598	n/a	transport.close()
1599	n/a	self.loop.run_until_complete(protocol.done)
1600	n/a	self.assertEqual('CLOSED', protocol.state)
1601	n/a
1602	n/a	coro = self.loop.create_datagram_endpoint(
1603	n/a	lambda: MyDatagramProto(create_future=True, loop=self.loop),
1604	n/a	local_addr=('127.0.0.1', 0),
1605	n/a	reuse_address=True,
1606	n/a	reuse_port=reuseport_supported,
1607	n/a	allow_broadcast=True)
1608	n/a	transport, protocol = self.loop.run_until_complete(coro)
1609	n/a	sock = transport.get_extra_info('socket')
1610	n/a
1611	n/a	self.assertTrue(
1612	n/a	sock.getsockopt(
1613	n/a	socket.SOL_SOCKET, socket.SO_REUSEADDR))
1614	n/a	if reuseport_supported:
1615	n/a	self.assertTrue(
1616	n/a	sock.getsockopt(
1617	n/a	socket.SOL_SOCKET, socket.SO_REUSEPORT))
1618	n/a	self.assertTrue(
1619	n/a	sock.getsockopt(
1620	n/a	socket.SOL_SOCKET, socket.SO_BROADCAST))
1621	n/a
1622	n/a	transport.close()
1623	n/a	self.loop.run_until_complete(protocol.done)
1624	n/a	self.assertEqual('CLOSED', protocol.state)
1625	n/a
1626	n/a	@patch_socket
1627	n/a	def test_create_datagram_endpoint_nosoreuseport(self, m_socket):
1628	n/a	del m_socket.SO_REUSEPORT
1629	n/a	m_socket.socket.return_value = mock.Mock()
1630	n/a
1631	n/a	coro = self.loop.create_datagram_endpoint(
1632	n/a	lambda: MyDatagramProto(loop=self.loop),
1633	n/a	local_addr=('127.0.0.1', 0),
1634	n/a	reuse_address=False,
1635	n/a	reuse_port=True)
1636	n/a
1637	n/a	self.assertRaises(ValueError, self.loop.run_until_complete, coro)
1638	n/a
1639	n/a	@patch_socket
1640	n/a	def test_create_datagram_endpoint_ip_addr(self, m_socket):
1641	n/a	def getaddrinfo(args, *kw):
1642	n/a	self.fail('should not have called getaddrinfo')
1643	n/a
1644	n/a	m_socket.getaddrinfo = getaddrinfo
1645	n/a	m_socket.socket.return_value.bind = bind = mock.Mock()
1646	n/a	self.loop._add_reader = mock.Mock()
1647	n/a	self.loop._add_reader._is_coroutine = False
1648	n/a
1649	n/a	reuseport_supported = hasattr(socket, 'SO_REUSEPORT')
1650	n/a	coro = self.loop.create_datagram_endpoint(
1651	n/a	lambda: MyDatagramProto(loop=self.loop),
1652	n/a	local_addr=('1.2.3.4', 0),
1653	n/a	reuse_address=False,
1654	n/a	reuse_port=reuseport_supported)
1655	n/a
1656	n/a	t, p = self.loop.run_until_complete(coro)
1657	n/a	try:
1658	n/a	bind.assert_called_with(('1.2.3.4', 0))
1659	n/a	m_socket.socket.assert_called_with(family=m_socket.AF_INET,
1660	n/a	proto=m_socket.IPPROTO_UDP,
1661	n/a	type=m_socket.SOCK_DGRAM)
1662	n/a	finally:
1663	n/a	t.close()
1664	n/a	test_utils.run_briefly(self.loop) # allow transport to close
1665	n/a
1666	n/a	def test_accept_connection_retry(self):
1667	n/a	sock = mock.Mock()
1668	n/a	sock.accept.side_effect = BlockingIOError()
1669	n/a
1670	n/a	self.loop._accept_connection(MyProto, sock)
1671	n/a	self.assertFalse(sock.close.called)
1672	n/a
1673	n/a	@mock.patch('asyncio.base_events.logger')
1674	n/a	def test_accept_connection_exception(self, m_log):
1675	n/a	sock = mock.Mock()
1676	n/a	sock.fileno.return_value = 10
1677	n/a	sock.accept.side_effect = OSError(errno.EMFILE, 'Too many open files')
1678	n/a	self.loop._remove_reader = mock.Mock()
1679	n/a	self.loop.call_later = mock.Mock()
1680	n/a
1681	n/a	self.loop._accept_connection(MyProto, sock)
1682	n/a	self.assertTrue(m_log.error.called)
1683	n/a	self.assertFalse(sock.close.called)
1684	n/a	self.loop._remove_reader.assert_called_with(10)
1685	n/a	self.loop.call_later.assert_called_with(constants.ACCEPT_RETRY_DELAY,
1686	n/a	# self.loop._start_serving
1687	n/a	mock.ANY,
1688	n/a	MyProto, sock, None, None, mock.ANY)
1689	n/a
1690	n/a	def test_call_coroutine(self):
1691	n/a	@asyncio.coroutine
1692	n/a	def simple_coroutine():
1693	n/a	pass
1694	n/a
1695	n/a	self.loop.set_debug(True)
1696	n/a	coro_func = simple_coroutine
1697	n/a	coro_obj = coro_func()
1698	n/a	self.addCleanup(coro_obj.close)
1699	n/a	for func in (coro_func, coro_obj):
1700	n/a	with self.assertRaises(TypeError):
1701	n/a	self.loop.call_soon(func)
1702	n/a	with self.assertRaises(TypeError):
1703	n/a	self.loop.call_soon_threadsafe(func)
1704	n/a	with self.assertRaises(TypeError):
1705	n/a	self.loop.call_later(60, func)
1706	n/a	with self.assertRaises(TypeError):
1707	n/a	self.loop.call_at(self.loop.time() + 60, func)
1708	n/a	with self.assertRaises(TypeError):
1709	n/a	self.loop.run_in_executor(None, func)
1710	n/a
1711	n/a	@mock.patch('asyncio.base_events.logger')
1712	n/a	def test_log_slow_callbacks(self, m_logger):
1713	n/a	def stop_loop_cb(loop):
1714	n/a	loop.stop()
1715	n/a
1716	n/a	@asyncio.coroutine
1717	n/a	def stop_loop_coro(loop):
1718	n/a	yield from ()
1719	n/a	loop.stop()
1720	n/a
1721	n/a	asyncio.set_event_loop(self.loop)
1722	n/a	self.loop.set_debug(True)
1723	n/a	self.loop.slow_callback_duration = 0.0
1724	n/a
1725	n/a	# slow callback
1726	n/a	self.loop.call_soon(stop_loop_cb, self.loop)
1727	n/a	self.loop.run_forever()
1728	n/a	fmt, *args = m_logger.warning.call_args[0]
1729	n/a	self.assertRegex(fmt % tuple(args),
1730	n/a	"^Executing <Handle.stop_loop_cb.> "
1731	n/a	"took .* seconds$")
1732	n/a
1733	n/a	# slow task
1734	n/a	asyncio.ensure_future(stop_loop_coro(self.loop), loop=self.loop)
1735	n/a	self.loop.run_forever()
1736	n/a	fmt, *args = m_logger.warning.call_args[0]
1737	n/a	self.assertRegex(fmt % tuple(args),
1738	n/a	"^Executing <Task.stop_loop_coro.> "
1739	n/a	"took .* seconds$")
1740	n/a
1741	n/a
1742	n/a	class RunningLoopTests(unittest.TestCase):
1743	n/a
1744	n/a	def test_running_loop_within_a_loop(self):
1745	n/a	@asyncio.coroutine
1746	n/a	def runner(loop):
1747	n/a	loop.run_forever()
1748	n/a
1749	n/a	loop = asyncio.new_event_loop()
1750	n/a	outer_loop = asyncio.new_event_loop()
1751	n/a	try:
1752	n/a	with self.assertRaisesRegex(RuntimeError,
1753	n/a	'while another loop is running'):
1754	n/a	outer_loop.run_until_complete(runner(loop))
1755	n/a	finally:
1756	n/a	loop.close()
1757	n/a	outer_loop.close()
1758	n/a
1759	n/a
1760	n/a	if __name__ == '__main__':
1761	n/a	unittest.main()