Python code coverage for Lib/test/support.py

#	count	content
1	n/a	"""Supporting definitions for the Python regression tests."""
2	n/a
3	n/a	if __name__ != 'test.support':
4	n/a	raise ImportError('support must be imported from the test package')
5	n/a
6	n/a	import contextlib
7	n/a	import errno
8	n/a	import functools
9	n/a	import gc
10	n/a	import socket
11	n/a	import sys
12	n/a	import os
13	n/a	import platform
14	n/a	import shutil
15	n/a	import warnings
16	n/a	import unittest
17	n/a	import importlib
18	n/a	import importlib.util
19	n/a	import collections.abc
20	n/a	import re
21	n/a	import subprocess
22	n/a	import time
23	n/a	import sysconfig
24	n/a	import fnmatch
25	n/a	import logging.handlers
26	n/a	import struct
27	n/a	import tempfile
28	n/a	import _testcapi
29	n/a
30	n/a	try:
31	n/a	import _thread, threading
32	n/a	except ImportError:
33	n/a	_thread = None
34	n/a	threading = None
35	n/a	try:
36	n/a	import multiprocessing.process
37	n/a	except ImportError:
38	n/a	multiprocessing = None
39	n/a
40	n/a	try:
41	n/a	import zlib
42	n/a	except ImportError:
43	n/a	zlib = None
44	n/a
45	n/a	try:
46	n/a	import gzip
47	n/a	except ImportError:
48	n/a	gzip = None
49	n/a
50	n/a	try:
51	n/a	import bz2
52	n/a	except ImportError:
53	n/a	bz2 = None
54	n/a
55	n/a	try:
56	n/a	import lzma
57	n/a	except ImportError:
58	n/a	lzma = None
59	n/a
60	n/a	__all__ = [
61	n/a	"Error", "TestFailed", "ResourceDenied", "import_module", "verbose",
62	n/a	"use_resources", "max_memuse", "record_original_stdout",
63	n/a	"get_original_stdout", "unload", "unlink", "rmtree", "forget",
64	n/a	"is_resource_enabled", "requires", "requires_freebsd_version",
65	n/a	"requires_linux_version", "requires_mac_ver", "find_unused_port",
66	n/a	"bind_port", "IPV6_ENABLED", "is_jython", "TESTFN", "HOST", "SAVEDCWD",
67	n/a	"temp_cwd", "findfile", "create_empty_file", "sortdict",
68	n/a	"check_syntax_error", "open_urlresource", "check_warnings", "CleanImport",
69	n/a	"EnvironmentVarGuard", "TransientResource", "captured_stdout",
70	n/a	"captured_stdin", "captured_stderr", "time_out", "socket_peer_reset",
71	n/a	"ioerror_peer_reset", "run_with_locale", 'temp_umask',
72	n/a	"transient_internet", "set_memlimit", "bigmemtest", "bigaddrspacetest",
73	n/a	"BasicTestRunner", "run_unittest", "run_doctest", "threading_setup",
74	n/a	"threading_cleanup", "reap_children", "cpython_only", "check_impl_detail",
75	n/a	"get_attribute", "swap_item", "swap_attr", "requires_IEEE_754",
76	n/a	"TestHandler", "Matcher", "can_symlink", "skip_unless_symlink",
77	n/a	"skip_unless_xattr", "import_fresh_module", "requires_zlib",
78	n/a	"PIPE_MAX_SIZE", "failfast", "anticipate_failure", "run_with_tz",
79	n/a	"requires_gzip", "requires_bz2", "requires_lzma", "suppress_crash_popup",
80	n/a	]
81	n/a
82	n/a	class Error(Exception):
83	n/a	"""Base class for regression test exceptions."""
84	n/a
85	n/a	class TestFailed(Error):
86	n/a	"""Test failed."""
87	n/a
88	n/a	class ResourceDenied(unittest.SkipTest):
89	n/a	"""Test skipped because it requested a disallowed resource.
90	n/a
91	n/a	This is raised when a test calls requires() for a resource that
92	n/a	has not be enabled. It is used to distinguish between expected
93	n/a	and unexpected skips.
94	n/a	"""
95	n/a
96	n/a	@contextlib.contextmanager
97	n/a	def _ignore_deprecated_imports(ignore=True):
98	n/a	"""Context manager to suppress package and module deprecation
99	n/a	warnings when importing them.
100	n/a
101	n/a	If ignore is False, this context manager has no effect.
102	n/a	"""
103	n/a	if ignore:
104	n/a	with warnings.catch_warnings():
105	n/a	warnings.filterwarnings("ignore", ".+ (module\|package)",
106	n/a	DeprecationWarning)
107	n/a	yield
108	n/a	else:
109	n/a	yield
110	n/a
111	n/a
112	n/a	def import_module(name, deprecated=False, *, required_on=()):
113	n/a	"""Import and return the module to be tested, raising SkipTest if
114	n/a	it is not available.
115	n/a
116	n/a	If deprecated is True, any module or package deprecation messages
117	n/a	will be suppressed. If a module is required on a platform but optional for
118	n/a	others, set required_on to an iterable of platform prefixes which will be
119	n/a	compared against sys.platform.
120	n/a	"""
121	n/a	with _ignore_deprecated_imports(deprecated):
122	n/a	try:
123	n/a	return importlib.import_module(name)
124	n/a	except ImportError as msg:
125	n/a	if sys.platform.startswith(tuple(required_on)):
126	n/a	raise
127	n/a	raise unittest.SkipTest(str(msg))
128	n/a
129	n/a
130	n/a	def _save_and_remove_module(name, orig_modules):
131	n/a	"""Helper function to save and remove a module from sys.modules
132	n/a
133	n/a	Raise ImportError if the module can't be imported.
134	n/a	"""
135	n/a	# try to import the module and raise an error if it can't be imported
136	n/a	if name not in sys.modules:
137	n/a	__import__(name)
138	n/a	del sys.modules[name]
139	n/a	for modname in list(sys.modules):
140	n/a	if modname == name or modname.startswith(name + '.'):
141	n/a	orig_modules[modname] = sys.modules[modname]
142	n/a	del sys.modules[modname]
143	n/a
144	n/a	def _save_and_block_module(name, orig_modules):
145	n/a	"""Helper function to save and block a module in sys.modules
146	n/a
147	n/a	Return True if the module was in sys.modules, False otherwise."""
148	n/a	saved = True
149	n/a	try:
150	n/a	orig_modules[name] = sys.modules[name]
151	n/a	except KeyError:
152	n/a	saved = False
153	n/a	sys.modules[name] = None
154	n/a	return saved
155	n/a
156	n/a
157	n/a	def anticipate_failure(condition):
158	n/a	"""Decorator to mark a test that is known to be broken in some cases
159	n/a
160	n/a	Any use of this decorator should have a comment identifying the
161	n/a	associated tracker issue.
162	n/a	"""
163	n/a	if condition:
164	n/a	return unittest.expectedFailure
165	n/a	return lambda f: f
166	n/a
167	n/a
168	n/a	def import_fresh_module(name, fresh=(), blocked=(), deprecated=False):
169	n/a	"""Imports and returns a module, deliberately bypassing the sys.modules cache
170	n/a	and importing a fresh copy of the module. Once the import is complete,
171	n/a	the sys.modules cache is restored to its original state.
172	n/a
173	n/a	Modules named in fresh are also imported anew if needed by the import.
174	n/a	If one of these modules can't be imported, None is returned.
175	n/a
176	n/a	Importing of modules named in blocked is prevented while the fresh import
177	n/a	takes place.
178	n/a
179	n/a	If deprecated is True, any module or package deprecation messages
180	n/a	will be suppressed."""
181	n/a	# NOTE: test_heapq, test_json and test_warnings include extra sanity checks
182	n/a	# to make sure that this utility function is working as expected
183	n/a	with _ignore_deprecated_imports(deprecated):
184	n/a	# Keep track of modules saved for later restoration as well
185	n/a	# as those which just need a blocking entry removed
186	n/a	orig_modules = {}
187	n/a	names_to_remove = []
188	n/a	_save_and_remove_module(name, orig_modules)
189	n/a	try:
190	n/a	for fresh_name in fresh:
191	n/a	_save_and_remove_module(fresh_name, orig_modules)
192	n/a	for blocked_name in blocked:
193	n/a	if not _save_and_block_module(blocked_name, orig_modules):
194	n/a	names_to_remove.append(blocked_name)
195	n/a	fresh_module = importlib.import_module(name)
196	n/a	except ImportError:
197	n/a	fresh_module = None
198	n/a	finally:
199	n/a	for orig_name, module in orig_modules.items():
200	n/a	sys.modules[orig_name] = module
201	n/a	for name_to_remove in names_to_remove:
202	n/a	del sys.modules[name_to_remove]
203	n/a	return fresh_module
204	n/a
205	n/a
206	n/a	def get_attribute(obj, name):
207	n/a	"""Get an attribute, raising SkipTest if AttributeError is raised."""
208	n/a	try:
209	n/a	attribute = getattr(obj, name)
210	n/a	except AttributeError:
211	n/a	raise unittest.SkipTest("object %r has no attribute %r" % (obj, name))
212	n/a	else:
213	n/a	return attribute
214	n/a
215	n/a	verbose = 1 # Flag set to 0 by regrtest.py
216	n/a	use_resources = None # Flag set to [] by regrtest.py
217	n/a	max_memuse = 0 # Disable bigmem tests (they will still be run with
218	n/a	# small sizes, to make sure they work.)
219	n/a	real_max_memuse = 0
220	n/a	failfast = False
221	n/a	match_tests = None
222	n/a
223	n/a	# _original_stdout is meant to hold stdout at the time regrtest began.
224	n/a	# This may be "the real" stdout, or IDLE's emulation of stdout, or whatever.
225	n/a	# The point is to have some flavor of stdout the user can actually see.
226	n/a	_original_stdout = None
227	n/a	def record_original_stdout(stdout):
228	n/a	global _original_stdout
229	n/a	_original_stdout = stdout
230	n/a
231	n/a	def get_original_stdout():
232	n/a	return _original_stdout or sys.stdout
233	n/a
234	n/a	def unload(name):
235	n/a	try:
236	n/a	del sys.modules[name]
237	n/a	except KeyError:
238	n/a	pass
239	n/a
240	n/a	if sys.platform.startswith("win"):
241	n/a	def _waitfor(func, pathname, waitall=False):
242	n/a	# Peform the operation
243	n/a	func(pathname)
244	n/a	# Now setup the wait loop
245	n/a	if waitall:
246	n/a	dirname = pathname
247	n/a	else:
248	n/a	dirname, name = os.path.split(pathname)
249	n/a	dirname = dirname or '.'
250	n/a	# Check for `pathname` to be removed from the filesystem.
251	n/a	# The exponential backoff of the timeout amounts to a total
252	n/a	# of ~1 second after which the deletion is probably an error
253	n/a	# anyway.
254	n/a	# Testing on a i7@4.3GHz shows that usually only 1 iteration is
255	n/a	# required when contention occurs.
256	n/a	timeout = 0.001
257	n/a	while timeout < 1.0:
258	n/a	# Note we are only testing for the existance of the file(s) in
259	n/a	# the contents of the directory regardless of any security or
260	n/a	# access rights. If we have made it this far, we have sufficient
261	n/a	# permissions to do that much using Python's equivalent of the
262	n/a	# Windows API FindFirstFile.
263	n/a	# Other Windows APIs can fail or give incorrect results when
264	n/a	# dealing with files that are pending deletion.
265	n/a	L = os.listdir(dirname)
266	n/a	if not (L if waitall else name in L):
267	n/a	return
268	n/a	# Increase the timeout and try again
269	n/a	time.sleep(timeout)
270	n/a	timeout *= 2
271	n/a	warnings.warn('tests may fail, delete still pending for ' + pathname,
272	n/a	RuntimeWarning, stacklevel=4)
273	n/a
274	n/a	def _unlink(filename):
275	n/a	_waitfor(os.unlink, filename)
276	n/a
277	n/a	def _rmdir(dirname):
278	n/a	_waitfor(os.rmdir, dirname)
279	n/a
280	n/a	def _rmtree(path):
281	n/a	def _rmtree_inner(path):
282	n/a	for name in os.listdir(path):
283	n/a	fullname = os.path.join(path, name)
284	n/a	if os.path.isdir(fullname):
285	n/a	_waitfor(_rmtree_inner, fullname, waitall=True)
286	n/a	os.rmdir(fullname)
287	n/a	else:
288	n/a	os.unlink(fullname)
289	n/a	_waitfor(_rmtree_inner, path, waitall=True)
290	n/a	_waitfor(os.rmdir, path)
291	n/a	else:
292	n/a	_unlink = os.unlink
293	n/a	_rmdir = os.rmdir
294	n/a	_rmtree = shutil.rmtree
295	n/a
296	n/a	def unlink(filename):
297	n/a	try:
298	n/a	_unlink(filename)
299	n/a	except (FileNotFoundError, NotADirectoryError):
300	n/a	pass
301	n/a
302	n/a	def rmdir(dirname):
303	n/a	try:
304	n/a	_rmdir(dirname)
305	n/a	except FileNotFoundError:
306	n/a	pass
307	n/a
308	n/a	def rmtree(path):
309	n/a	try:
310	n/a	_rmtree(path)
311	n/a	except FileNotFoundError:
312	n/a	pass
313	n/a
314	n/a	def make_legacy_pyc(source):
315	n/a	"""Move a PEP 3147 pyc/pyo file to its legacy pyc/pyo location.
316	n/a
317	n/a	The choice of .pyc or .pyo extension is done based on the __debug__ flag
318	n/a	value.
319	n/a
320	n/a	:param source: The file system path to the source file. The source file
321	n/a	does not need to exist, however the PEP 3147 pyc file must exist.
322	n/a	:return: The file system path to the legacy pyc file.
323	n/a	"""
324	n/a	pyc_file = importlib.util.cache_from_source(source)
325	n/a	up_one = os.path.dirname(os.path.abspath(source))
326	n/a	legacy_pyc = os.path.join(up_one, source + ('c' if __debug__ else 'o'))
327	n/a	os.rename(pyc_file, legacy_pyc)
328	n/a	return legacy_pyc
329	n/a
330	n/a	def forget(modname):
331	n/a	"""'Forget' a module was ever imported.
332	n/a
333	n/a	This removes the module from sys.modules and deletes any PEP 3147 or
334	n/a	legacy .pyc and .pyo files.
335	n/a	"""
336	n/a	unload(modname)
337	n/a	for dirname in sys.path:
338	n/a	source = os.path.join(dirname, modname + '.py')
339	n/a	# It doesn't matter if they exist or not, unlink all possible
340	n/a	# combinations of PEP 3147 and legacy pyc and pyo files.
341	n/a	unlink(source + 'c')
342	n/a	unlink(source + 'o')
343	n/a	unlink(importlib.util.cache_from_source(source, debug_override=True))
344	n/a	unlink(importlib.util.cache_from_source(source, debug_override=False))
345	n/a
346	n/a	# On some platforms, should not run gui test even if it is allowed
347	n/a	# in `use_resources'.
348	n/a	if sys.platform.startswith('win'):
349	n/a	import ctypes
350	n/a	import ctypes.wintypes
351	n/a	def _is_gui_available():
352	n/a	UOI_FLAGS = 1
353	n/a	WSF_VISIBLE = 0x0001
354	n/a	class USEROBJECTFLAGS(ctypes.Structure):
355	n/a	_fields_ = [("fInherit", ctypes.wintypes.BOOL),
356	n/a	("fReserved", ctypes.wintypes.BOOL),
357	n/a	("dwFlags", ctypes.wintypes.DWORD)]
358	n/a	dll = ctypes.windll.user32
359	n/a	h = dll.GetProcessWindowStation()
360	n/a	if not h:
361	n/a	raise ctypes.WinError()
362	n/a	uof = USEROBJECTFLAGS()
363	n/a	needed = ctypes.wintypes.DWORD()
364	n/a	res = dll.GetUserObjectInformationW(h,
365	n/a	UOI_FLAGS,
366	n/a	ctypes.byref(uof),
367	n/a	ctypes.sizeof(uof),
368	n/a	ctypes.byref(needed))
369	n/a	if not res:
370	n/a	raise ctypes.WinError()
371	n/a	return bool(uof.dwFlags & WSF_VISIBLE)
372	n/a	else:
373	n/a	def _is_gui_available():
374	n/a	return True
375	n/a
376	n/a	def is_resource_enabled(resource):
377	n/a	"""Test whether a resource is enabled. Known resources are set by
378	n/a	regrtest.py."""
379	n/a	return use_resources is not None and resource in use_resources
380	n/a
381	n/a	def requires(resource, msg=None):
382	n/a	"""Raise ResourceDenied if the specified resource is not available.
383	n/a
384	n/a	If the caller's module is __main__ then automatically return True. The
385	n/a	possibility of False being returned occurs when regrtest.py is
386	n/a	executing.
387	n/a	"""
388	n/a	if resource == 'gui' and not _is_gui_available():
389	n/a	raise unittest.SkipTest("Cannot use the 'gui' resource")
390	n/a	# see if the caller's module is __main__ - if so, treat as if
391	n/a	# the resource was set
392	n/a	if sys._getframe(1).f_globals.get("__name__") == "__main__":
393	n/a	return
394	n/a	if not is_resource_enabled(resource):
395	n/a	if msg is None:
396	n/a	msg = "Use of the %r resource not enabled" % resource
397	n/a	raise ResourceDenied(msg)
398	n/a
399	n/a	def _requires_unix_version(sysname, min_version):
400	n/a	"""Decorator raising SkipTest if the OS is `sysname` and the version is less
401	n/a	than `min_version`.
402	n/a
403	n/a	For example, @_requires_unix_version('FreeBSD', (7, 2)) raises SkipTest if
404	n/a	the FreeBSD version is less than 7.2.
405	n/a	"""
406	n/a	def decorator(func):
407	n/a	@functools.wraps(func)
408	n/a	def wrapper(args, *kw):
409	n/a	if platform.system() == sysname:
410	n/a	version_txt = platform.release().split('-', 1)[0]
411	n/a	try:
412	n/a	version = tuple(map(int, version_txt.split('.')))
413	n/a	except ValueError:
414	n/a	pass
415	n/a	else:
416	n/a	if version < min_version:
417	n/a	min_version_txt = '.'.join(map(str, min_version))
418	n/a	raise unittest.SkipTest(
419	n/a	"%s version %s or higher required, not %s"
420	n/a	% (sysname, min_version_txt, version_txt))
421	n/a	return wrapper
422	n/a	return decorator
423	n/a
424	n/a	def requires_freebsd_version(*min_version):
425	n/a	"""Decorator raising SkipTest if the OS is FreeBSD and the FreeBSD version is
426	n/a	less than `min_version`.
427	n/a
428	n/a	For example, @requires_freebsd_version(7, 2) raises SkipTest if the FreeBSD
429	n/a	version is less than 7.2.
430	n/a	"""
431	n/a	return _requires_unix_version('FreeBSD', min_version)
432	n/a
433	n/a	def requires_linux_version(*min_version):
434	n/a	"""Decorator raising SkipTest if the OS is Linux and the Linux version is
435	n/a	less than `min_version`.
436	n/a
437	n/a	For example, @requires_linux_version(2, 6, 32) raises SkipTest if the Linux
438	n/a	version is less than 2.6.32.
439	n/a	"""
440	n/a	return _requires_unix_version('Linux', min_version)
441	n/a
442	n/a	def requires_mac_ver(*min_version):
443	n/a	"""Decorator raising SkipTest if the OS is Mac OS X and the OS X
444	n/a	version if less than min_version.
445	n/a
446	n/a	For example, @requires_mac_ver(10, 5) raises SkipTest if the OS X version
447	n/a	is lesser than 10.5.
448	n/a	"""
449	n/a	def decorator(func):
450	n/a	@functools.wraps(func)
451	n/a	def wrapper(args, *kw):
452	n/a	if sys.platform == 'darwin':
453	n/a	version_txt = platform.mac_ver()[0]
454	n/a	try:
455	n/a	version = tuple(map(int, version_txt.split('.')))
456	n/a	except ValueError:
457	n/a	pass
458	n/a	else:
459	n/a	if version < min_version:
460	n/a	min_version_txt = '.'.join(map(str, min_version))
461	n/a	raise unittest.SkipTest(
462	n/a	"Mac OS X %s or higher required, not %s"
463	n/a	% (min_version_txt, version_txt))
464	n/a	return func(args, *kw)
465	n/a	wrapper.min_version = min_version
466	n/a	return wrapper
467	n/a	return decorator
468	n/a
469	n/a
470	n/a	HOST = 'localhost'
471	n/a
472	n/a	def find_unused_port(family=socket.AF_INET, socktype=socket.SOCK_STREAM):
473	n/a	"""Returns an unused port that should be suitable for binding. This is
474	n/a	achieved by creating a temporary socket with the same family and type as
475	n/a	the 'sock' parameter (default is AF_INET, SOCK_STREAM), and binding it to
476	n/a	the specified host address (defaults to 0.0.0.0) with the port set to 0,
477	n/a	eliciting an unused ephemeral port from the OS. The temporary socket is
478	n/a	then closed and deleted, and the ephemeral port is returned.
479	n/a
480	n/a	Either this method or bind_port() should be used for any tests where a
481	n/a	server socket needs to be bound to a particular port for the duration of
482	n/a	the test. Which one to use depends on whether the calling code is creating
483	n/a	a python socket, or if an unused port needs to be provided in a constructor
484	n/a	or passed to an external program (i.e. the -accept argument to openssl's
485	n/a	s_server mode). Always prefer bind_port() over find_unused_port() where
486	n/a	possible. Hard coded ports should NEVER be used. As soon as a server
487	n/a	socket is bound to a hard coded port, the ability to run multiple instances
488	n/a	of the test simultaneously on the same host is compromised, which makes the
489	n/a	test a ticking time bomb in a buildbot environment. On Unix buildbots, this
490	n/a	may simply manifest as a failed test, which can be recovered from without
491	n/a	intervention in most cases, but on Windows, the entire python process can
492	n/a	completely and utterly wedge, requiring someone to log in to the buildbot
493	n/a	and manually kill the affected process.
494	n/a
495	n/a	(This is easy to reproduce on Windows, unfortunately, and can be traced to
496	n/a	the SO_REUSEADDR socket option having different semantics on Windows versus
497	n/a	Unix/Linux. On Unix, you can't have two AF_INET SOCK_STREAM sockets bind,
498	n/a	listen and then accept connections on identical host/ports. An EADDRINUSE
499	n/a	OSError will be raised at some point (depending on the platform and
500	n/a	the order bind and listen were called on each socket).
501	n/a
502	n/a	However, on Windows, if SO_REUSEADDR is set on the sockets, no EADDRINUSE
503	n/a	will ever be raised when attempting to bind two identical host/ports. When
504	n/a	accept() is called on each socket, the second caller's process will steal
505	n/a	the port from the first caller, leaving them both in an awkwardly wedged
506	n/a	state where they'll no longer respond to any signals or graceful kills, and
507	n/a	must be forcibly killed via OpenProcess()/TerminateProcess().
508	n/a
509	n/a	The solution on Windows is to use the SO_EXCLUSIVEADDRUSE socket option
510	n/a	instead of SO_REUSEADDR, which effectively affords the same semantics as
511	n/a	SO_REUSEADDR on Unix. Given the propensity of Unix developers in the Open
512	n/a	Source world compared to Windows ones, this is a common mistake. A quick
513	n/a	look over OpenSSL's 0.9.8g source shows that they use SO_REUSEADDR when
514	n/a	openssl.exe is called with the 's_server' option, for example. See
515	n/a	http://bugs.python.org/issue2550 for more info. The following site also
516	n/a	has a very thorough description about the implications of both REUSEADDR
517	n/a	and EXCLUSIVEADDRUSE on Windows:
518	n/a	http://msdn2.microsoft.com/en-us/library/ms740621(VS.85).aspx)
519	n/a
520	n/a	XXX: although this approach is a vast improvement on previous attempts to
521	n/a	elicit unused ports, it rests heavily on the assumption that the ephemeral
522	n/a	port returned to us by the OS won't immediately be dished back out to some
523	n/a	other process when we close and delete our temporary socket but before our
524	n/a	calling code has a chance to bind the returned port. We can deal with this
525	n/a	issue if/when we come across it.
526	n/a	"""
527	n/a
528	n/a	tempsock = socket.socket(family, socktype)
529	n/a	port = bind_port(tempsock)
530	n/a	tempsock.close()
531	n/a	del tempsock
532	n/a	return port
533	n/a
534	n/a	def bind_port(sock, host=HOST):
535	n/a	"""Bind the socket to a free port and return the port number. Relies on
536	n/a	ephemeral ports in order to ensure we are using an unbound port. This is
537	n/a	important as many tests may be running simultaneously, especially in a
538	n/a	buildbot environment. This method raises an exception if the sock.family
539	n/a	is AF_INET and sock.type is SOCK_STREAM, and the socket has SO_REUSEADDR
540	n/a	or SO_REUSEPORT set on it. Tests should never set these socket options
541	n/a	for TCP/IP sockets. The only case for setting these options is testing
542	n/a	multicasting via multiple UDP sockets.
543	n/a
544	n/a	Additionally, if the SO_EXCLUSIVEADDRUSE socket option is available (i.e.
545	n/a	on Windows), it will be set on the socket. This will prevent anyone else
546	n/a	from bind()'ing to our host/port for the duration of the test.
547	n/a	"""
548	n/a
549	n/a	if sock.family == socket.AF_INET and sock.type == socket.SOCK_STREAM:
550	n/a	if hasattr(socket, 'SO_REUSEADDR'):
551	n/a	if sock.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR) == 1:
552	n/a	raise TestFailed("tests should never set the SO_REUSEADDR " \
553	n/a	"socket option on TCP/IP sockets!")
554	n/a	if hasattr(socket, 'SO_REUSEPORT'):
555	n/a	if sock.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT) == 1:
556	n/a	raise TestFailed("tests should never set the SO_REUSEPORT " \
557	n/a	"socket option on TCP/IP sockets!")
558	n/a	if hasattr(socket, 'SO_EXCLUSIVEADDRUSE'):
559	n/a	sock.setsockopt(socket.SOL_SOCKET, socket.SO_EXCLUSIVEADDRUSE, 1)
560	n/a
561	n/a	sock.bind((host, 0))
562	n/a	port = sock.getsockname()[1]
563	n/a	return port
564	n/a
565	n/a	def _is_ipv6_enabled():
566	n/a	"""Check whether IPv6 is enabled on this host."""
567	n/a	if socket.has_ipv6:
568	n/a	sock = None
569	n/a	try:
570	n/a	sock = socket.socket(socket.AF_INET6, socket.SOCK_STREAM)
571	n/a	sock.bind(('::1', 0))
572	n/a	return True
573	n/a	except OSError:
574	n/a	pass
575	n/a	finally:
576	n/a	if sock:
577	n/a	sock.close()
578	n/a	return False
579	n/a
580	n/a	IPV6_ENABLED = _is_ipv6_enabled()
581	n/a
582	n/a
583	n/a	# A constant likely larger than the underlying OS pipe buffer size, to
584	n/a	# make writes blocking.
585	n/a	# Windows limit seems to be around 512 B, and many Unix kernels have a
586	n/a	# 64 KiB pipe buffer size or 16 * PAGE_SIZE: take a few megs to be sure.
587	n/a	# (see issue #17835 for a discussion of this number).
588	n/a	PIPE_MAX_SIZE = 4 1024 1024 + 1
589	n/a
590	n/a
591	n/a	# decorator for skipping tests on non-IEEE 754 platforms
592	n/a	requires_IEEE_754 = unittest.skipUnless(
593	n/a	float.__getformat__("double").startswith("IEEE"),
594	n/a	"test requires IEEE 754 doubles")
595	n/a
596	n/a	requires_zlib = unittest.skipUnless(zlib, 'requires zlib')
597	n/a
598	n/a	requires_gzip = unittest.skipUnless(gzip, 'requires gzip')
599	n/a
600	n/a	requires_bz2 = unittest.skipUnless(bz2, 'requires bz2')
601	n/a
602	n/a	requires_lzma = unittest.skipUnless(lzma, 'requires lzma')
603	n/a
604	n/a	is_jython = sys.platform.startswith('java')
605	n/a
606	n/a	# Filename used for testing
607	n/a	if os.name == 'java':
608	n/a	# Jython disallows @ in module names
609	n/a	TESTFN = '$test'
610	n/a	else:
611	n/a	TESTFN = '@test'
612	n/a
613	n/a	# Disambiguate TESTFN for parallel testing, while letting it remain a valid
614	n/a	# module name.
615	n/a	TESTFN = "{}_{}_tmp".format(TESTFN, os.getpid())
616	n/a
617	n/a	# FS_NONASCII: non-ASCII character encodable by os.fsencode(),
618	n/a	# or None if there is no such character.
619	n/a	FS_NONASCII = None
620	n/a	for character in (
621	n/a	# First try printable and common characters to have a readable filename.
622	n/a	# For each character, the encoding list are just example of encodings able
623	n/a	# to encode the character (the list is not exhaustive).
624	n/a
625	n/a	# U+00E6 (Latin Small Letter Ae): cp1252, iso-8859-1
626	n/a	'\u00E6',
627	n/a	# U+0130 (Latin Capital Letter I With Dot Above): cp1254, iso8859_3
628	n/a	'\u0130',
629	n/a	# U+0141 (Latin Capital Letter L With Stroke): cp1250, cp1257
630	n/a	'\u0141',
631	n/a	# U+03C6 (Greek Small Letter Phi): cp1253
632	n/a	'\u03C6',
633	n/a	# U+041A (Cyrillic Capital Letter Ka): cp1251
634	n/a	'\u041A',
635	n/a	# U+05D0 (Hebrew Letter Alef): Encodable to cp424
636	n/a	'\u05D0',
637	n/a	# U+060C (Arabic Comma): cp864, cp1006, iso8859_6, mac_arabic
638	n/a	'\u060C',
639	n/a	# U+062A (Arabic Letter Teh): cp720
640	n/a	'\u062A',
641	n/a	# U+0E01 (Thai Character Ko Kai): cp874
642	n/a	'\u0E01',
643	n/a
644	n/a	# Then try more "special" characters. "special" because they may be
645	n/a	# interpreted or displayed differently depending on the exact locale
646	n/a	# encoding and the font.
647	n/a
648	n/a	# U+00A0 (No-Break Space)
649	n/a	'\u00A0',
650	n/a	# U+20AC (Euro Sign)
651	n/a	'\u20AC',
652	n/a	):
653	n/a	try:
654	n/a	os.fsdecode(os.fsencode(character))
655	n/a	except UnicodeError:
656	n/a	pass
657	n/a	else:
658	n/a	FS_NONASCII = character
659	n/a	break
660	n/a
661	n/a	# TESTFN_UNICODE is a non-ascii filename
662	n/a	TESTFN_UNICODE = TESTFN + "-\xe0\xf2\u0258\u0141\u011f"
663	n/a	if sys.platform == 'darwin':
664	n/a	# In Mac OS X's VFS API file names are, by definition, canonically
665	n/a	# decomposed Unicode, encoded using UTF-8. See QA1173:
666	n/a	# http://developer.apple.com/mac/library/qa/qa2001/qa1173.html
667	n/a	import unicodedata
668	n/a	TESTFN_UNICODE = unicodedata.normalize('NFD', TESTFN_UNICODE)
669	n/a	TESTFN_ENCODING = sys.getfilesystemencoding()
670	n/a
671	n/a	# TESTFN_UNENCODABLE is a filename (str type) that should not be able to be
672	n/a	# encoded by the filesystem encoding (in strict mode). It can be None if we
673	n/a	# cannot generate such filename.
674	n/a	TESTFN_UNENCODABLE = None
675	n/a	if os.name in ('nt', 'ce'):
676	n/a	# skip win32s (0) or Windows 9x/ME (1)
677	n/a	if sys.getwindowsversion().platform >= 2:
678	n/a	# Different kinds of characters from various languages to minimize the
679	n/a	# probability that the whole name is encodable to MBCS (issue #9819)
680	n/a	TESTFN_UNENCODABLE = TESTFN + "-\u5171\u0141\u2661\u0363\uDC80"
681	n/a	try:
682	n/a	TESTFN_UNENCODABLE.encode(TESTFN_ENCODING)
683	n/a	except UnicodeEncodeError:
684	n/a	pass
685	n/a	else:
686	n/a	print('WARNING: The filename %r CAN be encoded by the filesystem encoding (%s). '
687	n/a	'Unicode filename tests may not be effective'
688	n/a	% (TESTFN_UNENCODABLE, TESTFN_ENCODING))
689	n/a	TESTFN_UNENCODABLE = None
690	n/a	# Mac OS X denies unencodable filenames (invalid utf-8)
691	n/a	elif sys.platform != 'darwin':
692	n/a	try:
693	n/a	# ascii and utf-8 cannot encode the byte 0xff
694	n/a	b'\xff'.decode(TESTFN_ENCODING)
695	n/a	except UnicodeDecodeError:
696	n/a	# 0xff will be encoded using the surrogate character u+DCFF
697	n/a	TESTFN_UNENCODABLE = TESTFN \
698	n/a	+ b'-\xff'.decode(TESTFN_ENCODING, 'surrogateescape')
699	n/a	else:
700	n/a	# File system encoding (eg. ISO-8859-* encodings) can encode
701	n/a	# the byte 0xff. Skip some unicode filename tests.
702	n/a	pass
703	n/a
704	n/a	# TESTFN_UNDECODABLE is a filename (bytes type) that should not be able to be
705	n/a	# decoded from the filesystem encoding (in strict mode). It can be None if we
706	n/a	# cannot generate such filename (ex: the latin1 encoding can decode any byte
707	n/a	# sequence). On UNIX, TESTFN_UNDECODABLE can be decoded by os.fsdecode() thanks
708	n/a	# to the surrogateescape error handler (PEP 383), but not from the filesystem
709	n/a	# encoding in strict mode.
710	n/a	TESTFN_UNDECODABLE = None
711	n/a	for name in (
712	n/a	# b'\xff' is not decodable by os.fsdecode() with code page 932. Windows
713	n/a	# accepts it to create a file or a directory, or don't accept to enter to
714	n/a	# such directory (when the bytes name is used). So test b'\xe7' first: it is
715	n/a	# not decodable from cp932.
716	n/a	b'\xe7w\xf0',
717	n/a	# undecodable from ASCII, UTF-8
718	n/a	b'\xff',
719	n/a	# undecodable from iso8859-3, iso8859-6, iso8859-7, cp424, iso8859-8, cp856
720	n/a	# and cp857
721	n/a	b'\xae\xd5'
722	n/a	# undecodable from UTF-8 (UNIX and Mac OS X)
723	n/a	b'\xed\xb2\x80', b'\xed\xb4\x80',
724	n/a	# undecodable from shift_jis, cp869, cp874, cp932, cp1250, cp1251, cp1252,
725	n/a	# cp1253, cp1254, cp1255, cp1257, cp1258
726	n/a	b'\x81\x98',
727	n/a	):
728	n/a	try:
729	n/a	name.decode(TESTFN_ENCODING)
730	n/a	except UnicodeDecodeError:
731	n/a	TESTFN_UNDECODABLE = os.fsencode(TESTFN) + name
732	n/a	break
733	n/a
734	n/a	if FS_NONASCII:
735	n/a	TESTFN_NONASCII = TESTFN + '-' + FS_NONASCII
736	n/a	else:
737	n/a	TESTFN_NONASCII = None
738	n/a
739	n/a	# Save the initial cwd
740	n/a	SAVEDCWD = os.getcwd()
741	n/a
742	n/a	@contextlib.contextmanager
743	n/a	def temp_cwd(name='tempcwd', quiet=False, path=None):
744	n/a	"""
745	n/a	Context manager that temporarily changes the CWD.
746	n/a
747	n/a	An existing path may be provided as path, in which case this
748	n/a	function makes no changes to the file system.
749	n/a
750	n/a	Otherwise, the new CWD is created in the current directory and it's
751	n/a	named name. If quiet is False (default) and it's not possible to
752	n/a	create or change the CWD, an error is raised. If it's True, only a
753	n/a	warning is raised and the original CWD is used.
754	n/a	"""
755	n/a	saved_dir = os.getcwd()
756	n/a	is_temporary = False
757	n/a	if path is None:
758	n/a	path = name
759	n/a	try:
760	n/a	os.mkdir(name)
761	n/a	is_temporary = True
762	n/a	except OSError:
763	n/a	if not quiet:
764	n/a	raise
765	n/a	warnings.warn('tests may fail, unable to create temp CWD ' + name,
766	n/a	RuntimeWarning, stacklevel=3)
767	n/a	try:
768	n/a	os.chdir(path)
769	n/a	except OSError:
770	n/a	if not quiet:
771	n/a	raise
772	n/a	warnings.warn('tests may fail, unable to change the CWD to ' + path,
773	n/a	RuntimeWarning, stacklevel=3)
774	n/a	try:
775	n/a	yield os.getcwd()
776	n/a	finally:
777	n/a	os.chdir(saved_dir)
778	n/a	if is_temporary:
779	n/a	rmtree(name)
780	n/a
781	n/a
782	n/a	if hasattr(os, "umask"):
783	n/a	@contextlib.contextmanager
784	n/a	def temp_umask(umask):
785	n/a	"""Context manager that temporarily sets the process umask."""
786	n/a	oldmask = os.umask(umask)
787	n/a	try:
788	n/a	yield
789	n/a	finally:
790	n/a	os.umask(oldmask)
791	n/a
792	n/a
793	n/a	def findfile(file, here=__file__, subdir=None):
794	n/a	"""Try to find a file on sys.path and the working directory. If it is not
795	n/a	found the argument passed to the function is returned (this does not
796	n/a	necessarily signal failure; could still be the legitimate path)."""
797	n/a	if os.path.isabs(file):
798	n/a	return file
799	n/a	if subdir is not None:
800	n/a	file = os.path.join(subdir, file)
801	n/a	path = sys.path
802	n/a	path = [os.path.dirname(here)] + path
803	n/a	for dn in path:
804	n/a	fn = os.path.join(dn, file)
805	n/a	if os.path.exists(fn): return fn
806	n/a	return file
807	n/a
808	n/a	def create_empty_file(filename):
809	n/a	"""Create an empty file. If the file already exists, truncate it."""
810	n/a	fd = os.open(filename, os.O_WRONLY \| os.O_CREAT \| os.O_TRUNC)
811	n/a	os.close(fd)
812	n/a
813	n/a	def sortdict(dict):
814	n/a	"Like repr(dict), but in sorted order."
815	n/a	items = sorted(dict.items())
816	n/a	reprpairs = ["%r: %r" % pair for pair in items]
817	n/a	withcommas = ", ".join(reprpairs)
818	n/a	return "{%s}" % withcommas
819	n/a
820	n/a	def make_bad_fd():
821	n/a	"""
822	n/a	Create an invalid file descriptor by opening and closing a file and return
823	n/a	its fd.
824	n/a	"""
825	n/a	file = open(TESTFN, "wb")
826	n/a	try:
827	n/a	return file.fileno()
828	n/a	finally:
829	n/a	file.close()
830	n/a	unlink(TESTFN)
831	n/a
832	n/a	def check_syntax_error(testcase, statement):
833	n/a	testcase.assertRaises(SyntaxError, compile, statement,
834	n/a	'<test string>', 'exec')
835	n/a
836	n/a	def open_urlresource(url, args, *kw):
837	n/a	import urllib.request, urllib.parse
838	n/a
839	n/a	check = kw.pop('check', None)
840	n/a
841	n/a	filename = urllib.parse.urlparse(url)[2].split('/')[-1] # '/': it's URL!
842	n/a
843	n/a	fn = os.path.join(os.path.dirname(__file__), "data", filename)
844	n/a
845	n/a	def check_valid_file(fn):
846	n/a	f = open(fn, args, *kw)
847	n/a	if check is None:
848	n/a	return f
849	n/a	elif check(f):
850	n/a	f.seek(0)
851	n/a	return f
852	n/a	f.close()
853	n/a
854	n/a	if os.path.exists(fn):
855	n/a	f = check_valid_file(fn)
856	n/a	if f is not None:
857	n/a	return f
858	n/a	unlink(fn)
859	n/a
860	n/a	# Verify the requirement before downloading the file
861	n/a	requires('urlfetch')
862	n/a
863	n/a	print('\tfetching %s ...' % url, file=get_original_stdout())
864	n/a	f = urllib.request.urlopen(url, timeout=15)
865	n/a	try:
866	n/a	with open(fn, "wb") as out:
867	n/a	s = f.read()
868	n/a	while s:
869	n/a	out.write(s)
870	n/a	s = f.read()
871	n/a	finally:
872	n/a	f.close()
873	n/a
874	n/a	f = check_valid_file(fn)
875	n/a	if f is not None:
876	n/a	return f
877	n/a	raise TestFailed('invalid resource %r' % fn)
878	n/a
879	n/a
880	n/a	class WarningsRecorder(object):
881	n/a	"""Convenience wrapper for the warnings list returned on
882	n/a	entry to the warnings.catch_warnings() context manager.
883	n/a	"""
884	n/a	def __init__(self, warnings_list):
885	n/a	self._warnings = warnings_list
886	n/a	self._last = 0
887	n/a
888	n/a	def __getattr__(self, attr):
889	n/a	if len(self._warnings) > self._last:
890	n/a	return getattr(self._warnings[-1], attr)
891	n/a	elif attr in warnings.WarningMessage._WARNING_DETAILS:
892	n/a	return None
893	n/a	raise AttributeError("%r has no attribute %r" % (self, attr))
894	n/a
895	n/a	@property
896	n/a	def warnings(self):
897	n/a	return self._warnings[self._last:]
898	n/a
899	n/a	def reset(self):
900	n/a	self._last = len(self._warnings)
901	n/a
902	n/a
903	n/a	def _filterwarnings(filters, quiet=False):
904	n/a	"""Catch the warnings, then check if all the expected
905	n/a	warnings have been raised and re-raise unexpected warnings.
906	n/a	If 'quiet' is True, only re-raise the unexpected warnings.
907	n/a	"""
908	n/a	# Clear the warning registry of the calling module
909	n/a	# in order to re-raise the warnings.
910	n/a	frame = sys._getframe(2)
911	n/a	registry = frame.f_globals.get('__warningregistry__')
912	n/a	if registry:
913	n/a	registry.clear()
914	n/a	with warnings.catch_warnings(record=True) as w:
915	n/a	# Set filter "always" to record all warnings. Because
916	n/a	# test_warnings swap the module, we need to look up in
917	n/a	# the sys.modules dictionary.
918	n/a	sys.modules['warnings'].simplefilter("always")
919	n/a	yield WarningsRecorder(w)
920	n/a	# Filter the recorded warnings
921	n/a	reraise = list(w)
922	n/a	missing = []
923	n/a	for msg, cat in filters:
924	n/a	seen = False
925	n/a	for w in reraise[:]:
926	n/a	warning = w.message
927	n/a	# Filter out the matching messages
928	n/a	if (re.match(msg, str(warning), re.I) and
929	n/a	issubclass(warning.__class__, cat)):
930	n/a	seen = True
931	n/a	reraise.remove(w)
932	n/a	if not seen and not quiet:
933	n/a	# This filter caught nothing
934	n/a	missing.append((msg, cat.__name__))
935	n/a	if reraise:
936	n/a	raise AssertionError("unhandled warning %s" % reraise[0])
937	n/a	if missing:
938	n/a	raise AssertionError("filter (%r, %s) did not catch any warning" %
939	n/a	missing[0])
940	n/a
941	n/a
942	n/a	@contextlib.contextmanager
943	n/a	def check_warnings(filters, *kwargs):
944	n/a	"""Context manager to silence warnings.
945	n/a
946	n/a	Accept 2-tuples as positional arguments:
947	n/a	("message regexp", WarningCategory)
948	n/a
949	n/a	Optional argument:
950	n/a	- if 'quiet' is True, it does not fail if a filter catches nothing
951	n/a	(default True without argument,
952	n/a	default False if some filters are defined)
953	n/a
954	n/a	Without argument, it defaults to:
955	n/a	check_warnings(("", Warning), quiet=True)
956	n/a	"""
957	n/a	quiet = kwargs.get('quiet')
958	n/a	if not filters:
959	n/a	filters = (("", Warning),)
960	n/a	# Preserve backward compatibility
961	n/a	if quiet is None:
962	n/a	quiet = True
963	n/a	return _filterwarnings(filters, quiet)
964	n/a
965	n/a
966	n/a	class CleanImport(object):
967	n/a	"""Context manager to force import to return a new module reference.
968	n/a
969	n/a	This is useful for testing module-level behaviours, such as
970	n/a	the emission of a DeprecationWarning on import.
971	n/a
972	n/a	Use like this:
973	n/a
974	n/a	with CleanImport("foo"):
975	n/a	importlib.import_module("foo") # new reference
976	n/a	"""
977	n/a
978	n/a	def __init__(self, *module_names):
979	n/a	self.original_modules = sys.modules.copy()
980	n/a	for module_name in module_names:
981	n/a	if module_name in sys.modules:
982	n/a	module = sys.modules[module_name]
983	n/a	# It is possible that module_name is just an alias for
984	n/a	# another module (e.g. stub for modules renamed in 3.x).
985	n/a	# In that case, we also need delete the real module to clear
986	n/a	# the import cache.
987	n/a	if module.__name__ != module_name:
988	n/a	del sys.modules[module.__name__]
989	n/a	del sys.modules[module_name]
990	n/a
991	n/a	def __enter__(self):
992	n/a	return self
993	n/a
994	n/a	def __exit__(self, *ignore_exc):
995	n/a	sys.modules.update(self.original_modules)
996	n/a
997	n/a
998	n/a	class EnvironmentVarGuard(collections.abc.MutableMapping):
999	n/a
1000	n/a	"""Class to help protect the environment variable properly. Can be used as
1001	n/a	a context manager."""
1002	n/a
1003	n/a	def __init__(self):
1004	n/a	self._environ = os.environ
1005	n/a	self._changed = {}
1006	n/a
1007	n/a	def __getitem__(self, envvar):
1008	n/a	return self._environ[envvar]
1009	n/a
1010	n/a	def __setitem__(self, envvar, value):
1011	n/a	# Remember the initial value on the first access
1012	n/a	if envvar not in self._changed:
1013	n/a	self._changed[envvar] = self._environ.get(envvar)
1014	n/a	self._environ[envvar] = value
1015	n/a
1016	n/a	def __delitem__(self, envvar):
1017	n/a	# Remember the initial value on the first access
1018	n/a	if envvar not in self._changed:
1019	n/a	self._changed[envvar] = self._environ.get(envvar)
1020	n/a	if envvar in self._environ:
1021	n/a	del self._environ[envvar]
1022	n/a
1023	n/a	def keys(self):
1024	n/a	return self._environ.keys()
1025	n/a
1026	n/a	def __iter__(self):
1027	n/a	return iter(self._environ)
1028	n/a
1029	n/a	def __len__(self):
1030	n/a	return len(self._environ)
1031	n/a
1032	n/a	def set(self, envvar, value):
1033	n/a	self[envvar] = value
1034	n/a
1035	n/a	def unset(self, envvar):
1036	n/a	del self[envvar]
1037	n/a
1038	n/a	def __enter__(self):
1039	n/a	return self
1040	n/a
1041	n/a	def __exit__(self, *ignore_exc):
1042	n/a	for (k, v) in self._changed.items():
1043	n/a	if v is None:
1044	n/a	if k in self._environ:
1045	n/a	del self._environ[k]
1046	n/a	else:
1047	n/a	self._environ[k] = v
1048	n/a	os.environ = self._environ
1049	n/a
1050	n/a
1051	n/a	class DirsOnSysPath(object):
1052	n/a	"""Context manager to temporarily add directories to sys.path.
1053	n/a
1054	n/a	This makes a copy of sys.path, appends any directories given
1055	n/a	as positional arguments, then reverts sys.path to the copied
1056	n/a	settings when the context ends.
1057	n/a
1058	n/a	Note that all sys.path modifications in the body of the
1059	n/a	context manager, including replacement of the object,
1060	n/a	will be reverted at the end of the block.
1061	n/a	"""
1062	n/a
1063	n/a	def __init__(self, *paths):
1064	n/a	self.original_value = sys.path[:]
1065	n/a	self.original_object = sys.path
1066	n/a	sys.path.extend(paths)
1067	n/a
1068	n/a	def __enter__(self):
1069	n/a	return self
1070	n/a
1071	n/a	def __exit__(self, *ignore_exc):
1072	n/a	sys.path = self.original_object
1073	n/a	sys.path[:] = self.original_value
1074	n/a
1075	n/a
1076	n/a	class TransientResource(object):
1077	n/a
1078	n/a	"""Raise ResourceDenied if an exception is raised while the context manager
1079	n/a	is in effect that matches the specified exception and attributes."""
1080	n/a
1081	n/a	def __init__(self, exc, **kwargs):
1082	n/a	self.exc = exc
1083	n/a	self.attrs = kwargs
1084	n/a
1085	n/a	def __enter__(self):
1086	n/a	return self
1087	n/a
1088	n/a	def __exit__(self, type_=None, value=None, traceback=None):
1089	n/a	"""If type_ is a subclass of self.exc and value has attributes matching
1090	n/a	self.attrs, raise ResourceDenied. Otherwise let the exception
1091	n/a	propagate (if any)."""
1092	n/a	if type_ is not None and issubclass(self.exc, type_):
1093	n/a	for attr, attr_value in self.attrs.items():
1094	n/a	if not hasattr(value, attr):
1095	n/a	break
1096	n/a	if getattr(value, attr) != attr_value:
1097	n/a	break
1098	n/a	else:
1099	n/a	raise ResourceDenied("an optional resource is not available")
1100	n/a
1101	n/a	# Context managers that raise ResourceDenied when various issues
1102	n/a	# with the Internet connection manifest themselves as exceptions.
1103	n/a	# XXX deprecate these and use transient_internet() instead
1104	n/a	time_out = TransientResource(OSError, errno=errno.ETIMEDOUT)
1105	n/a	socket_peer_reset = TransientResource(OSError, errno=errno.ECONNRESET)
1106	n/a	ioerror_peer_reset = TransientResource(OSError, errno=errno.ECONNRESET)
1107	n/a
1108	n/a
1109	n/a	@contextlib.contextmanager
1110	n/a	def transient_internet(resource_name, *, timeout=30.0, errnos=()):
1111	n/a	"""Return a context manager that raises ResourceDenied when various issues
1112	n/a	with the Internet connection manifest themselves as exceptions."""
1113	n/a	default_errnos = [
1114	n/a	('ECONNREFUSED', 111),
1115	n/a	('ECONNRESET', 104),
1116	n/a	('EHOSTUNREACH', 113),
1117	n/a	('ENETUNREACH', 101),
1118	n/a	('ETIMEDOUT', 110),
1119	n/a	]
1120	n/a	default_gai_errnos = [
1121	n/a	('EAI_AGAIN', -3),
1122	n/a	('EAI_FAIL', -4),
1123	n/a	('EAI_NONAME', -2),
1124	n/a	('EAI_NODATA', -5),
1125	n/a	# Encountered when trying to resolve IPv6-only hostnames
1126	n/a	('WSANO_DATA', 11004),
1127	n/a	]
1128	n/a
1129	n/a	denied = ResourceDenied("Resource %r is not available" % resource_name)
1130	n/a	captured_errnos = errnos
1131	n/a	gai_errnos = []
1132	n/a	if not captured_errnos:
1133	n/a	captured_errnos = [getattr(errno, name, num)
1134	n/a	for (name, num) in default_errnos]
1135	n/a	gai_errnos = [getattr(socket, name, num)
1136	n/a	for (name, num) in default_gai_errnos]
1137	n/a
1138	n/a	def filter_error(err):
1139	n/a	n = getattr(err, 'errno', None)
1140	n/a	if (isinstance(err, socket.timeout) or
1141	n/a	(isinstance(err, socket.gaierror) and n in gai_errnos) or
1142	n/a	n in captured_errnos):
1143	n/a	if not verbose:
1144	n/a	sys.stderr.write(denied.args[0] + "\n")
1145	n/a	raise denied from err
1146	n/a
1147	n/a	old_timeout = socket.getdefaulttimeout()
1148	n/a	try:
1149	n/a	if timeout is not None:
1150	n/a	socket.setdefaulttimeout(timeout)
1151	n/a	yield
1152	n/a	except OSError as err:
1153	n/a	# urllib can wrap original socket errors multiple times (!), we must
1154	n/a	# unwrap to get at the original error.
1155	n/a	while True:
1156	n/a	a = err.args
1157	n/a	if len(a) >= 1 and isinstance(a[0], OSError):
1158	n/a	err = a[0]
1159	n/a	# The error can also be wrapped as args[1]:
1160	n/a	# except socket.error as msg:
1161	n/a	# raise OSError('socket error', msg).with_traceback(sys.exc_info()[2])
1162	n/a	elif len(a) >= 2 and isinstance(a[1], OSError):
1163	n/a	err = a[1]
1164	n/a	else:
1165	n/a	break
1166	n/a	filter_error(err)
1167	n/a	raise
1168	n/a	# XXX should we catch generic exceptions and look for their
1169	n/a	# __cause__ or __context__?
1170	n/a	finally:
1171	n/a	socket.setdefaulttimeout(old_timeout)
1172	n/a
1173	n/a
1174	n/a	@contextlib.contextmanager
1175	n/a	def captured_output(stream_name):
1176	n/a	"""Return a context manager used by captured_stdout/stdin/stderr
1177	n/a	that temporarily replaces the sys stream stream_name with a StringIO."""
1178	n/a	import io
1179	n/a	orig_stdout = getattr(sys, stream_name)
1180	n/a	setattr(sys, stream_name, io.StringIO())
1181	n/a	try:
1182	n/a	yield getattr(sys, stream_name)
1183	n/a	finally:
1184	n/a	setattr(sys, stream_name, orig_stdout)
1185	n/a
1186	n/a	def captured_stdout():
1187	n/a	"""Capture the output of sys.stdout:
1188	n/a
1189	n/a	with captured_stdout() as stdout:
1190	n/a	print("hello")
1191	n/a	self.assertEqual(stdout.getvalue(), "hello\n")
1192	n/a	"""
1193	n/a	return captured_output("stdout")
1194	n/a
1195	n/a	def captured_stderr():
1196	n/a	"""Capture the output of sys.stderr:
1197	n/a
1198	n/a	with captured_stderr() as stderr:
1199	n/a	print("hello", file=sys.stderr)
1200	n/a	self.assertEqual(stderr.getvalue(), "hello\n")
1201	n/a	"""
1202	n/a	return captured_output("stderr")
1203	n/a
1204	n/a	def captured_stdin():
1205	n/a	"""Capture the input to sys.stdin:
1206	n/a
1207	n/a	with captured_stdin() as stdin:
1208	n/a	stdin.write('hello\n')
1209	n/a	stdin.seek(0)
1210	n/a	# call test code that consumes from sys.stdin
1211	n/a	captured = input()
1212	n/a	self.assertEqual(captured, "hello")
1213	n/a	"""
1214	n/a	return captured_output("stdin")
1215	n/a
1216	n/a
1217	n/a	def gc_collect():
1218	n/a	"""Force as many objects as possible to be collected.
1219	n/a
1220	n/a	In non-CPython implementations of Python, this is needed because timely
1221	n/a	deallocation is not guaranteed by the garbage collector. (Even in CPython
1222	n/a	this can be the case in case of reference cycles.) This means that __del__
1223	n/a	methods may be called later than expected and weakrefs may remain alive for
1224	n/a	longer than expected. This function tries its best to force all garbage
1225	n/a	objects to disappear.
1226	n/a	"""
1227	n/a	gc.collect()
1228	n/a	if is_jython:
1229	n/a	time.sleep(0.1)
1230	n/a	gc.collect()
1231	n/a	gc.collect()
1232	n/a
1233	n/a	@contextlib.contextmanager
1234	n/a	def disable_gc():
1235	n/a	have_gc = gc.isenabled()
1236	n/a	gc.disable()
1237	n/a	try:
1238	n/a	yield
1239	n/a	finally:
1240	n/a	if have_gc:
1241	n/a	gc.enable()
1242	n/a
1243	n/a
1244	n/a	def python_is_optimized():
1245	n/a	"""Find if Python was built with optimizations."""
1246	n/a	cflags = sysconfig.get_config_var('PY_CFLAGS') or ''
1247	n/a	final_opt = ""
1248	n/a	for opt in cflags.split():
1249	n/a	if opt.startswith('-O'):
1250	n/a	final_opt = opt
1251	n/a	return final_opt != '' and final_opt != '-O0'
1252	n/a
1253	n/a
1254	n/a	_header = 'nP'
1255	n/a	_align = '0n'
1256	n/a	if hasattr(sys, "gettotalrefcount"):
1257	n/a	_header = '2P' + _header
1258	n/a	_align = '0P'
1259	n/a	_vheader = _header + 'n'
1260	n/a
1261	n/a	def calcobjsize(fmt):
1262	n/a	return struct.calcsize(_header + fmt + _align)
1263	n/a
1264	n/a	def calcvobjsize(fmt):
1265	n/a	return struct.calcsize(_vheader + fmt + _align)
1266	n/a
1267	n/a
1268	n/a	_TPFLAGS_HAVE_GC = 1<<14
1269	n/a	_TPFLAGS_HEAPTYPE = 1<<9
1270	n/a
1271	n/a	def check_sizeof(test, o, size):
1272	n/a	result = sys.getsizeof(o)
1273	n/a	# add GC header size
1274	n/a	if ((type(o) == type) and (o.__flags__ & _TPFLAGS_HEAPTYPE) or\
1275	n/a	((type(o) != type) and (type(o).__flags__ & _TPFLAGS_HAVE_GC))):
1276	n/a	size += _testcapi.SIZEOF_PYGC_HEAD
1277	n/a	msg = 'wrong size for %s: got %d, expected %d' \
1278	n/a	% (type(o), result, size)
1279	n/a	test.assertEqual(result, size, msg)
1280	n/a
1281	n/a	#=======================================================================
1282	n/a	# Decorator for running a function in a different locale, correctly resetting
1283	n/a	# it afterwards.
1284	n/a
1285	n/a	def run_with_locale(catstr, *locales):
1286	n/a	def decorator(func):
1287	n/a	def inner(args, *kwds):
1288	n/a	try:
1289	n/a	import locale
1290	n/a	category = getattr(locale, catstr)
1291	n/a	orig_locale = locale.setlocale(category)
1292	n/a	except AttributeError:
1293	n/a	# if the test author gives us an invalid category string
1294	n/a	raise
1295	n/a	except:
1296	n/a	# cannot retrieve original locale, so do nothing
1297	n/a	locale = orig_locale = None
1298	n/a	else:
1299	n/a	for loc in locales:
1300	n/a	try:
1301	n/a	locale.setlocale(category, loc)
1302	n/a	break
1303	n/a	except:
1304	n/a	pass
1305	n/a
1306	n/a	# now run the function, resetting the locale on exceptions
1307	n/a	try:
1308	n/a	return func(args, *kwds)
1309	n/a	finally:
1310	n/a	if locale and orig_locale:
1311	n/a	locale.setlocale(category, orig_locale)
1312	n/a	inner.__name__ = func.__name__
1313	n/a	inner.__doc__ = func.__doc__
1314	n/a	return inner
1315	n/a	return decorator
1316	n/a
1317	n/a	#=======================================================================
1318	n/a	# Decorator for running a function in a specific timezone, correctly
1319	n/a	# resetting it afterwards.
1320	n/a
1321	n/a	def run_with_tz(tz):
1322	n/a	def decorator(func):
1323	n/a	def inner(args, *kwds):
1324	n/a	try:
1325	n/a	tzset = time.tzset
1326	n/a	except AttributeError:
1327	n/a	raise unittest.SkipTest("tzset required")
1328	n/a	if 'TZ' in os.environ:
1329	n/a	orig_tz = os.environ['TZ']
1330	n/a	else:
1331	n/a	orig_tz = None
1332	n/a	os.environ['TZ'] = tz
1333	n/a	tzset()
1334	n/a
1335	n/a	# now run the function, resetting the tz on exceptions
1336	n/a	try:
1337	n/a	return func(args, *kwds)
1338	n/a	finally:
1339	n/a	if orig_tz is None:
1340	n/a	del os.environ['TZ']
1341	n/a	else:
1342	n/a	os.environ['TZ'] = orig_tz
1343	n/a	time.tzset()
1344	n/a
1345	n/a	inner.__name__ = func.__name__
1346	n/a	inner.__doc__ = func.__doc__
1347	n/a	return inner
1348	n/a	return decorator
1349	n/a
1350	n/a	#=======================================================================
1351	n/a	# Big-memory-test support. Separate from 'resources' because memory use
1352	n/a	# should be configurable.
1353	n/a
1354	n/a	# Some handy shorthands. Note that these are used for byte-limits as well
1355	n/a	# as size-limits, in the various bigmem tests
1356	n/a	_1M = 1024*1024
1357	n/a	_1G = 1024 * _1M
1358	n/a	_2G = 2 * _1G
1359	n/a	_4G = 4 * _1G
1360	n/a
1361	n/a	MAX_Py_ssize_t = sys.maxsize
1362	n/a
1363	n/a	def set_memlimit(limit):
1364	n/a	global max_memuse
1365	n/a	global real_max_memuse
1366	n/a	sizes = {
1367	n/a	'k': 1024,
1368	n/a	'm': _1M,
1369	n/a	'g': _1G,
1370	n/a	't': 1024*_1G,
1371	n/a	}
1372	n/a	m = re.match(r'(\d+(\.\d+)?) (K\|M\|G\|T)b?$', limit,
1373	n/a	re.IGNORECASE \| re.VERBOSE)
1374	n/a	if m is None:
1375	n/a	raise ValueError('Invalid memory limit %r' % (limit,))
1376	n/a	memlimit = int(float(m.group(1)) * sizes[m.group(3).lower()])
1377	n/a	real_max_memuse = memlimit
1378	n/a	if memlimit > MAX_Py_ssize_t:
1379	n/a	memlimit = MAX_Py_ssize_t
1380	n/a	if memlimit < _2G - 1:
1381	n/a	raise ValueError('Memory limit %r too low to be useful' % (limit,))
1382	n/a	max_memuse = memlimit
1383	n/a
1384	n/a	class _MemoryWatchdog:
1385	n/a	"""An object which periodically watches the process' memory consumption
1386	n/a	and prints it out.
1387	n/a	"""
1388	n/a
1389	n/a	def __init__(self):
1390	n/a	self.procfile = '/proc/{pid}/statm'.format(pid=os.getpid())
1391	n/a	self.started = False
1392	n/a
1393	n/a	def start(self):
1394	n/a	try:
1395	n/a	f = open(self.procfile, 'r')
1396	n/a	except OSError as e:
1397	n/a	warnings.warn('/proc not available for stats: {}'.format(e),
1398	n/a	RuntimeWarning)
1399	n/a	sys.stderr.flush()
1400	n/a	return
1401	n/a
1402	n/a	watchdog_script = findfile("memory_watchdog.py")
1403	n/a	self.mem_watchdog = subprocess.Popen([sys.executable, watchdog_script],
1404	n/a	stdin=f, stderr=subprocess.DEVNULL)
1405	n/a	f.close()
1406	n/a	self.started = True
1407	n/a
1408	n/a	def stop(self):
1409	n/a	if self.started:
1410	n/a	self.mem_watchdog.terminate()
1411	n/a	self.mem_watchdog.wait()
1412	n/a
1413	n/a
1414	n/a	def bigmemtest(size, memuse, dry_run=True):
1415	n/a	"""Decorator for bigmem tests.
1416	n/a
1417	n/a	'minsize' is the minimum useful size for the test (in arbitrary,
1418	n/a	test-interpreted units.) 'memuse' is the number of 'bytes per size' for
1419	n/a	the test, or a good estimate of it.
1420	n/a
1421	n/a	if 'dry_run' is False, it means the test doesn't support dummy runs
1422	n/a	when -M is not specified.
1423	n/a	"""
1424	n/a	def decorator(f):
1425	n/a	def wrapper(self):
1426	n/a	size = wrapper.size
1427	n/a	memuse = wrapper.memuse
1428	n/a	if not real_max_memuse:
1429	n/a	maxsize = 5147
1430	n/a	else:
1431	n/a	maxsize = size
1432	n/a
1433	n/a	if ((real_max_memuse or not dry_run)
1434	n/a	and real_max_memuse < maxsize * memuse):
1435	n/a	raise unittest.SkipTest(
1436	n/a	"not enough memory: %.1fG minimum needed"
1437	n/a	% (size * memuse / (1024 ** 3)))
1438	n/a
1439	n/a	if real_max_memuse and verbose:
1440	n/a	print()
1441	n/a	print(" ... expected peak memory use: {peak:.1f}G"
1442	n/a	.format(peak=size * memuse / (1024 ** 3)))
1443	n/a	watchdog = _MemoryWatchdog()
1444	n/a	watchdog.start()
1445	n/a	else:
1446	n/a	watchdog = None
1447	n/a
1448	n/a	try:
1449	n/a	return f(self, maxsize)
1450	n/a	finally:
1451	n/a	if watchdog:
1452	n/a	watchdog.stop()
1453	n/a
1454	n/a	wrapper.size = size
1455	n/a	wrapper.memuse = memuse
1456	n/a	return wrapper
1457	n/a	return decorator
1458	n/a
1459	n/a	def bigaddrspacetest(f):
1460	n/a	"""Decorator for tests that fill the address space."""
1461	n/a	def wrapper(self):
1462	n/a	if max_memuse < MAX_Py_ssize_t:
1463	n/a	if MAX_Py_ssize_t >= 263 - 1 and max_memuse >= 231:
1464	n/a	raise unittest.SkipTest(
1465	n/a	"not enough memory: try a 32-bit build instead")
1466	n/a	else:
1467	n/a	raise unittest.SkipTest(
1468	n/a	"not enough memory: %.1fG minimum needed"
1469	n/a	% (MAX_Py_ssize_t / (1024 ** 3)))
1470	n/a	else:
1471	n/a	return f(self)
1472	n/a	return wrapper
1473	n/a
1474	n/a	#=======================================================================
1475	n/a	# unittest integration.
1476	n/a
1477	n/a	class BasicTestRunner:
1478	n/a	def run(self, test):
1479	n/a	result = unittest.TestResult()
1480	n/a	test(result)
1481	n/a	return result
1482	n/a
1483	n/a	def _id(obj):
1484	n/a	return obj
1485	n/a
1486	n/a	def requires_resource(resource):
1487	n/a	if resource == 'gui' and not _is_gui_available():
1488	n/a	return unittest.skip("resource 'gui' is not available")
1489	n/a	if is_resource_enabled(resource):
1490	n/a	return _id
1491	n/a	else:
1492	n/a	return unittest.skip("resource {0!r} is not enabled".format(resource))
1493	n/a
1494	n/a	def cpython_only(test):
1495	n/a	"""
1496	n/a	Decorator for tests only applicable on CPython.
1497	n/a	"""
1498	n/a	return impl_detail(cpython=True)(test)
1499	n/a
1500	n/a	def impl_detail(msg=None, **guards):
1501	n/a	if check_impl_detail(**guards):
1502	n/a	return _id
1503	n/a	if msg is None:
1504	n/a	guardnames, default = _parse_guards(guards)
1505	n/a	if default:
1506	n/a	msg = "implementation detail not available on {0}"
1507	n/a	else:
1508	n/a	msg = "implementation detail specific to {0}"
1509	n/a	guardnames = sorted(guardnames.keys())
1510	n/a	msg = msg.format(' or '.join(guardnames))
1511	n/a	return unittest.skip(msg)
1512	n/a
1513	n/a	def _parse_guards(guards):
1514	n/a	# Returns a tuple ({platform_name: run_me}, default_value)
1515	n/a	if not guards:
1516	n/a	return ({'cpython': True}, False)
1517	n/a	is_true = list(guards.values())[0]
1518	n/a	assert list(guards.values()) == [is_true] * len(guards) # all True or all False
1519	n/a	return (guards, not is_true)
1520	n/a
1521	n/a	# Use the following check to guard CPython's implementation-specific tests --
1522	n/a	# or to run them only on the implementation(s) guarded by the arguments.
1523	n/a	def check_impl_detail(**guards):
1524	n/a	"""This function returns True or False depending on the host platform.
1525	n/a	Examples:
1526	n/a	if check_impl_detail(): # only on CPython (default)
1527	n/a	if check_impl_detail(jython=True): # only on Jython
1528	n/a	if check_impl_detail(cpython=False): # everywhere except on CPython
1529	n/a	"""
1530	n/a	guards, default = _parse_guards(guards)
1531	n/a	return guards.get(platform.python_implementation().lower(), default)
1532	n/a
1533	n/a
1534	n/a	def no_tracing(func):
1535	n/a	"""Decorator to temporarily turn off tracing for the duration of a test."""
1536	n/a	if not hasattr(sys, 'gettrace'):
1537	n/a	return func
1538	n/a	else:
1539	n/a	@functools.wraps(func)
1540	n/a	def wrapper(args, *kwargs):
1541	n/a	original_trace = sys.gettrace()
1542	n/a	try:
1543	n/a	sys.settrace(None)
1544	n/a	return func(args, *kwargs)
1545	n/a	finally:
1546	n/a	sys.settrace(original_trace)
1547	n/a	return wrapper
1548	n/a
1549	n/a
1550	n/a	def refcount_test(test):
1551	n/a	"""Decorator for tests which involve reference counting.
1552	n/a
1553	n/a	To start, the decorator does not run the test if is not run by CPython.
1554	n/a	After that, any trace function is unset during the test to prevent
1555	n/a	unexpected refcounts caused by the trace function.
1556	n/a
1557	n/a	"""
1558	n/a	return no_tracing(cpython_only(test))
1559	n/a
1560	n/a
1561	n/a	def _filter_suite(suite, pred):
1562	n/a	"""Recursively filter test cases in a suite based on a predicate."""
1563	n/a	newtests = []
1564	n/a	for test in suite._tests:
1565	n/a	if isinstance(test, unittest.TestSuite):
1566	n/a	_filter_suite(test, pred)
1567	n/a	newtests.append(test)
1568	n/a	else:
1569	n/a	if pred(test):
1570	n/a	newtests.append(test)
1571	n/a	suite._tests = newtests
1572	n/a
1573	n/a	def _run_suite(suite):
1574	n/a	"""Run tests from a unittest.TestSuite-derived class."""
1575	n/a	if verbose:
1576	n/a	runner = unittest.TextTestRunner(sys.stdout, verbosity=2,
1577	n/a	failfast=failfast)
1578	n/a	else:
1579	n/a	runner = BasicTestRunner()
1580	n/a
1581	n/a	result = runner.run(suite)
1582	n/a	if not result.wasSuccessful():
1583	n/a	if len(result.errors) == 1 and not result.failures:
1584	n/a	err = result.errors[0][1]
1585	n/a	elif len(result.failures) == 1 and not result.errors:
1586	n/a	err = result.failures[0][1]
1587	n/a	else:
1588	n/a	err = "multiple errors occurred"
1589	n/a	if not verbose: err += "; run in verbose mode for details"
1590	n/a	raise TestFailed(err)
1591	n/a
1592	n/a
1593	n/a	def run_unittest(*classes):
1594	n/a	"""Run tests from unittest.TestCase-derived classes."""
1595	n/a	valid_types = (unittest.TestSuite, unittest.TestCase)
1596	n/a	suite = unittest.TestSuite()
1597	n/a	for cls in classes:
1598	n/a	if isinstance(cls, str):
1599	n/a	if cls in sys.modules:
1600	n/a	suite.addTest(unittest.findTestCases(sys.modules[cls]))
1601	n/a	else:
1602	n/a	raise ValueError("str arguments must be keys in sys.modules")
1603	n/a	elif isinstance(cls, valid_types):
1604	n/a	suite.addTest(cls)
1605	n/a	else:
1606	n/a	suite.addTest(unittest.makeSuite(cls))
1607	n/a	def case_pred(test):
1608	n/a	if match_tests is None:
1609	n/a	return True
1610	n/a	for name in test.id().split("."):
1611	n/a	if fnmatch.fnmatchcase(name, match_tests):
1612	n/a	return True
1613	n/a	return False
1614	n/a	_filter_suite(suite, case_pred)
1615	n/a	_run_suite(suite)
1616	n/a
1617	n/a	#=======================================================================
1618	n/a	# Check for the presence of docstrings.
1619	n/a
1620	n/a	HAVE_DOCSTRINGS = (check_impl_detail(cpython=False) or
1621	n/a	sys.platform == 'win32' or
1622	n/a	sysconfig.get_config_var('WITH_DOC_STRINGS'))
1623	n/a
1624	n/a	requires_docstrings = unittest.skipUnless(HAVE_DOCSTRINGS,
1625	n/a	"test requires docstrings")
1626	n/a
1627	n/a
1628	n/a	#=======================================================================
1629	n/a	# doctest driver.
1630	n/a
1631	n/a	def run_doctest(module, verbosity=None, optionflags=0):
1632	n/a	"""Run doctest on the given module. Return (#failures, #tests).
1633	n/a
1634	n/a	If optional argument verbosity is not specified (or is None), pass
1635	n/a	support's belief about verbosity on to doctest. Else doctest's
1636	n/a	usual behavior is used (it searches sys.argv for -v).
1637	n/a	"""
1638	n/a
1639	n/a	import doctest
1640	n/a
1641	n/a	if verbosity is None:
1642	n/a	verbosity = verbose
1643	n/a	else:
1644	n/a	verbosity = None
1645	n/a
1646	n/a	f, t = doctest.testmod(module, verbose=verbosity, optionflags=optionflags)
1647	n/a	if f:
1648	n/a	raise TestFailed("%d of %d doctests failed" % (f, t))
1649	n/a	if verbose:
1650	n/a	print('doctest (%s) ... %d tests with zero failures' %
1651	n/a	(module.__name__, t))
1652	n/a	return f, t
1653	n/a
1654	n/a
1655	n/a	#=======================================================================
1656	n/a	# Support for saving and restoring the imported modules.
1657	n/a
1658	n/a	def modules_setup():
1659	n/a	return sys.modules.copy(),
1660	n/a
1661	n/a	def modules_cleanup(oldmodules):
1662	n/a	# Encoders/decoders are registered permanently within the internal
1663	n/a	# codec cache. If we destroy the corresponding modules their
1664	n/a	# globals will be set to None which will trip up the cached functions.
1665	n/a	encodings = [(k, v) for k, v in sys.modules.items()
1666	n/a	if k.startswith('encodings.')]
1667	n/a	sys.modules.clear()
1668	n/a	sys.modules.update(encodings)
1669	n/a	# XXX: This kind of problem can affect more than just encodings. In particular
1670	n/a	# extension modules (such as _ssl) don't cope with reloading properly.
1671	n/a	# Really, test modules should be cleaning out the test specific modules they
1672	n/a	# know they added (ala test_runpy) rather than relying on this function (as
1673	n/a	# test_importhooks and test_pkg do currently).
1674	n/a	# Implicitly imported real modules should be left alone (see issue 10556).
1675	n/a	sys.modules.update(oldmodules)
1676	n/a
1677	n/a	#=======================================================================
1678	n/a	# Threading support to prevent reporting refleaks when running regrtest.py -R
1679	n/a
1680	n/a	# NOTE: we use thread._count() rather than threading.enumerate() (or the
1681	n/a	# moral equivalent thereof) because a threading.Thread object is still alive
1682	n/a	# until its __bootstrap() method has returned, even after it has been
1683	n/a	# unregistered from the threading module.
1684	n/a	# thread._count(), on the other hand, only gets decremented after the
1685	n/a	# __bootstrap() method has returned, which gives us reliable reference counts
1686	n/a	# at the end of a test run.
1687	n/a
1688	n/a	def threading_setup():
1689	n/a	if _thread:
1690	n/a	return _thread._count(), threading._dangling.copy()
1691	n/a	else:
1692	n/a	return 1, ()
1693	n/a
1694	n/a	def threading_cleanup(*original_values):
1695	n/a	if not _thread:
1696	n/a	return
1697	n/a	_MAX_COUNT = 10
1698	n/a	for count in range(_MAX_COUNT):
1699	n/a	values = _thread._count(), threading._dangling
1700	n/a	if values == original_values:
1701	n/a	break
1702	n/a	time.sleep(0.1)
1703	n/a	gc_collect()
1704	n/a	# XXX print a warning in case of failure?
1705	n/a
1706	n/a	def reap_threads(func):
1707	n/a	"""Use this function when threads are being used. This will
1708	n/a	ensure that the threads are cleaned up even when the test fails.
1709	n/a	If threading is unavailable this function does nothing.
1710	n/a	"""
1711	n/a	if not _thread:
1712	n/a	return func
1713	n/a
1714	n/a	@functools.wraps(func)
1715	n/a	def decorator(*args):
1716	n/a	key = threading_setup()
1717	n/a	try:
1718	n/a	return func(*args)
1719	n/a	finally:
1720	n/a	threading_cleanup(*key)
1721	n/a	return decorator
1722	n/a
1723	n/a	def reap_children():
1724	n/a	"""Use this function at the end of test_main() whenever sub-processes
1725	n/a	are started. This will help ensure that no extra children (zombies)
1726	n/a	stick around to hog resources and create problems when looking
1727	n/a	for refleaks.
1728	n/a	"""
1729	n/a
1730	n/a	# Reap all our dead child processes so we don't leave zombies around.
1731	n/a	# These hog resources and might be causing some of the buildbots to die.
1732	n/a	if hasattr(os, 'waitpid'):
1733	n/a	any_process = -1
1734	n/a	while True:
1735	n/a	try:
1736	n/a	# This will raise an exception on Windows. That's ok.
1737	n/a	pid, status = os.waitpid(any_process, os.WNOHANG)
1738	n/a	if pid == 0:
1739	n/a	break
1740	n/a	except:
1741	n/a	break
1742	n/a
1743	n/a	@contextlib.contextmanager
1744	n/a	def swap_attr(obj, attr, new_val):
1745	n/a	"""Temporary swap out an attribute with a new object.
1746	n/a
1747	n/a	Usage:
1748	n/a	with swap_attr(obj, "attr", 5):
1749	n/a	...
1750	n/a
1751	n/a	This will set obj.attr to 5 for the duration of the with: block,
1752	n/a	restoring the old value at the end of the block. If `attr` doesn't
1753	n/a	exist on `obj`, it will be created and then deleted at the end of the
1754	n/a	block.
1755	n/a	"""
1756	n/a	if hasattr(obj, attr):
1757	n/a	real_val = getattr(obj, attr)
1758	n/a	setattr(obj, attr, new_val)
1759	n/a	try:
1760	n/a	yield
1761	n/a	finally:
1762	n/a	setattr(obj, attr, real_val)
1763	n/a	else:
1764	n/a	setattr(obj, attr, new_val)
1765	n/a	try:
1766	n/a	yield
1767	n/a	finally:
1768	n/a	delattr(obj, attr)
1769	n/a
1770	n/a	@contextlib.contextmanager
1771	n/a	def swap_item(obj, item, new_val):
1772	n/a	"""Temporary swap out an item with a new object.
1773	n/a
1774	n/a	Usage:
1775	n/a	with swap_item(obj, "item", 5):
1776	n/a	...
1777	n/a
1778	n/a	This will set obj["item"] to 5 for the duration of the with: block,
1779	n/a	restoring the old value at the end of the block. If `item` doesn't
1780	n/a	exist on `obj`, it will be created and then deleted at the end of the
1781	n/a	block.
1782	n/a	"""
1783	n/a	if item in obj:
1784	n/a	real_val = obj[item]
1785	n/a	obj[item] = new_val
1786	n/a	try:
1787	n/a	yield
1788	n/a	finally:
1789	n/a	obj[item] = real_val
1790	n/a	else:
1791	n/a	obj[item] = new_val
1792	n/a	try:
1793	n/a	yield
1794	n/a	finally:
1795	n/a	del obj[item]
1796	n/a
1797	n/a	def strip_python_stderr(stderr):
1798	n/a	"""Strip the stderr of a Python process from potential debug output
1799	n/a	emitted by the interpreter.
1800	n/a
1801	n/a	This will typically be run on the result of the communicate() method
1802	n/a	of a subprocess.Popen object.
1803	n/a	"""
1804	n/a	stderr = re.sub(br"\[\d+ refs, \d+ blocks\]\r?\n?", b"", stderr).strip()
1805	n/a	return stderr
1806	n/a
1807	n/a	def args_from_interpreter_flags():
1808	n/a	"""Return a list of command-line arguments reproducing the current
1809	n/a	settings in sys.flags and sys.warnoptions."""
1810	n/a	return subprocess._args_from_interpreter_flags()
1811	n/a
1812	n/a	#============================================================
1813	n/a	# Support for assertions about logging.
1814	n/a	#============================================================
1815	n/a
1816	n/a	class TestHandler(logging.handlers.BufferingHandler):
1817	n/a	def __init__(self, matcher):
1818	n/a	# BufferingHandler takes a "capacity" argument
1819	n/a	# so as to know when to flush. As we're overriding
1820	n/a	# shouldFlush anyway, we can set a capacity of zero.
1821	n/a	# You can call flush() manually to clear out the
1822	n/a	# buffer.
1823	n/a	logging.handlers.BufferingHandler.__init__(self, 0)
1824	n/a	self.matcher = matcher
1825	n/a
1826	n/a	def shouldFlush(self):
1827	n/a	return False
1828	n/a
1829	n/a	def emit(self, record):
1830	n/a	self.format(record)
1831	n/a	self.buffer.append(record.__dict__)
1832	n/a
1833	n/a	def matches(self, **kwargs):
1834	n/a	"""
1835	n/a	Look for a saved dict whose keys/values match the supplied arguments.
1836	n/a	"""
1837	n/a	result = False
1838	n/a	for d in self.buffer:
1839	n/a	if self.matcher.matches(d, **kwargs):
1840	n/a	result = True
1841	n/a	break
1842	n/a	return result
1843	n/a
1844	n/a	class Matcher(object):
1845	n/a
1846	n/a	_partial_matches = ('msg', 'message')
1847	n/a
1848	n/a	def matches(self, d, **kwargs):
1849	n/a	"""
1850	n/a	Try to match a single dict with the supplied arguments.
1851	n/a
1852	n/a	Keys whose values are strings and which are in self._partial_matches
1853	n/a	will be checked for partial (i.e. substring) matches. You can extend
1854	n/a	this scheme to (for example) do regular expression matching, etc.
1855	n/a	"""
1856	n/a	result = True
1857	n/a	for k in kwargs:
1858	n/a	v = kwargs[k]
1859	n/a	dv = d.get(k)
1860	n/a	if not self.match_value(k, dv, v):
1861	n/a	result = False
1862	n/a	break
1863	n/a	return result
1864	n/a
1865	n/a	def match_value(self, k, dv, v):
1866	n/a	"""
1867	n/a	Try to match a single stored value (dv) with a supplied value (v).
1868	n/a	"""
1869	n/a	if type(v) != type(dv):
1870	n/a	result = False
1871	n/a	elif type(dv) is not str or k not in self._partial_matches:
1872	n/a	result = (v == dv)
1873	n/a	else:
1874	n/a	result = dv.find(v) >= 0
1875	n/a	return result
1876	n/a
1877	n/a
1878	n/a	_can_symlink = None
1879	n/a	def can_symlink():
1880	n/a	global _can_symlink
1881	n/a	if _can_symlink is not None:
1882	n/a	return _can_symlink
1883	n/a	symlink_path = TESTFN + "can_symlink"
1884	n/a	try:
1885	n/a	os.symlink(TESTFN, symlink_path)
1886	n/a	can = True
1887	n/a	except (OSError, NotImplementedError, AttributeError):
1888	n/a	can = False
1889	n/a	else:
1890	n/a	os.remove(symlink_path)
1891	n/a	_can_symlink = can
1892	n/a	return can
1893	n/a
1894	n/a	def skip_unless_symlink(test):
1895	n/a	"""Skip decorator for tests that require functional symlink"""
1896	n/a	ok = can_symlink()
1897	n/a	msg = "Requires functional symlink implementation"
1898	n/a	return test if ok else unittest.skip(msg)(test)
1899	n/a
1900	n/a	_can_xattr = None
1901	n/a	def can_xattr():
1902	n/a	global _can_xattr
1903	n/a	if _can_xattr is not None:
1904	n/a	return _can_xattr
1905	n/a	if not hasattr(os, "setxattr"):
1906	n/a	can = False
1907	n/a	else:
1908	n/a	tmp_fp, tmp_name = tempfile.mkstemp()
1909	n/a	try:
1910	n/a	with open(TESTFN, "wb") as fp:
1911	n/a	try:
1912	n/a	# TESTFN & tempfile may use different file systems with
1913	n/a	# different capabilities
1914	n/a	os.setxattr(tmp_fp, b"user.test", b"")
1915	n/a	os.setxattr(fp.fileno(), b"user.test", b"")
1916	n/a	# Kernels < 2.6.39 don't respect setxattr flags.
1917	n/a	kernel_version = platform.release()
1918	n/a	m = re.match("2.6.(\d{1,2})", kernel_version)
1919	n/a	can = m is None or int(m.group(1)) >= 39
1920	n/a	except OSError:
1921	n/a	can = False
1922	n/a	finally:
1923	n/a	unlink(TESTFN)
1924	n/a	unlink(tmp_name)
1925	n/a	_can_xattr = can
1926	n/a	return can
1927	n/a
1928	n/a	def skip_unless_xattr(test):
1929	n/a	"""Skip decorator for tests that require functional extended attributes"""
1930	n/a	ok = can_xattr()
1931	n/a	msg = "no non-broken extended attribute support"
1932	n/a	return test if ok else unittest.skip(msg)(test)
1933	n/a
1934	n/a
1935	n/a	if sys.platform.startswith('win'):
1936	n/a	@contextlib.contextmanager
1937	n/a	def suppress_crash_popup():
1938	n/a	"""Disable Windows Error Reporting dialogs using SetErrorMode."""
1939	n/a	# see http://msdn.microsoft.com/en-us/library/windows/desktop/ms680621%28v=vs.85%29.aspx
1940	n/a	# GetErrorMode is not available on Windows XP and Windows Server 2003,
1941	n/a	# but SetErrorMode returns the previous value, so we can use that
1942	n/a	import ctypes
1943	n/a	k32 = ctypes.windll.kernel32
1944	n/a	SEM_NOGPFAULTERRORBOX = 0x02
1945	n/a	old_error_mode = k32.SetErrorMode(SEM_NOGPFAULTERRORBOX)
1946	n/a	k32.SetErrorMode(old_error_mode \| SEM_NOGPFAULTERRORBOX)
1947	n/a	try:
1948	n/a	yield
1949	n/a	finally:
1950	n/a	k32.SetErrorMode(old_error_mode)
1951	n/a	else:
1952	n/a	# this is a no-op for other platforms
1953	n/a	@contextlib.contextmanager
1954	n/a	def suppress_crash_popup():
1955	n/a	yield
1956	n/a
1957	n/a
1958	n/a	def patch(test_instance, object_to_patch, attr_name, new_value):
1959	n/a	"""Override 'object_to_patch'.'attr_name' with 'new_value'.
1960	n/a
1961	n/a	Also, add a cleanup procedure to 'test_instance' to restore
1962	n/a	'object_to_patch' value for 'attr_name'.
1963	n/a	The 'attr_name' should be a valid attribute for 'object_to_patch'.
1964	n/a
1965	n/a	"""
1966	n/a	# check that 'attr_name' is a real attribute for 'object_to_patch'
1967	n/a	# will raise AttributeError if it does not exist
1968	n/a	getattr(object_to_patch, attr_name)
1969	n/a
1970	n/a	# keep a copy of the old value
1971	n/a	attr_is_local = False
1972	n/a	try:
1973	n/a	old_value = object_to_patch.__dict__[attr_name]
1974	n/a	except (AttributeError, KeyError):
1975	n/a	old_value = getattr(object_to_patch, attr_name, None)
1976	n/a	else:
1977	n/a	attr_is_local = True
1978	n/a
1979	n/a	# restore the value when the test is done
1980	n/a	def cleanup():
1981	n/a	if attr_is_local:
1982	n/a	setattr(object_to_patch, attr_name, old_value)
1983	n/a	else:
1984	n/a	delattr(object_to_patch, attr_name)
1985	n/a
1986	n/a	test_instance.addCleanup(cleanup)
1987	n/a
1988	n/a	# actually override the attribute
1989	n/a	setattr(object_to_patch, attr_name, new_value)