Python code coverage for Lib/asyncio/test_utils.py

#	count	content
1	n/a	"""Utilities shared by tests."""
2	n/a
3	n/a	import collections
4	n/a	import contextlib
5	n/a	import io
6	n/a	import logging
7	n/a	import os
8	n/a	import re
9	n/a	import socket
10	n/a	import socketserver
11	n/a	import sys
12	n/a	import tempfile
13	n/a	import threading
14	n/a	import time
15	n/a	import unittest
16	n/a	import weakref
17	n/a
18	n/a	from unittest import mock
19	n/a
20	n/a	from http.server import HTTPServer
21	n/a	from wsgiref.simple_server import WSGIRequestHandler, WSGIServer
22	n/a
23	n/a	try:
24	n/a	import ssl
25	n/a	except ImportError: # pragma: no cover
26	n/a	ssl = None
27	n/a
28	n/a	from . import base_events
29	n/a	from . import compat
30	n/a	from . import events
31	n/a	from . import futures
32	n/a	from . import selectors
33	n/a	from . import tasks
34	n/a	from .coroutines import coroutine
35	n/a	from .log import logger
36	n/a
37	n/a
38	n/a	if sys.platform == 'win32': # pragma: no cover
39	n/a	from .windows_utils import socketpair
40	n/a	else:
41	n/a	from socket import socketpair # pragma: no cover
42	n/a
43	n/a
44	n/a	def dummy_ssl_context():
45	n/a	if ssl is None:
46	n/a	return None
47	n/a	else:
48	n/a	return ssl.SSLContext(ssl.PROTOCOL_SSLv23)
49	n/a
50	n/a
51	n/a	def run_briefly(loop):
52	n/a	@coroutine
53	n/a	def once():
54	n/a	pass
55	n/a	gen = once()
56	n/a	t = loop.create_task(gen)
57	n/a	# Don't log a warning if the task is not done after run_until_complete().
58	n/a	# It occurs if the loop is stopped or if a task raises a BaseException.
59	n/a	t._log_destroy_pending = False
60	n/a	try:
61	n/a	loop.run_until_complete(t)
62	n/a	finally:
63	n/a	gen.close()
64	n/a
65	n/a
66	n/a	def run_until(loop, pred, timeout=30):
67	n/a	deadline = time.time() + timeout
68	n/a	while not pred():
69	n/a	if timeout is not None:
70	n/a	timeout = deadline - time.time()
71	n/a	if timeout <= 0:
72	n/a	raise futures.TimeoutError()
73	n/a	loop.run_until_complete(tasks.sleep(0.001, loop=loop))
74	n/a
75	n/a
76	n/a	def run_once(loop):
77	n/a	"""Legacy API to run once through the event loop.
78	n/a
79	n/a	This is the recommended pattern for test code. It will poll the
80	n/a	selector once and run all callbacks scheduled in response to I/O
81	n/a	events.
82	n/a	"""
83	n/a	loop.call_soon(loop.stop)
84	n/a	loop.run_forever()
85	n/a
86	n/a
87	n/a	class SilentWSGIRequestHandler(WSGIRequestHandler):
88	n/a
89	n/a	def get_stderr(self):
90	n/a	return io.StringIO()
91	n/a
92	n/a	def log_message(self, format, *args):
93	n/a	pass
94	n/a
95	n/a
96	n/a	class SilentWSGIServer(WSGIServer):
97	n/a
98	n/a	request_timeout = 2
99	n/a
100	n/a	def get_request(self):
101	n/a	request, client_addr = super().get_request()
102	n/a	request.settimeout(self.request_timeout)
103	n/a	return request, client_addr
104	n/a
105	n/a	def handle_error(self, request, client_address):
106	n/a	pass
107	n/a
108	n/a
109	n/a	class SSLWSGIServerMixin:
110	n/a
111	n/a	def finish_request(self, request, client_address):
112	n/a	# The relative location of our test directory (which
113	n/a	# contains the ssl key and certificate files) differs
114	n/a	# between the stdlib and stand-alone asyncio.
115	n/a	# Prefer our own if we can find it.
116	n/a	here = os.path.join(os.path.dirname(__file__), '..', 'tests')
117	n/a	if not os.path.isdir(here):
118	n/a	here = os.path.join(os.path.dirname(os.__file__),
119	n/a	'test', 'test_asyncio')
120	n/a	keyfile = os.path.join(here, 'ssl_key.pem')
121	n/a	certfile = os.path.join(here, 'ssl_cert.pem')
122	n/a	context = ssl.SSLContext()
123	n/a	context.load_cert_chain(certfile, keyfile)
124	n/a
125	n/a	ssock = context.wrap_socket(request, server_side=True)
126	n/a	try:
127	n/a	self.RequestHandlerClass(ssock, client_address, self)
128	n/a	ssock.close()
129	n/a	except OSError:
130	n/a	# maybe socket has been closed by peer
131	n/a	pass
132	n/a
133	n/a
134	n/a	class SSLWSGIServer(SSLWSGIServerMixin, SilentWSGIServer):
135	n/a	pass
136	n/a
137	n/a
138	n/a	def _run_test_server(*, address, use_ssl=False, server_cls, server_ssl_cls):
139	n/a
140	n/a	def app(environ, start_response):
141	n/a	status = '200 OK'
142	n/a	headers = [('Content-type', 'text/plain')]
143	n/a	start_response(status, headers)
144	n/a	return [b'Test message']
145	n/a
146	n/a	# Run the test WSGI server in a separate thread in order not to
147	n/a	# interfere with event handling in the main thread
148	n/a	server_class = server_ssl_cls if use_ssl else server_cls
149	n/a	httpd = server_class(address, SilentWSGIRequestHandler)
150	n/a	httpd.set_app(app)
151	n/a	httpd.address = httpd.server_address
152	n/a	server_thread = threading.Thread(
153	n/a	target=lambda: httpd.serve_forever(poll_interval=0.05))
154	n/a	server_thread.start()
155	n/a	try:
156	n/a	yield httpd
157	n/a	finally:
158	n/a	httpd.shutdown()
159	n/a	httpd.server_close()
160	n/a	server_thread.join()
161	n/a
162	n/a
163	n/a	if hasattr(socket, 'AF_UNIX'):
164	n/a
165	n/a	class UnixHTTPServer(socketserver.UnixStreamServer, HTTPServer):
166	n/a
167	n/a	def server_bind(self):
168	n/a	socketserver.UnixStreamServer.server_bind(self)
169	n/a	self.server_name = '127.0.0.1'
170	n/a	self.server_port = 80
171	n/a
172	n/a
173	n/a	class UnixWSGIServer(UnixHTTPServer, WSGIServer):
174	n/a
175	n/a	request_timeout = 2
176	n/a
177	n/a	def server_bind(self):
178	n/a	UnixHTTPServer.server_bind(self)
179	n/a	self.setup_environ()
180	n/a
181	n/a	def get_request(self):
182	n/a	request, client_addr = super().get_request()
183	n/a	request.settimeout(self.request_timeout)
184	n/a	# Code in the stdlib expects that get_request
185	n/a	# will return a socket and a tuple (host, port).
186	n/a	# However, this isn't true for UNIX sockets,
187	n/a	# as the second return value will be a path;
188	n/a	# hence we return some fake data sufficient
189	n/a	# to get the tests going
190	n/a	return request, ('127.0.0.1', '')
191	n/a
192	n/a
193	n/a	class SilentUnixWSGIServer(UnixWSGIServer):
194	n/a
195	n/a	def handle_error(self, request, client_address):
196	n/a	pass
197	n/a
198	n/a
199	n/a	class UnixSSLWSGIServer(SSLWSGIServerMixin, SilentUnixWSGIServer):
200	n/a	pass
201	n/a
202	n/a
203	n/a	def gen_unix_socket_path():
204	n/a	with tempfile.NamedTemporaryFile() as file:
205	n/a	return file.name
206	n/a
207	n/a
208	n/a	@contextlib.contextmanager
209	n/a	def unix_socket_path():
210	n/a	path = gen_unix_socket_path()
211	n/a	try:
212	n/a	yield path
213	n/a	finally:
214	n/a	try:
215	n/a	os.unlink(path)
216	n/a	except OSError:
217	n/a	pass
218	n/a
219	n/a
220	n/a	@contextlib.contextmanager
221	n/a	def run_test_unix_server(*, use_ssl=False):
222	n/a	with unix_socket_path() as path:
223	n/a	yield from _run_test_server(address=path, use_ssl=use_ssl,
224	n/a	server_cls=SilentUnixWSGIServer,
225	n/a	server_ssl_cls=UnixSSLWSGIServer)
226	n/a
227	n/a
228	n/a	@contextlib.contextmanager
229	n/a	def run_test_server(*, host='127.0.0.1', port=0, use_ssl=False):
230	n/a	yield from _run_test_server(address=(host, port), use_ssl=use_ssl,
231	n/a	server_cls=SilentWSGIServer,
232	n/a	server_ssl_cls=SSLWSGIServer)
233	n/a
234	n/a
235	n/a	def make_test_protocol(base):
236	n/a	dct = {}
237	n/a	for name in dir(base):
238	n/a	if name.startswith('__') and name.endswith('__'):
239	n/a	# skip magic names
240	n/a	continue
241	n/a	dct[name] = MockCallback(return_value=None)
242	n/a	return type('TestProtocol', (base,) + base.__bases__, dct)()
243	n/a
244	n/a
245	n/a	class TestSelector(selectors.BaseSelector):
246	n/a
247	n/a	def __init__(self):
248	n/a	self.keys = {}
249	n/a
250	n/a	def register(self, fileobj, events, data=None):
251	n/a	key = selectors.SelectorKey(fileobj, 0, events, data)
252	n/a	self.keys[fileobj] = key
253	n/a	return key
254	n/a
255	n/a	def unregister(self, fileobj):
256	n/a	return self.keys.pop(fileobj)
257	n/a
258	n/a	def select(self, timeout):
259	n/a	return []
260	n/a
261	n/a	def get_map(self):
262	n/a	return self.keys
263	n/a
264	n/a
265	n/a	class TestLoop(base_events.BaseEventLoop):
266	n/a	"""Loop for unittests.
267	n/a
268	n/a	It manages self time directly.
269	n/a	If something scheduled to be executed later then
270	n/a	on next loop iteration after all ready handlers done
271	n/a	generator passed to __init__ is calling.
272	n/a
273	n/a	Generator should be like this:
274	n/a
275	n/a	def gen():
276	n/a	...
277	n/a	when = yield ...
278	n/a	... = yield time_advance
279	n/a
280	n/a	Value returned by yield is absolute time of next scheduled handler.
281	n/a	Value passed to yield is time advance to move loop's time forward.
282	n/a	"""
283	n/a
284	n/a	def __init__(self, gen=None):
285	n/a	super().__init__()
286	n/a
287	n/a	if gen is None:
288	n/a	def gen():
289	n/a	yield
290	n/a	self._check_on_close = False
291	n/a	else:
292	n/a	self._check_on_close = True
293	n/a
294	n/a	self._gen = gen()
295	n/a	next(self._gen)
296	n/a	self._time = 0
297	n/a	self._clock_resolution = 1e-9
298	n/a	self._timers = []
299	n/a	self._selector = TestSelector()
300	n/a
301	n/a	self.readers = {}
302	n/a	self.writers = {}
303	n/a	self.reset_counters()
304	n/a
305	n/a	self._transports = weakref.WeakValueDictionary()
306	n/a
307	n/a	def time(self):
308	n/a	return self._time
309	n/a
310	n/a	def advance_time(self, advance):
311	n/a	"""Move test time forward."""
312	n/a	if advance:
313	n/a	self._time += advance
314	n/a
315	n/a	def close(self):
316	n/a	super().close()
317	n/a	if self._check_on_close:
318	n/a	try:
319	n/a	self._gen.send(0)
320	n/a	except StopIteration:
321	n/a	pass
322	n/a	else: # pragma: no cover
323	n/a	raise AssertionError("Time generator is not finished")
324	n/a
325	n/a	def _add_reader(self, fd, callback, *args):
326	n/a	self.readers[fd] = events.Handle(callback, args, self)
327	n/a
328	n/a	def _remove_reader(self, fd):
329	n/a	self.remove_reader_count[fd] += 1
330	n/a	if fd in self.readers:
331	n/a	del self.readers[fd]
332	n/a	return True
333	n/a	else:
334	n/a	return False
335	n/a
336	n/a	def assert_reader(self, fd, callback, *args):
337	n/a	assert fd in self.readers, 'fd {} is not registered'.format(fd)
338	n/a	handle = self.readers[fd]
339	n/a	assert handle._callback == callback, '{!r} != {!r}'.format(
340	n/a	handle._callback, callback)
341	n/a	assert handle._args == args, '{!r} != {!r}'.format(
342	n/a	handle._args, args)
343	n/a
344	n/a	def _add_writer(self, fd, callback, *args):
345	n/a	self.writers[fd] = events.Handle(callback, args, self)
346	n/a
347	n/a	def _remove_writer(self, fd):
348	n/a	self.remove_writer_count[fd] += 1
349	n/a	if fd in self.writers:
350	n/a	del self.writers[fd]
351	n/a	return True
352	n/a	else:
353	n/a	return False
354	n/a
355	n/a	def assert_writer(self, fd, callback, *args):
356	n/a	assert fd in self.writers, 'fd {} is not registered'.format(fd)
357	n/a	handle = self.writers[fd]
358	n/a	assert handle._callback == callback, '{!r} != {!r}'.format(
359	n/a	handle._callback, callback)
360	n/a	assert handle._args == args, '{!r} != {!r}'.format(
361	n/a	handle._args, args)
362	n/a
363	n/a	def _ensure_fd_no_transport(self, fd):
364	n/a	try:
365	n/a	transport = self._transports[fd]
366	n/a	except KeyError:
367	n/a	pass
368	n/a	else:
369	n/a	raise RuntimeError(
370	n/a	'File descriptor {!r} is used by transport {!r}'.format(
371	n/a	fd, transport))
372	n/a
373	n/a	def add_reader(self, fd, callback, *args):
374	n/a	"""Add a reader callback."""
375	n/a	self._ensure_fd_no_transport(fd)
376	n/a	return self._add_reader(fd, callback, *args)
377	n/a
378	n/a	def remove_reader(self, fd):
379	n/a	"""Remove a reader callback."""
380	n/a	self._ensure_fd_no_transport(fd)
381	n/a	return self._remove_reader(fd)
382	n/a
383	n/a	def add_writer(self, fd, callback, *args):
384	n/a	"""Add a writer callback.."""
385	n/a	self._ensure_fd_no_transport(fd)
386	n/a	return self._add_writer(fd, callback, *args)
387	n/a
388	n/a	def remove_writer(self, fd):
389	n/a	"""Remove a writer callback."""
390	n/a	self._ensure_fd_no_transport(fd)
391	n/a	return self._remove_writer(fd)
392	n/a
393	n/a	def reset_counters(self):
394	n/a	self.remove_reader_count = collections.defaultdict(int)
395	n/a	self.remove_writer_count = collections.defaultdict(int)
396	n/a
397	n/a	def _run_once(self):
398	n/a	super()._run_once()
399	n/a	for when in self._timers:
400	n/a	advance = self._gen.send(when)
401	n/a	self.advance_time(advance)
402	n/a	self._timers = []
403	n/a
404	n/a	def call_at(self, when, callback, *args):
405	n/a	self._timers.append(when)
406	n/a	return super().call_at(when, callback, *args)
407	n/a
408	n/a	def _process_events(self, event_list):
409	n/a	return
410	n/a
411	n/a	def _write_to_self(self):
412	n/a	pass
413	n/a
414	n/a
415	n/a	def MockCallback(**kwargs):
416	n/a	return mock.Mock(spec=['__call__'], **kwargs)
417	n/a
418	n/a
419	n/a	class MockPattern(str):
420	n/a	"""A regex based str with a fuzzy __eq__.
421	n/a
422	n/a	Use this helper with 'mock.assert_called_with', or anywhere
423	n/a	where a regex comparison between strings is needed.
424	n/a
425	n/a	For instance:
426	n/a	mock_call.assert_called_with(MockPattern('spam.*ham'))
427	n/a	"""
428	n/a	def __eq__(self, other):
429	n/a	return bool(re.search(str(self), other, re.S))
430	n/a
431	n/a
432	n/a	def get_function_source(func):
433	n/a	source = events._get_function_source(func)
434	n/a	if source is None:
435	n/a	raise ValueError("unable to get the source of %r" % (func,))
436	n/a	return source
437	n/a
438	n/a
439	n/a	class TestCase(unittest.TestCase):
440	n/a	def set_event_loop(self, loop, *, cleanup=True):
441	n/a	assert loop is not None
442	n/a	# ensure that the event loop is passed explicitly in asyncio
443	n/a	events.set_event_loop(None)
444	n/a	if cleanup:
445	n/a	self.addCleanup(loop.close)
446	n/a
447	n/a	def new_test_loop(self, gen=None):
448	n/a	loop = TestLoop(gen)
449	n/a	self.set_event_loop(loop)
450	n/a	return loop
451	n/a
452	n/a	def setUp(self):
453	n/a	self._get_running_loop = events._get_running_loop
454	n/a	events._get_running_loop = lambda: None
455	n/a
456	n/a	def tearDown(self):
457	n/a	events._get_running_loop = self._get_running_loop
458	n/a
459	n/a	events.set_event_loop(None)
460	n/a
461	n/a	# Detect CPython bug #23353: ensure that yield/yield-from is not used
462	n/a	# in an except block of a generator
463	n/a	self.assertEqual(sys.exc_info(), (None, None, None))
464	n/a
465	n/a	if not compat.PY34:
466	n/a	# Python 3.3 compatibility
467	n/a	def subTest(self, args, *kwargs):
468	n/a	class EmptyCM:
469	n/a	def __enter__(self):
470	n/a	pass
471	n/a	def __exit__(self, *exc):
472	n/a	pass
473	n/a	return EmptyCM()
474	n/a
475	n/a
476	n/a	@contextlib.contextmanager
477	n/a	def disable_logger():
478	n/a	"""Context manager to disable asyncio logger.
479	n/a
480	n/a	For example, it can be used to ignore warnings in debug mode.
481	n/a	"""
482	n/a	old_level = logger.level
483	n/a	try:
484	n/a	logger.setLevel(logging.CRITICAL+1)
485	n/a	yield
486	n/a	finally:
487	n/a	logger.setLevel(old_level)
488	n/a
489	n/a
490	n/a	def mock_nonblocking_socket(proto=socket.IPPROTO_TCP, type=socket.SOCK_STREAM,
491	n/a	family=socket.AF_INET):
492	n/a	"""Create a mock of a non-blocking socket."""
493	n/a	sock = mock.MagicMock(socket.socket)
494	n/a	sock.proto = proto
495	n/a	sock.type = type
496	n/a	sock.family = family
497	n/a	sock.gettimeout.return_value = 0.0
498	n/a	return sock
499	n/a
500	n/a
501	n/a	def force_legacy_ssl_support():
502	n/a	return mock.patch('asyncio.sslproto._is_sslproto_available',
503	n/a	return_value=False)