# globals

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

# exceptions

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

# io

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

"captured_stdin", "captured_stderr", "captured_output",

# unittest

"is_resource_enabled", "get_resource_value", "requires", "requires_resource",

"requires_freebsd_version",

"requires_gil_enabled", "requires_linux_version", "requires_mac_ver",

"requires_ios_ver",

"check_syntax_error",

"requires_gzip", "requires_bz2", "requires_lzma", "requires_zstd",

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

"requires_IEEE_754", "requires_zlib",

"has_fork_support", "requires_fork",

"has_subprocess_support", "requires_subprocess",

"has_socket_support", "requires_working_socket",

"has_remote_subprocess_debugging", "requires_remote_subprocess_debugging",

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

"check__all__", "skip_if_buggy_ucrt_strfptime",

"check_disallow_instantiation", "check_sanitizer", "skip_if_sanitizer",

"requires_limited_api", "requires_specialization", "thread_unsafe",

"skip_if_unlimited_stack_size", "skip_if_huge_c_stack",

"run_with_limited_c_stack",

# sys

"MS_WINDOWS", "is_jython", "is_android", "is_emscripten", "is_wasi",

"is_apple_mobile", "check_impl_detail", "unix_shell", "setswitchinterval",

"support_remote_exec_only",

# os

"get_pagesize",

# network

"open_urlresource",

# processes

"reap_children",

# miscellaneous

"run_with_locale", "swap_item", "findfile", "infinite_recursion",

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

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

"ALWAYS_EQ", "NEVER_EQ", "LARGEST", "SMALLEST",

"LOOPBACK_TIMEOUT", "INTERNET_TIMEOUT", "SHORT_TIMEOUT", "LONG_TIMEOUT",

"Py_DEBUG", "exceeds_recursion_limit", "skip_on_s390x",

"requires_jit_enabled",

"requires_jit_disabled",

"force_not_colorized",

"force_not_colorized_test_class",

"make_clean_env",

"BrokenIter",

"in_systemd_nspawn_sync_suppressed",

"run_no_yield_async_fn", "run_yielding_async_fn", "async_yield",

"reset_code", "on_github_actions",

"requires_root_user", "requires_non_root_user",

"skip_if_double_rounding", "built_with_c_assertions",

"""Test failed."""

def __init__(self, msg, *args, stats=None):

self.msg = msg

self.stats = stats

super().__init__(msg, *args)

def __str__(self):

"""Test failed."""

def __init__(self, msg, errors, failures, stats):

self.errors = errors

self.failures = failures

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

"""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 = STDLIB_DIR

package_tests = loader.discover(start_dir=pkg_dir,

top_level_dir=top_dir,

pattern=pattern)

standard_tests.addTests(package_tests)

"""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:

return func(*args)

except FileNotFoundError as err:

# chmod() won't fix a missing file.

if verbose >= 2:

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

raise

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)

if hasattr(_is_gui_available, 'result'):

return _is_gui_available.result

reason = None

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

reason = "gui is not available on Windows IoT Core"

elif 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.util

import ctypes.wintypes

UOI_FLAGS = 1

WSF_VISIBLE = 0x0001

@ctypes.util.struct

class USEROBJECTFLAGS:

fInherit: ctypes.wintypes.BOOL

fReserved: ctypes.wintypes.BOOL

dwFlags: ctypes.wintypes.DWORD

user32 = ctypes.windll.user32

@ctypes.util.wrap_dll_function(user32)

def GetProcessWindowStation() -> ctypes.wintypes.HANDLE:

...

h = GetProcessWindowStation()

if not h:

raise ctypes.WinError()

@ctypes.util.wrap_dll_function(user32)

def GetUserObjectInformationW(

hObj: ctypes.wintypes.HANDLE,

nIndex: ctypes.c_int,

pvInfo: ctypes.c_void_p,

nLength: ctypes.wintypes.DWORD,

lpnLengthNeeded: ctypes.POINTER(ctypes.wintypes.DWORD),

) -> ctypes.wintypes.BOOL:

...

uof = USEROBJECTFLAGS()

needed = ctypes.wintypes.DWORD()

res = 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.

import subprocess

try:

rc = subprocess.run(["launchctl", "managername"],

capture_output=True, check=True)

managername = rc.stdout.decode("utf-8").strip()

except subprocess.CalledProcessError:

reason = "unable to detect macOS launchd job manager"

else:

if managername != "Aqua":

reason = f"{managername=} -- can only run in a macOS GUI session"

# 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.

"""Test whether a resource is enabled.

if use_resources is None:

return None

"""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 in {"network", "urlfetch"} and not has_socket_support:

raise ResourceDenied("No socket support")

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

