# globals

"PIPE_MAX_SIZE", "verbose", "max_memuse", "use_resources", "failfast",

# exceptions

"Error", "TestFailed", "TestDidNotRun", "ResourceDenied",

# imports

"import_module", "import_fresh_module", "CleanImport",

# modules

"unload", "forget",

# io

"record_original_stdout", "get_original_stdout", "captured_stdout",

"captured_stdin", "captured_stderr",

# filesystem

"TESTFN", "SAVEDCWD", "unlink", "rmtree", "temp_cwd", "findfile",

"create_empty_file", "can_symlink", "fs_is_case_insensitive",

# unittest

"is_resource_enabled", "requires", "requires_freebsd_version",

"requires_linux_version", "requires_mac_ver", "check_syntax_error",

"TransientResource", "time_out", "socket_peer_reset", "ioerror_peer_reset",

"transient_internet", "BasicTestRunner", "run_unittest", "run_doctest",

"skip_unless_symlink", "requires_gzip", "requires_bz2", "requires_lzma",

"bigmemtest", "bigaddrspacetest", "cpython_only", "get_attribute",

"requires_IEEE_754", "skip_unless_xattr", "requires_zlib",

"anticipate_failure", "load_package_tests", "detect_api_mismatch",

"check__all__", "requires_android_level", "requires_multiprocessing_queue",

# sys

"is_jython", "is_android", "check_impl_detail", "unix_shell",

"setswitchinterval",

# network

"HOST", "IPV6_ENABLED", "find_unused_port", "bind_port", "open_urlresource",

"bind_unix_socket",

# processes

'temp_umask', "reap_children",

# logging

"TestHandler",

# threads

"threading_setup", "threading_cleanup", "reap_threads", "start_threads",

# miscellaneous

"check_warnings", "check_no_resource_warning", "EnvironmentVarGuard",

"run_with_locale", "swap_item",

"swap_attr", "Matcher", "set_memlimit", "SuppressCrashReport", "sortdict",

"run_with_tz", "PGO", "missing_compiler_executable", "fd_count",

"""Test skipped because it requested a disallowed resource.

"""Context manager to suppress package and module deprecation

if ignore:

with warnings.catch_warnings():

warnings.filterwarnings("ignore", ".+ (module|package)",

DeprecationWarning)

yield

else:

"""Import and return the module to be tested, raising SkipTest if

with _ignore_deprecated_imports(deprecated):

try:

return importlib.import_module(name)

except ImportError as msg:

if sys.platform.startswith(tuple(required_on)):

raise

"""Helper function to save and remove a module from sys.modules

# try to import the module and raise an error if it can't be imported

if name not in sys.modules:

__import__(name)

del sys.modules[name]

for modname in list(sys.modules):

if modname == name or modname.startswith(name + '.'):

orig_modules[modname] = sys.modules[modname]

"""Helper function to save and block a module in sys.modules

saved = True

try:

orig_modules[name] = sys.modules[name]

except KeyError:

saved = False

sys.modules[name] = None

"""Decorator to mark a test that is known to be broken in some cases

if condition:

return unittest.expectedFailure

"""Generic load_tests implementation for simple test packages.

if pattern is None:

pattern = "test*"

top_dir = os.path.dirname( # Lib

os.path.dirname( # test

os.path.dirname(__file__))) # support

package_tests = loader.discover(start_dir=pkg_dir,

top_level_dir=top_dir,

pattern=pattern)

standard_tests.addTests(package_tests)

"""Import and return a module, deliberately bypassing sys.modules.

# NOTE: test_heapq, test_json and test_warnings include extra sanity checks

# to make sure that this utility function is working as expected

with _ignore_deprecated_imports(deprecated):

# Keep track of modules saved for later restoration as well

# as those which just need a blocking entry removed

orig_modules = {}

names_to_remove = []

_save_and_remove_module(name, orig_modules)

try:

for fresh_name in fresh:

_save_and_remove_module(fresh_name, orig_modules)

for blocked_name in blocked:

if not _save_and_block_module(blocked_name, orig_modules):

names_to_remove.append(blocked_name)

fresh_module = importlib.import_module(name)

except ImportError:

fresh_module = None

finally:

for orig_name, module in orig_modules.items():

sys.modules[orig_name] = module

for name_to_remove in names_to_remove:

del sys.modules[name_to_remove]

"""Get an attribute, raising SkipTest if AttributeError is raised."""

try:

attribute = getattr(obj, name)

except AttributeError:

raise unittest.SkipTest("object %r has no attribute %r" % (obj, name))

else:

try:

del sys.modules[name]

except KeyError:

try:

return func(*args)

except OSError as err:

if verbose >= 2:

print('%s: %s' % (err.__class__.__name__, err))

print('re-run %s%r' % (func.__name__, args))

os.chmod(path, stat.S_IRWXU)

# Perform the operation

func(pathname)

# Now setup the wait loop

if waitall:

dirname = pathname

else:

dirname, name = os.path.split(pathname)

dirname = dirname or '.'

# Check for `pathname` to be removed from the filesystem.

# The exponential backoff of the timeout amounts to a total

# of ~1 second after which the deletion is probably an error

# anyway.

# Testing on an i7@4.3GHz shows that usually only 1 iteration is

# required when contention occurs.

timeout = 0.001

while timeout < 1.0:

# Note we are only testing for the existence of the file(s) in

# the contents of the directory regardless of any security or

# access rights. If we have made it this far, we have sufficient

# permissions to do that much using Python's equivalent of the

# Windows API FindFirstFile.

# Other Windows APIs can fail or give incorrect results when

# dealing with files that are pending deletion.

L = os.listdir(dirname)

if not (L if waitall else name in L):

return

# Increase the timeout and try again

time.sleep(timeout)

timeout *= 2

warnings.warn('tests may fail, delete still pending for ' + pathname,

def _rmtree_inner(path):

for name in _force_run(path, os.listdir, path):

fullname = os.path.join(path, name)

try:

mode = os.lstat(fullname).st_mode

except OSError as exc:

print("support.rmtree(): os.lstat(%r) failed with %s" % (fullname, exc),

file=sys.__stderr__)

mode = 0

if stat.S_ISDIR(mode):

_waitfor(_rmtree_inner, fullname, waitall=True)

_force_run(fullname, os.rmdir, fullname)

else:

_force_run(fullname, os.unlink, fullname)

_waitfor(_rmtree_inner, path, waitall=True)

try:

import ctypes

except ImportError:

# No ctypes means we can't expands paths.

pass

else:

buffer = ctypes.create_unicode_buffer(len(path) * 2)

length = ctypes.windll.kernel32.GetLongPathNameW(path, buffer,

len(buffer))

if length:

return buffer[:length]

try:

shutil.rmtree(path)

return

except OSError:

pass

def _rmtree_inner(path):

for name in _force_run(path, os.listdir, path):

fullname = os.path.join(path, name)

try:

mode = os.lstat(fullname).st_mode

except OSError:

mode = 0

if stat.S_ISDIR(mode):

_rmtree_inner(fullname)

_force_run(path, os.rmdir, fullname)

else:

_force_run(path, os.unlink, fullname)

_rmtree_inner(path)

try:

_unlink(filename)

except (FileNotFoundError, NotADirectoryError):

try:

_rmdir(dirname)

except FileNotFoundError:

try:

_rmtree(path)

except FileNotFoundError:

"""Move a PEP 3147/488 pyc file to its legacy pyc location.

pyc_file = importlib.util.cache_from_source(source)

up_one = os.path.dirname(os.path.abspath(source))

legacy_pyc = os.path.join(up_one, source + 'c')

os.rename(pyc_file, legacy_pyc)

"""'Forget' a module was ever imported.

unload(modname)

for dirname in sys.path:

source = os.path.join(dirname, modname + '.py')

# It doesn't matter if they exist or not, unlink all possible

# combinations of PEP 3147/488 and legacy pyc files.

unlink(source + 'c')

for opt in ('', 1, 2):

if hasattr(_is_gui_available, 'result'):

return _is_gui_available.result

reason = None

if sys.platform.startswith('win'):

# if Python is running as a service (such as the buildbot service),

# gui interaction may be disallowed

import ctypes

import ctypes.wintypes

UOI_FLAGS = 1

WSF_VISIBLE = 0x0001

class USEROBJECTFLAGS(ctypes.Structure):

_fields_ = [("fInherit", ctypes.wintypes.BOOL),

("fReserved", ctypes.wintypes.BOOL),

("dwFlags", ctypes.wintypes.DWORD)]

dll = ctypes.windll.user32

h = dll.GetProcessWindowStation()

if not h:

raise ctypes.WinError()

uof = USEROBJECTFLAGS()

needed = ctypes.wintypes.DWORD()

res = dll.GetUserObjectInformationW(h,

UOI_FLAGS,

ctypes.byref(uof),

ctypes.sizeof(uof),

ctypes.byref(needed))

if not res:

raise ctypes.WinError()

if not bool(uof.dwFlags & WSF_VISIBLE):

reason = "gui not available (WSF_VISIBLE flag not set)"

elif sys.platform == 'darwin':

# The Aqua Tk implementations on OS X can abort the process if

# being called in an environment where a window server connection

# cannot be made, for instance when invoked by a buildbot or ssh

# process not running under the same user id as the current console

# user. To avoid that, raise an exception if the window manager

# connection is not available.

from ctypes import cdll, c_int, pointer, Structure

from ctypes.util import find_library

app_services = cdll.LoadLibrary(find_library("ApplicationServices"))

if app_services.CGMainDisplayID() == 0:

reason = "gui tests cannot run without OS X window manager"

else:

class ProcessSerialNumber(Structure):

_fields_ = [("highLongOfPSN", c_int),

("lowLongOfPSN", c_int)]

psn = ProcessSerialNumber()

psn_p = pointer(psn)

if ( (app_services.GetCurrentProcess(psn_p) < 0) or

(app_services.SetFrontProcess(psn_p) < 0) ):

reason = "cannot run without OS X gui process"

# check on every platform whether tkinter can actually do anything

if not reason:

try:

from tkinter import Tk

root = Tk()

root.withdraw()

root.update()

root.destroy()

except Exception as e:

err_string = str(e)

if len(err_string) > 50:

err_string = err_string[:50] + ' [...]'

reason = 'Tk unavailable due to {}: {}'.format(type(e).__name__,

err_string)

_is_gui_available.reason = reason

_is_gui_available.result = not reason

"""Test whether a resource is enabled.

"""Raise ResourceDenied if the specified resource is not available."""

if not is_resource_enabled(resource):

if msg is None:

msg = "Use of the %r resource not enabled" % resource

raise ResourceDenied(msg)

if resource == 'gui' and not _is_gui_available():

"""Decorator raising SkipTest if the OS is `sysname` and the version is less

def decorator(func):

@functools.wraps(func)

def wrapper(*args, **kw):

if platform.system() == sysname:

version_txt = platform.release().split('-', 1)[0]

try:

version = tuple(map(int, version_txt.split('.')))

except ValueError:

pass

else:

if version < min_version:

min_version_txt = '.'.join(map(str, min_version))

raise unittest.SkipTest(

"%s version %s or higher required, not %s"

% (sysname, min_version_txt, version_txt))

return func(*args, **kw)

wrapper.min_version = min_version

return wrapper

"""Decorator raising SkipTest if the OS is FreeBSD and the FreeBSD version is

"""Decorator raising SkipTest if the OS is Linux and the Linux version is

"""Decorator raising SkipTest if the OS is Mac OS X and the OS X

def decorator(func):

@functools.wraps(func)

def wrapper(*args, **kw):

if sys.platform == 'darwin':

version_txt = platform.mac_ver()[0]

try:

version = tuple(map(int, version_txt.split('.')))

except ValueError:

pass

else:

if version < min_version:

min_version_txt = '.'.join(map(str, min_version))

raise unittest.SkipTest(

"Mac OS X %s or higher required, not %s"

% (min_version_txt, version_txt))

return func(*args, **kw)

wrapper.min_version = min_version

return wrapper

"""Returns an unused port that should be suitable for binding. This is

tempsock = socket.socket(family, socktype)

port = bind_port(tempsock)

tempsock.close()

del tempsock

"""Bind the socket to a free port and return the port number. Relies on

if sock.family == socket.AF_INET and sock.type == socket.SOCK_STREAM:

if hasattr(socket, 'SO_REUSEADDR'):

if sock.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR) == 1:

raise TestFailed("tests should never set the SO_REUSEADDR " \

"socket option on TCP/IP sockets!")

if hasattr(socket, 'SO_REUSEPORT'):

try:

if sock.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT) == 1:

raise TestFailed("tests should never set the SO_REUSEPORT " \

"socket option on TCP/IP sockets!")

except OSError:

# Python's socket module was compiled using modern headers

# thus defining SO_REUSEPORT but this process is running

# under an older kernel that does not support SO_REUSEPORT.

pass

if hasattr(socket, 'SO_EXCLUSIVEADDRUSE'):

sock.setsockopt(socket.SOL_SOCKET, socket.SO_EXCLUSIVEADDRUSE, 1)

sock.bind((host, 0))

port = sock.getsockname()[1]

"""Bind a unix socket, raising SkipTest if PermissionError is raised."""

assert sock.family == socket.AF_UNIX

try:

sock.bind(addr)

except PermissionError:

sock.close()

"""Check whether IPv6 is enabled on this host."""

if socket.has_ipv6:

sock = None

try:

sock = socket.socket(socket.AF_INET6, socket.SOCK_STREAM)

sock.bind((HOSTv6, 0))

return True

except OSError:

pass

finally:

if sock:

sock.close()

"""Skip the test on TLS certificate validation failures."""

@functools.wraps(f)

def dec(*args, **kwargs):

try:

f(*args, **kwargs)

except IOError as e:

if "CERTIFICATE_VERIFY_FAILED" in str(e):

raise unittest.SkipTest("system does not contain "

"necessary certificates")

raise

"""Return a context manager that creates a temporary directory.

dir_created = False

if path is None:

path = tempfile.mkdtemp()

dir_created = True

path = os.path.realpath(path)

else:

try:

os.mkdir(path)

dir_created = True

except OSError:

if not quiet:

raise

warnings.warn('tests may fail, unable to create temp dir: ' + path,

RuntimeWarning, stacklevel=3)

if dir_created:

pid = os.getpid()

try:

yield path

finally:

# In case the process forks, let only the parent remove the

# directory. The child has a diffent process id. (bpo-30028)

if dir_created and pid == os.getpid():

"""Return a context manager that changes the current working directory.