"""

version_txt = get_version_text()

if sys.platform == sys_platform:

try:

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

except ValueError:

skip = False

else:

skip = version < min_version

else:

skip = False

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

return unittest.skipIf(

skip,

f"{name} version {min_version_txt} or higher required, not "

f"{version_txt}"

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

return _requires_platform_version(

'FreeBSD', 'freebsd',

get_version_text=_get_platform_release,

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

return _requires_platform_version(

"Linux", "linux",

get_version_text=_get_platform_release,

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

return _requires_platform_version(

"macOS", "darwin",

get_version_text=lambda: platform.mac_ver()[0],

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

return _requires_platform_version(

"iOS", "ios",

get_version_text=lambda: platform.ios_ver().release,

"""Mark a test as not thread safe. When the test runner is run with

def decorator(test_item):

test_item.__unittest_thread_unsafe__ = True

# the reason is not currently used

test_item.__unittest_thread_unsafe__why__ = reason

return test_item

if isinstance(reason, types.FunctionType):

test_item = reason

reason = ''

return decorator(test_item)

"""Decorator raising SkipTest if running on a buildbot."""

import getpass

if not reason:

reason = 'not suitable for buildbots'

try:

isbuildbot = getpass.getuser().lower() == 'buildbot'

except (KeyError, OSError) as err:

logging.getLogger(__name__).warning('getpass.getuser() failed %s.', err, exc_info=err)

isbuildbot = False

thread=False, function=True):

"""Returns True if Python is compiled with sanitizer support"""

if not (address or hwaddress or memory or ub or thread):

raise ValueError('At least one of address, hwaddress, memory, ub or '

'thread must be True')

cflags = sysconfig.get_config_var('CFLAGS') or ''

config_args = sysconfig.get_config_var('CONFIG_ARGS') or ''

memory_sanitizer = (

'-fsanitize=memory' in cflags or

'--with-memory-sanitizer' in config_args

)

hwaddress_sanitizer = (

'-fsanitize=hwaddress' in cflags or

'--with-hwaddress-sanitizer' in config_args

)

address_sanitizer = (

'-fsanitize=address' in cflags or

'--with-address-sanitizer' in config_args or

hwaddress_sanitizer

)

ub_sanitizer = (

'-fsanitize=undefined' in cflags or

'--with-undefined-behavior-sanitizer' in config_args

)

thread_sanitizer = (

'-fsanitize=thread' in cflags or

'--with-thread-sanitizer' in config_args

)

function_sanitizer = (

'-fsanitize=function' in cflags

)

return (

(memory and memory_sanitizer) or

(address and address_sanitizer) or

(hwaddress and hwaddress_sanitizer) or

(ub and ub_sanitizer) or

(thread and thread_sanitizer) or

(function and function_sanitizer)

"""Decorator raising SkipTest if running with a sanitizer active."""

if not reason:

reason = 'not working with sanitizers active'

skip = check_sanitizer(address=address, memory=memory, ub=ub, thread=thread)

for name in ('ASAN_OPTIONS', 'MSAN_OPTIONS', 'UBSAN_OPTIONS', 'TSAN_OPTIONS'):

if name in env:

env[name] += f':{option}'

else:

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

@functools.wraps(f)

def dec(*args, **kwargs):

try:

f(*args, **kwargs)

except OSError as e:

if "CERTIFICATE_VERIFY_FAILED" in str(e):

raise unittest.SkipTest("system does not contain "

"necessary certificates")

raise

try:

import zlib

except ImportError:

zlib = None

try:

import gzip

except ImportError:

gzip = None

try:

import bz2

except ImportError:

bz2 = None

try:

import lzma

except ImportError:

lzma = None

try:

from compression import zstd

except ImportError:

zstd = None

try:

import _testcapi

except ImportError:

raise unittest.SkipTest("_testinternalcapi required")

try:

skip = has_no_debug_ranges()

except unittest.SkipTest as e:

skip = True

reason = e.args[0] if e.args else reason

# WASM and Apple mobile platforms do not support subprocesses.

is_emscripten

or is_wasi

or is_apple_mobile

# Although Android supports fork, it's unsafe to call it from Python because

# all Android apps are multi-threaded.

or is_android

# WASM and Apple mobile platforms do not support subprocesses.

is_emscripten

or is_wasi

or is_apple_mobile

# Although Android supports subproceses, they're almost never useful in

# practice (see PEP 738). And most of the tests that use them are calling

# sys.executable, which won't work when Python is embedded in an Android app.

or is_android

"""Skip tests or modules that require working sockets

msg = "requires socket support"

if module:

if not has_socket_support:

raise unittest.SkipTest(msg)

else:

"""Check if we have permissions to debug subprocesses remotely.

# Check platform support

if sys.platform not in ("linux", "darwin", "win32"):

return False

try:

import _remote_debugging

except ImportError:

return False

# On Linux, check for process_vm_readv support

if sys.platform == "linux":

if not getattr(_remote_debugging, "PROCESS_VM_READV_SUPPORTED", False):

return False

# First check if we can read our own process

if not _remote_debugging.is_python_process(os.getpid()):

return False

# Check subprocess access - debugging child processes may require

# additional permissions depending on platform security settings

import socket

import subprocess

# Create a socket for child to signal readiness

server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

server.bind(("127.0.0.1", 0))

server.listen(1)

port = server.getsockname()[1]

# Child connects to signal it's ready, then waits for parent to close

child_code = f"""

proc = subprocess.Popen(

[sys.executable, "-c", child_code],

stdout=subprocess.DEVNULL,

stderr=subprocess.DEVNULL,

)

try:

server.settimeout(5.0)

conn, _ = server.accept()

# Child is ready, test if we can probe it

result = _remote_debugging.is_python_process(proc.pid)

# Check if subprocess is still alive after probing

if proc.poll() is not None:

return False

conn.close() # Signal child to exit

return result

except (socket.timeout, OSError):

return False

finally:

server.close()

proc.kill()

"""Skip tests that require remote subprocess debugging permissions.

if not has_subprocess_support:

return unittest.skip("requires subprocess support")

return unittest.skipUnless(

has_remote_subprocess_debugging(),

"requires remote subprocess debugging permissions"

"""Assure user that loud errors generated by macOS libc's malloc are

if sys.platform != 'darwin':

return

import shutil

msg = ' NOTICE '

detail = (f'{test_name} may generate "malloc can\'t allocate region"\n'

'warnings on macOS systems. This behavior is known. Do not\n'

'report a bug unless tests are also failing.\n'

'See https://github.com/python/cpython/issues/85100')

padding, _ = shutil.get_terminal_size()

print(msg.center(padding, '-'))

print(detail)

"""Try to find a file on sys.path or in the test directory. If it is not

if os.path.isabs(filename):

return filename

if subdir is not None:

filename = os.path.join(subdir, filename)

path = [TEST_HOME_DIR] + sys.path

for dn in path:

fn = os.path.join(dn, filename)

if os.path.exists(fn): return fn

"Like repr(dict), but in sorted order."

items = sorted(dict.items())

reprpairs = ["%r: %r" % pair for pair in items]

withcommas = ", ".join(reprpairs)

"""Run a piece of code after dedenting it, and return its global namespace."""

ns = {}

if extra_names:

ns.update(extra_names)

exec(textwrap.dedent(code), ns)

with testcase.assertRaisesRegex(SyntaxError, errtext) as cm:

compile(statement, '<test string>', 'exec')

err = cm.exception

testcase.assertIsNotNone(err.lineno)

if lineno is not None:

testcase.assertEqual(err.lineno, lineno)

testcase.assertIsNotNone(err.offset)

if offset is not None:

import urllib.request, urllib.parse

from .os_helper import unlink

try:

import gzip

except ImportError:

gzip = None

check = kw.pop('check', None)

filename = urllib.parse.urlparse(url).path.split('/')[-1] # '/': it's URL!

fn = os.path.join(TEST_DATA_DIR, filename)

def check_valid_file(fn):

f = open(fn, *args, **kw)

if check is None:

return f

elif check(f):

f.seek(0)

return f

f.close()

if os.path.exists(fn):

f = check_valid_file(fn)

if f is not None:

return f

unlink(fn)

# Verify the requirement before downloading the file

requires('urlfetch')

if verbose:

print('\tfetching %s ...' % url, file=get_original_stdout())

opener = urllib.request.build_opener()

if gzip:

opener.addheaders.append(('Accept-Encoding', 'gzip'))

f = opener.open(url, timeout=INTERNET_TIMEOUT)

if gzip and f.headers.get('Content-Encoding') == 'gzip':

f = gzip.GzipFile(fileobj=f)

try:

with open(fn, "wb") as out:

s = f.read()

while s:

out.write(s)

s = f.read()

finally:

f.close()

f = check_valid_file(fn)

if f is not None:

return f

"""Return a context manager used by captured_stdout/stdin/stderr

import io

orig_stdout = getattr(sys, stream_name)

setattr(sys, stream_name, io.StringIO())

try:

yield getattr(sys, stream_name)

finally:

"""Capture the output of sys.stdout:

"""Capture the output of sys.stderr:

"""Capture the input to sys.stdin:

"""Force as many objects as possible to be collected.

import gc

gc.collect()

gc.collect()

import gc

have_gc = gc.isenabled()

gc.disable()

try:

yield

finally: