diff --git a/Lib/test/support/__init__.py b/Lib/test/support/__init__.py index 444ca2219..e221ee046 100644 --- a/Lib/test/support/__init__.py +++ b/Lib/test/support/__init__.py @@ -3,11 +3,11 @@ if __name__ != 'test.support': raise ImportError('support must be imported from the test package') -import contextlib -import dataclasses -import functools -import logging import _opcode +import contextlib +import functools +import inspect +import logging import os import re import stat @@ -19,6 +19,7 @@ import types import unittest import warnings +import annotationlib __all__ = [ # globals @@ -32,7 +33,7 @@ __all__ = [ "is_resource_enabled", "requires", "requires_freebsd_version", "requires_gil_enabled", "requires_linux_version", "requires_mac_ver", "check_syntax_error", - "requires_gzip", "requires_bz2", "requires_lzma", + "requires_gzip", "requires_bz2", "requires_lzma", "requires_zstd", "bigmemtest", "bigaddrspacetest", "cpython_only", "get_attribute", "requires_IEEE_754", "requires_zlib", "has_fork_support", "requires_fork", @@ -41,10 +42,11 @@ __all__ = [ "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", + "requires_limited_api", "requires_specialization", "thread_unsafe", # 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 @@ -57,13 +59,16 @@ __all__ = [ "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", "get_c_recursion_limit", - "skip_on_s390x", - "without_optimizer", + "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" ] @@ -389,6 +394,21 @@ def requires_mac_ver(*min_version): return decorator +def thread_unsafe(reason): + """Mark a test as not thread safe. When the test runner is run with + --parallel-threads=N, the test will be run in a single thread.""" + 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) + return decorator + + def skip_if_buildbot(reason=None): """Decorator raising SkipTest if running on a buildbot.""" import getpass @@ -401,7 +421,8 @@ def skip_if_buildbot(reason=None): isbuildbot = False return unittest.skipIf(isbuildbot, reason) -def check_sanitizer(*, address=False, memory=False, ub=False, thread=False): +def check_sanitizer(*, address=False, memory=False, ub=False, thread=False, + function=True): """Returns True if Python is compiled with sanitizer support""" if not (address or memory or ub or thread): raise ValueError('At least one of address, memory, ub or thread must be True') @@ -425,11 +446,15 @@ def check_sanitizer(*, address=False, memory=False, ub=False, thread=False): '-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 (ub and ub_sanitizer) or - (thread and thread_sanitizer) + (thread and thread_sanitizer) or + (function and function_sanitizer) ) @@ -514,13 +539,19 @@ def requires_lzma(reason='requires lzma'): lzma = None # XXX: RUSTPYTHON; xz is not supported yet return unittest.skipUnless(lzma, reason) +def requires_zstd(reason='requires zstd'): + try: + from compression import zstd + except ImportError: + zstd = None + return unittest.skipUnless(zstd, reason) + def has_no_debug_ranges(): try: - import _testinternalcapi + import _testcapi except ImportError: raise unittest.SkipTest("_testinternalcapi required") - config = _testinternalcapi.get_config() - return not bool(config['code_debug_ranges']) + return not _testcapi.config_get('code_debug_ranges') def requires_debug_ranges(reason='requires co_positions / debug_ranges'): try: @@ -531,6 +562,7 @@ def requires_debug_ranges(reason='requires co_positions / debug_ranges'): return unittest.skipIf(skip, reason) +# XXX: RUSTPYTHON; this is not belong to 3.14 def can_use_suppress_immortalization(suppress=True): """Check if suppress_immortalization(suppress) can be used. @@ -583,6 +615,11 @@ is_jython = sys.platform.startswith('java') is_android = sys.platform == "android" +def skip_android_selinux(name): + return unittest.skipIf( + sys.platform == "android", f"Android blocks {name} with SELinux" + ) + if sys.platform not in {"win32", "vxworks", "ios", "tvos", "watchos"}: unix_shell = '/system/bin/sh' if is_android else '/bin/sh' else: @@ -593,6 +630,15 @@ else: is_emscripten = sys.platform == "emscripten" is_wasi = sys.platform == "wasi" +# Use is_wasm32 as a generic check for WebAssembly platforms. +is_wasm32 = is_emscripten or is_wasi + +def skip_emscripten_stack_overflow(): + return unittest.skipIf(is_emscripten, "Exhausts stack on Emscripten") + +def skip_wasi_stack_overflow(): + return unittest.skipIf(is_wasi, "Exhausts stack on WASI") + is_apple_mobile = sys.platform in {"ios", "tvos", "watchos"} is_apple = is_apple_mobile or sys.platform == "darwin" @@ -715,9 +761,11 @@ def sortdict(dict): return "{%s}" % withcommas -def run_code(code: str) -> dict[str, object]: +def run_code(code: str, extra_names: dict[str, object] | None = None) -> dict[str, object]: """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) return ns @@ -735,7 +783,9 @@ def check_syntax_error(testcase, statement, errtext='', *, lineno=None, offset=N def open_urlresource(url, *args, **kw): - import urllib.request, urllib.parse + import urllib.parse + import urllib.request + from .os_helper import unlink try: import gzip @@ -953,8 +1003,16 @@ def calcvobjsize(fmt): return struct.calcsize(_vheader + fmt + _align) -_TPFLAGS_HAVE_GC = 1<<14 +_TPFLAGS_STATIC_BUILTIN = 1<<1 +_TPFLAGS_DISALLOW_INSTANTIATION = 1<<7 +_TPFLAGS_IMMUTABLETYPE = 1<<8 _TPFLAGS_HEAPTYPE = 1<<9 +_TPFLAGS_BASETYPE = 1<<10 +_TPFLAGS_READY = 1<<12 +_TPFLAGS_READYING = 1<<13 +_TPFLAGS_HAVE_GC = 1<<14 +_TPFLAGS_BASE_EXC_SUBCLASS = 1<<30 +_TPFLAGS_TYPE_SUBCLASS = 1<<31 def check_sizeof(test, o, size): try: @@ -1318,6 +1376,26 @@ def no_tracing(func): return coverage_wrapper +def no_rerun(reason): + """Skip rerunning for a particular test. + + WARNING: Use this decorator with care; skipping rerunning makes it + impossible to find reference leaks. Provide a clear reason for skipping the + test using the 'reason' parameter. + """ + def deco(func): + assert not isinstance(func, type), func + _has_run = False + def wrapper(self): + nonlocal _has_run + if _has_run: + self.skipTest(reason) + func(self) + _has_run = True + return wrapper + return deco + + def refcount_test(test): """Decorator for tests which involve reference counting. @@ -1331,8 +1409,9 @@ def refcount_test(test): def requires_limited_api(test): try: - import _testcapi - import _testlimitedcapi + import _testcapi # noqa: F401 + + import _testlimitedcapi # noqa: F401 except ImportError: return unittest.skip('needs _testcapi and _testlimitedcapi modules')(test) return test @@ -1347,6 +1426,18 @@ def requires_specialization(test): _opcode.ENABLE_SPECIALIZATION, "requires specialization")(test) +def requires_specialization_ft(test): + return unittest.skipUnless( + _opcode.ENABLE_SPECIALIZATION_FT, "requires specialization")(test) + + +def reset_code(f: types.FunctionType) -> types.FunctionType: + """Clear all specializations, local instrumentation, and JIT code for the given function.""" + f.__code__ = f.__code__.replace() + return f + +on_github_actions = "GITHUB_ACTIONS" in os.environ + #======================================================================= # Check for the presence of docstrings. @@ -1575,8 +1666,8 @@ class PythonSymlink: if sys.platform == "win32": def _platform_specific(self): - import glob import _winapi + import glob if os.path.lexists(self.real) and not os.path.exists(self.real): # App symlink appears to not exist, but we want the @@ -1972,10 +2063,11 @@ def missing_compiler_executable(cmd_names=[]): missing. """ - from setuptools._distutils import ccompiler, sysconfig - from setuptools import errors import shutil + from setuptools import errors + from setuptools._distutils import ccompiler, sysconfig + compiler = ccompiler.new_compiler() sysconfig.customize_compiler(compiler) if compiler.compiler_type == "msvc": @@ -2304,7 +2396,15 @@ def skip_if_broken_multiprocessing_synchronize(): # bpo-38377: On Linux, creating a semaphore fails with OSError # if the current user does not have the permission to create # a file in /dev/shm/ directory. - synchronize.Lock(ctx=None) + import multiprocessing + synchronize.Lock(ctx=multiprocessing.get_context('fork')) + # The explicit fork mp context is required in order for + # TestResourceTracker.test_resource_tracker_reused to work. + # synchronize creates a new multiprocessing.resource_tracker + # process at module import time via the above call in that + # scenario. Awkward. This enables gh-84559. No code involved + # should have threads at that point so fork() should be safe. + except OSError as exc: raise unittest.SkipTest(f"broken multiprocessing SemLock: {exc!r}") @@ -2396,8 +2496,9 @@ def clear_ignored_deprecations(*tokens: object) -> None: raise ValueError("Provide token or tokens returned by ignore_deprecations_from") new_filters = [] + old_filters = warnings._get_filters() endswith = tuple(rf"(?#support{id(token)})" for token in tokens) - for action, message, category, module, lineno in warnings.filters: + for action, message, category, module, lineno in old_filters: if action == "ignore" and category is DeprecationWarning: if isinstance(message, re.Pattern): msg = message.pattern @@ -2406,8 +2507,8 @@ def clear_ignored_deprecations(*tokens: object) -> None: if msg.endswith(endswith): continue new_filters.append((action, message, category, module, lineno)) - if warnings.filters != new_filters: - warnings.filters[:] = new_filters + if old_filters != new_filters: + old_filters[:] = new_filters warnings._filters_mutated() @@ -2415,7 +2516,7 @@ def clear_ignored_deprecations(*tokens: object) -> None: def requires_venv_with_pip(): # ensurepip requires zlib to open ZIP archives (.whl binary wheel packages) try: - import zlib + import zlib # noqa: F401 except ImportError: return unittest.skipIf(True, "venv: ensurepip requires zlib") @@ -2455,6 +2556,7 @@ def _findwheel(pkgname): @contextlib.contextmanager def setup_venv_with_pip_setuptools(venv_dir): import subprocess + from .os_helper import temp_cwd def run_command(cmd): @@ -2610,30 +2712,30 @@ def sleeping_retry(timeout, err_msg=None, /, delay = min(delay * 2, max_delay) -class CPUStopwatch: +class Stopwatch: """Context manager to roughly time a CPU-bound operation. - Disables GC. Uses CPU time if it can (i.e. excludes sleeps & time of - other processes). + Disables GC. Uses perf_counter, which is a clock with the highest + available resolution. It is chosen even though it does include + time elapsed during sleep and is system-wide, because the + resolution of process_time is too coarse on Windows and + process_time does not exist everywhere (for example, WASM). - N.B.: - - This *includes* time spent in other threads. + Note: + - This *includes* time spent in other threads/processes. - Some systems only have a coarse resolution; check - stopwatch.clock_info.rseolution if. + stopwatch.clock_info.resolution when using the results. Usage: - with ProcessStopwatch() as stopwatch: + with Stopwatch() as stopwatch: ... elapsed = stopwatch.seconds resolution = stopwatch.clock_info.resolution """ def __enter__(self): - get_time = time.process_time - clock_info = time.get_clock_info('process_time') - if get_time() <= 0: # some platforms like WASM lack process_time() - get_time = time.monotonic - clock_info = time.get_clock_info('monotonic') + get_time = time.perf_counter + clock_info = time.get_clock_info('perf_counter') self.context = disable_gc() self.context.__enter__() self.get_time = get_time @@ -2661,6 +2763,7 @@ def adjust_int_max_str_digits(max_digits): sys.set_int_max_str_digits(current) +# XXX: RUSTPYTHON; removed in 3.14 def get_c_recursion_limit(): try: import _testcapi @@ -2671,7 +2774,7 @@ def get_c_recursion_limit(): def exceeds_recursion_limit(): """For recursion tests, easily exceeds default recursion limit.""" - return get_c_recursion_limit() * 3 + return 150_000 # Windows doesn't have os.uname() but it doesn't support s390x. @@ -2680,21 +2783,11 @@ skip_on_s390x = unittest.skipIf(is_s390x, 'skipped on s390x') Py_TRACE_REFS = hasattr(sys, 'getobjects') -# Decorator to disable optimizer while a function run -def without_optimizer(func): - try: - from _testinternalcapi import get_optimizer, set_optimizer - except ImportError: - return func - @functools.wraps(func) - def wrapper(*args, **kwargs): - save_opt = get_optimizer() - try: - set_optimizer(None) - return func(*args, **kwargs) - finally: - set_optimizer(save_opt) - return wrapper +# XXX: RUSTPYTHON; we don't have sys._jit yet +# _JIT_ENABLED = sys._jit.is_enabled() +_JIT_ENABLED = False +requires_jit_enabled = unittest.skipUnless(_JIT_ENABLED, "requires JIT enabled") +requires_jit_disabled = unittest.skipIf(_JIT_ENABLED, "requires JIT disabled") _BASE_COPY_SRC_DIR_IGNORED_NAMES = frozenset({ @@ -2724,19 +2817,121 @@ def copy_python_src_ignore(path, names): return ignored +# XXX Move this to the inspect module? +def walk_class_hierarchy(top, *, topdown=True): + # This is based on the logic in os.walk(). + assert isinstance(top, type), repr(top) + stack = [top] + while stack: + top = stack.pop() + if isinstance(top, tuple): + yield top + continue + + subs = type(top).__subclasses__(top) + if topdown: + # Yield before subclass traversal if going top down. + yield top, subs + # Traverse into subclasses. + for sub in reversed(subs): + stack.append(sub) + else: + # Yield after subclass traversal if going bottom up. + stack.append((top, subs)) + # Traverse into subclasses. + for sub in reversed(subs): + stack.append(sub) + + def iter_builtin_types(): - for obj in __builtins__.values(): - if not isinstance(obj, type): + # First try the explicit route. + try: + import _testinternalcapi + except ImportError: + _testinternalcapi = None + if _testinternalcapi is not None: + yield from _testinternalcapi.get_static_builtin_types() + return + + # Fall back to making a best-effort guess. + if hasattr(object, '__flags__'): + # Look for any type object with the Py_TPFLAGS_STATIC_BUILTIN flag set. + import datetime + seen = set() + for cls, subs in walk_class_hierarchy(object): + if cls in seen: + continue + seen.add(cls) + if not (cls.__flags__ & _TPFLAGS_STATIC_BUILTIN): + # Do not walk its subclasses. + subs[:] = [] + continue + yield cls + else: + # Fall back to a naive approach. + seen = set() + for obj in __builtins__.values(): + if not isinstance(obj, type): + continue + cls = obj + # XXX? + if cls.__module__ != 'builtins': + continue + if cls == ExceptionGroup: + # It's a heap type. + continue + if cls in seen: + continue + seen.add(cls) + yield cls + + +# XXX Move this to the inspect module? +def iter_name_in_mro(cls, name): + """Yield matching items found in base.__dict__ across the MRO. + + The descriptor protocol is not invoked. + + list(iter_name_in_mro(cls, name))[0] is roughly equivalent to + find_name_in_mro() in Objects/typeobject.c (AKA PyType_Lookup()). + + inspect.getattr_static() is similar. + """ + # This can fail if "cls" is weird. + for base in inspect._static_getmro(cls): + # This can fail if "base" is weird. + ns = inspect._get_dunder_dict_of_class(base) + try: + obj = ns[name] + except KeyError: continue - cls = obj - if cls.__module__ != 'builtins': - continue - yield cls + yield obj, base + + +# XXX Move this to the inspect module? +def find_name_in_mro(cls, name, default=inspect._sentinel): + for res in iter_name_in_mro(cls, name): + # Return the first one. + return res + if default is not inspect._sentinel: + return default, None + raise AttributeError(name) + + +# XXX The return value should always be exactly the same... +def identify_type_slot_wrappers(): + try: + import _testinternalcapi + except ImportError: + _testinternalcapi = None + if _testinternalcapi is not None: + names = {n: None for n in _testinternalcapi.identify_type_slot_wrappers()} + return list(names) + else: + raise NotImplementedError def iter_slot_wrappers(cls): - assert cls.__module__ == 'builtins', cls - def is_slot_wrapper(name, value): if not isinstance(value, types.WrapperDescriptorType): assert not repr(value).startswith(' dict[str, str]: return clean_env -def initialized_with_pyrepl(): - """Detect whether PyREPL was used during Python initialization.""" - # If the main module has a __file__ attribute it's a Python module, which means PyREPL. - return hasattr(sys.modules["__main__"], "__file__") +WINDOWS_STATUS = { + 0xC0000005: "STATUS_ACCESS_VIOLATION", + 0xC00000FD: "STATUS_STACK_OVERFLOW", + 0xC000013A: "STATUS_CONTROL_C_EXIT", +} +def get_signal_name(exitcode): + import signal + + if exitcode < 0: + signum = -exitcode + try: + return signal.Signals(signum).name + except ValueError: + pass + + # Shell exit code (ex: WASI build) + if 128 < exitcode < 256: + signum = exitcode - 128 + try: + return signal.Signals(signum).name + except ValueError: + pass + + try: + return WINDOWS_STATUS[exitcode] + except KeyError: + pass + + return None class BrokenIter: def __init__(self, init_raises=False, next_raises=False, iter_raises=False): @@ -2849,222 +3106,166 @@ class BrokenIter: return self +def in_systemd_nspawn_sync_suppressed() -> bool: + """ + Test whether the test suite is runing in systemd-nspawn + with ``--suppress-sync=true``. + + This can be used to skip tests that rely on ``fsync()`` calls + and similar not being intercepted. + """ + + if not hasattr(os, "O_SYNC"): + return False + + try: + with open("/run/systemd/container", "rb") as fp: + if fp.read().rstrip() != b"systemd-nspawn": + return False + except FileNotFoundError: + return False + + # If systemd-nspawn is used, O_SYNC flag will immediately + # trigger EINVAL. Otherwise, ENOENT will be given instead. + import errno + try: + fd = os.open(__file__, os.O_RDONLY | os.O_SYNC) + except OSError as err: + if err.errno == errno.EINVAL: + return True + else: + os.close(fd) + + return False + +def run_no_yield_async_fn(async_fn, /, *args, **kwargs): + coro = async_fn(*args, **kwargs) + try: + coro.send(None) + except StopIteration as e: + return e.value + else: + raise AssertionError("coroutine did not complete") + finally: + coro.close() + + +@types.coroutine +def async_yield(v): + return (yield v) + + +def run_yielding_async_fn(async_fn, /, *args, **kwargs): + coro = async_fn(*args, **kwargs) + try: + while True: + try: + coro.send(None) + except StopIteration as e: + return e.value + finally: + coro.close() + + +def is_libssl_fips_mode(): + try: + from _hashlib import get_fips_mode # ask _hashopenssl.c + except ImportError: + return False # more of a maybe, unless we add this to the _ssl module. + return get_fips_mode() != 0 + +def _supports_remote_attaching(): + PROCESS_VM_READV_SUPPORTED = False + + try: + from _remote_debugging import PROCESS_VM_READV_SUPPORTED + except ImportError: + pass + + return PROCESS_VM_READV_SUPPORTED + +def _support_remote_exec_only_impl(): + if not sys.is_remote_debug_enabled(): + return unittest.skip("Remote debugging is not enabled") + if sys.platform not in ("darwin", "linux", "win32"): + return unittest.skip("Test only runs on Linux, Windows and macOS") + if sys.platform == "linux" and not _supports_remote_attaching(): + return unittest.skip("Test only runs on Linux with process_vm_readv support") + return _id + +def support_remote_exec_only(test): + return _support_remote_exec_only_impl()(test) + +class EqualToForwardRef: + """Helper to ease use of annotationlib.ForwardRef in tests. + + This checks only attributes that can be set using the constructor. + + """ + + def __init__( + self, + arg, + *, + module=None, + owner=None, + is_class=False, + ): + self.__forward_arg__ = arg + self.__forward_is_class__ = is_class + self.__forward_module__ = module + self.__owner__ = owner + + def __eq__(self, other): + if not isinstance(other, (EqualToForwardRef, annotationlib.ForwardRef)): + return NotImplemented + return ( + self.__forward_arg__ == other.__forward_arg__ + and self.__forward_module__ == other.__forward_module__ + and self.__forward_is_class__ == other.__forward_is_class__ + and self.__owner__ == other.__owner__ + ) + + def __repr__(self): + extra = [] + if self.__forward_module__ is not None: + extra.append(f", module={self.__forward_module__!r}") + if self.__forward_is_class__: + extra.append(", is_class=True") + if self.__owner__ is not None: + extra.append(f", owner={self.__owner__!r}") + return f"EqualToForwardRef({self.__forward_arg__!r}{''.join(extra)})" + + +_linked_to_musl = None def linked_to_musl(): """ - Test if the Python executable is linked to the musl C library. + Report if the Python executable is linked to the musl C library. + + Return False if we don't think it is, or a version triple otherwise. """ + # This is can be a relatively expensive check, so we use a cache. + global _linked_to_musl + if _linked_to_musl is not None: + return _linked_to_musl + + # emscripten (at least as far as we're concerned) and wasi use musl, + # but platform doesn't know how to get the version, so set it to zero. + if is_wasm32: + _linked_to_musl = (0, 0, 0) + return _linked_to_musl + + # On all other non-linux platforms assume no musl. if sys.platform != 'linux': - return False + _linked_to_musl = False + return _linked_to_musl - import subprocess - exe = getattr(sys, '_base_executable', sys.executable) - cmd = ['ldd', exe] - try: - stdout = subprocess.check_output(cmd, - text=True, - stderr=subprocess.STDOUT) - except (OSError, subprocess.CalledProcessError): - return False - return ('musl' in stdout) - - -# TODO: RUSTPYTHON -# Every line of code below allowed us to update `Lib/test/support/__init__.py` without -# needing to update `libregtest` and its dependencies. -# Ideally we want to remove all code below and update `libregtest`. -# -# Code below was copied from: https://github.com/RustPython/RustPython/blob/9499d39f55b73535e2405bf208d5380241f79ada/Lib/test/support/__init__.py - -from .testresult import get_test_runner - -def _filter_suite(suite, pred): - """Recursively filter test cases in a suite based on a predicate.""" - newtests = [] - for test in suite._tests: - if isinstance(test, unittest.TestSuite): - _filter_suite(test, pred) - newtests.append(test) - else: - if pred(test): - newtests.append(test) - suite._tests = newtests - -# By default, don't filter tests -_match_test_func = None - -_accept_test_patterns = None -_ignore_test_patterns = None - -def match_test(test): - # Function used by support.run_unittest() and regrtest --list-cases - if _match_test_func is None: - return True - else: - return _match_test_func(test.id()) - -def _is_full_match_test(pattern): - # If a pattern contains at least one dot, it's considered - # as a full test identifier. - # Example: 'test.test_os.FileTests.test_access'. - # - # ignore patterns which contain fnmatch patterns: '*', '?', '[...]' - # or '[!...]'. For example, ignore 'test_access*'. - return ('.' in pattern) and (not re.search(r'[?*\[\]]', pattern)) - -def set_match_tests(accept_patterns=None, ignore_patterns=None): - global _match_test_func, _accept_test_patterns, _ignore_test_patterns - - if accept_patterns is None: - accept_patterns = () - if ignore_patterns is None: - ignore_patterns = () - - accept_func = ignore_func = None - - if accept_patterns != _accept_test_patterns: - accept_patterns, accept_func = _compile_match_function(accept_patterns) - if ignore_patterns != _ignore_test_patterns: - ignore_patterns, ignore_func = _compile_match_function(ignore_patterns) - - # Create a copy since patterns can be mutable and so modified later - _accept_test_patterns = tuple(accept_patterns) - _ignore_test_patterns = tuple(ignore_patterns) - - if accept_func is not None or ignore_func is not None: - def match_function(test_id): - accept = True - ignore = False - if accept_func: - accept = accept_func(test_id) - if ignore_func: - ignore = ignore_func(test_id) - return accept and not ignore - - _match_test_func = match_function - -def _compile_match_function(patterns): - if not patterns: - func = None - # set_match_tests(None) behaves as set_match_tests(()) - patterns = () - elif all(map(_is_full_match_test, patterns)): - # Simple case: all patterns are full test identifier. - # The test.bisect_cmd utility only uses such full test identifiers. - func = set(patterns).__contains__ - else: - import fnmatch - regex = '|'.join(map(fnmatch.translate, patterns)) - # The search *is* case sensitive on purpose: - # don't use flags=re.IGNORECASE - regex_match = re.compile(regex).match - - def match_test_regex(test_id): - if regex_match(test_id): - # The regex matches the whole identifier, for example - # 'test.test_os.FileTests.test_access'. - return True - else: - # Try to match parts of the test identifier. - # For example, split 'test.test_os.FileTests.test_access' - # into: 'test', 'test_os', 'FileTests' and 'test_access'. - return any(map(regex_match, test_id.split("."))) - - func = match_test_regex - - return patterns, func - -def run_unittest(*classes): - """Run tests from unittest.TestCase-derived classes.""" - valid_types = (unittest.TestSuite, unittest.TestCase) - loader = unittest.TestLoader() - suite = unittest.TestSuite() - for cls in classes: - if isinstance(cls, str): - if cls in sys.modules: - suite.addTest(loader.loadTestsFromModule(sys.modules[cls])) - else: - raise ValueError("str arguments must be keys in sys.modules") - elif isinstance(cls, valid_types): - suite.addTest(cls) - else: - suite.addTest(loader.loadTestsFromTestCase(cls)) - _filter_suite(suite, match_test) - return _run_suite(suite) - -def _run_suite(suite): - """Run tests from a unittest.TestSuite-derived class.""" - runner = get_test_runner(sys.stdout, - verbosity=verbose, - capture_output=(junit_xml_list is not None)) - - result = runner.run(suite) - - if junit_xml_list is not None: - junit_xml_list.append(result.get_xml_element()) - - if not result.testsRun and not result.skipped and not result.errors: - raise TestDidNotRun - if not result.wasSuccessful(): - stats = TestStats.from_unittest(result) - if len(result.errors) == 1 and not result.failures: - err = result.errors[0][1] - elif len(result.failures) == 1 and not result.errors: - err = result.failures[0][1] - else: - err = "multiple errors occurred" - if not verbose: err += "; run in verbose mode for details" - errors = [(str(tc), exc_str) for tc, exc_str in result.errors] - failures = [(str(tc), exc_str) for tc, exc_str in result.failures] - raise TestFailedWithDetails(err, errors, failures, stats=stats) - return result - -@dataclasses.dataclass(slots=True) -class TestStats: - tests_run: int = 0 - failures: int = 0 - skipped: int = 0 - - @staticmethod - def from_unittest(result): - return TestStats(result.testsRun, - len(result.failures), - len(result.skipped)) - - @staticmethod - def from_doctest(results): - return TestStats(results.attempted, - results.failed) - - def accumulate(self, stats): - self.tests_run += stats.tests_run - self.failures += stats.failures - self.skipped += stats.skipped - - -def run_doctest(module, verbosity=None, optionflags=0): - """Run doctest on the given module. Return (#failures, #tests). - - If optional argument verbosity is not specified (or is None), pass - support's belief about verbosity on to doctest. Else doctest's - usual behavior is used (it searches sys.argv for -v). - """ - - import doctest - - if verbosity is None: - verbosity = verbose - else: - verbosity = None - - results = doctest.testmod(module, - verbose=verbosity, - optionflags=optionflags) - if results.failed: - stats = TestStats.from_doctest(results) - raise TestFailed(f"{results.failed} of {results.attempted} " - f"doctests failed", - stats=stats) - if verbose: - print('doctest (%s) ... %d tests with zero failures' % - (module.__name__, results.attempted)) - return results + # On linux, we'll depend on the platform module to do the check, so new + # musl platforms should add support in that module if possible. + import platform + lib, version = platform.libc_ver() + if lib != 'musl': + _linked_to_musl = False + return _linked_to_musl + _linked_to_musl = tuple(map(int, version.split('.'))) + return _linked_to_musl diff --git a/Lib/test/support/_hypothesis_stubs/__init__.py b/Lib/test/support/_hypothesis_stubs/__init__.py index 6ba5bb814..9a57c3096 100644 --- a/Lib/test/support/_hypothesis_stubs/__init__.py +++ b/Lib/test/support/_hypothesis_stubs/__init__.py @@ -1,6 +1,6 @@ -from enum import Enum import functools import unittest +from enum import Enum __all__ = [ "given", diff --git a/Lib/test/support/ast_helper.py b/Lib/test/support/ast_helper.py index 8a0415b6a..98eaf0b27 100644 --- a/Lib/test/support/ast_helper.py +++ b/Lib/test/support/ast_helper.py @@ -1,5 +1,6 @@ import ast + class ASTTestMixin: """Test mixing to have basic assertions for AST nodes.""" @@ -16,6 +17,9 @@ class ASTTestMixin: self.fail(f"{type(a)!r} is not {type(b)!r}") if isinstance(a, ast.AST): for field in a._fields: + if isinstance(a, ast.Constant) and field == "kind": + # Skip the 'kind' field for ast.Constant + continue value1 = getattr(a, field, missing) value2 = getattr(b, field, missing) # Singletons are equal by definition, so further diff --git a/Lib/test/support/asynchat.py b/Lib/test/support/asynchat.py index 38c47a1fd..a8c6b28a9 100644 --- a/Lib/test/support/asynchat.py +++ b/Lib/test/support/asynchat.py @@ -1,5 +1,5 @@ # TODO: This module was deprecated and removed from CPython 3.12 -# Now it is a test-only helper. Any attempts to rewrite exising tests that +# Now it is a test-only helper. Any attempts to rewrite existing tests that # are using this module and remove it completely are appreciated! # See: https://github.com/python/cpython/issues/72719 diff --git a/Lib/test/support/asyncore.py b/Lib/test/support/asyncore.py index b397aca55..658c22fdc 100644 --- a/Lib/test/support/asyncore.py +++ b/Lib/test/support/asyncore.py @@ -1,5 +1,5 @@ # TODO: This module was deprecated and removed from CPython 3.12 -# Now it is a test-only helper. Any attempts to rewrite exising tests that +# Now it is a test-only helper. Any attempts to rewrite existing tests that # are using this module and remove it completely are appreciated! # See: https://github.com/python/cpython/issues/72719 @@ -51,17 +51,27 @@ many of the difficult problems for you, making the task of building sophisticated high-performance network servers and clients a snap. """ +import os import select import socket import sys import time import warnings - -import os -from errno import EALREADY, EINPROGRESS, EWOULDBLOCK, ECONNRESET, EINVAL, \ - ENOTCONN, ESHUTDOWN, EISCONN, EBADF, ECONNABORTED, EPIPE, EAGAIN, \ - errorcode - +from errno import ( + EAGAIN, + EALREADY, + EBADF, + ECONNABORTED, + ECONNRESET, + EINPROGRESS, + EINVAL, + EISCONN, + ENOTCONN, + EPIPE, + ESHUTDOWN, + EWOULDBLOCK, + errorcode, +) _DISCONNECTED = frozenset({ECONNRESET, ENOTCONN, ESHUTDOWN, ECONNABORTED, EPIPE, EBADF}) diff --git a/Lib/test/support/bytecode_helper.py b/Lib/test/support/bytecode_helper.py index 85bcd1f0f..4a3c8c2c4 100644 --- a/Lib/test/support/bytecode_helper.py +++ b/Lib/test/support/bytecode_helper.py @@ -1,9 +1,10 @@ """bytecode_helper - support tools for testing correct bytecode generation""" -import unittest import dis import io import opcode +import unittest + try: import _testinternalcapi except ImportError: @@ -71,7 +72,7 @@ class CompilationStepTestCase(unittest.TestCase): def assertInstructionsMatch(self, actual_seq, expected): # get an InstructionSequence and an expected list, where each - # entry is a label or an instruction tuple. Construct an expcted + # entry is a label or an instruction tuple. Construct an expected # instruction sequence and compare with the one given. self.assertIsInstance(expected, list) diff --git a/Lib/test/support/interpreters/channels.py b/Lib/test/support/channels.py similarity index 73% rename from Lib/test/support/interpreters/channels.py rename to Lib/test/support/channels.py index d2bd93d77..3f7b46030 100644 --- a/Lib/test/support/interpreters/channels.py +++ b/Lib/test/support/channels.py @@ -1,18 +1,22 @@ """Cross-interpreter Channels High Level Module.""" import time +from concurrent.interpreters import _crossinterp +from concurrent.interpreters._crossinterp import ( + UNBOUND_ERROR, + UNBOUND_REMOVE, +) + import _interpchannels as _channels -from . import _crossinterp # aliases: from _interpchannels import ( - ChannelError, ChannelNotFoundError, ChannelClosedError, - ChannelEmptyError, ChannelNotEmptyError, + ChannelClosedError, + ChannelEmptyError, + ChannelError, + ChannelNotEmptyError, + ChannelNotFoundError, ) -from ._crossinterp import ( - UNBOUND_ERROR, UNBOUND_REMOVE, -) - __all__ = [ 'UNBOUND', 'UNBOUND_ERROR', 'UNBOUND_REMOVE', @@ -55,15 +59,23 @@ def create(*, unbounditems=UNBOUND): """ unbound = _serialize_unbound(unbounditems) unboundop, = unbound - cid = _channels.create(unboundop) - recv, send = RecvChannel(cid), SendChannel(cid, _unbound=unbound) + cid = _channels.create(unboundop, -1) + recv, send = RecvChannel(cid), SendChannel(cid) + send._set_unbound(unboundop, unbounditems) return recv, send def list_all(): """Return a list of (recv, send) for all open channels.""" - return [(RecvChannel(cid), SendChannel(cid, _unbound=unbound)) - for cid, unbound in _channels.list_all()] + channels = [] + for cid, unboundop, _ in _channels.list_all(): + chan = _, send = RecvChannel(cid), SendChannel(cid) + if not hasattr(send, '_unboundop'): + send._set_unbound(unboundop) + else: + assert send._unbound[0] == unboundop + channels.append(chan) + return channels class _ChannelEnd: @@ -97,12 +109,8 @@ class _ChannelEnd: return other._id == self._id # for pickling: - def __getnewargs__(self): - return (int(self._id),) - - # for pickling: - def __getstate__(self): - return None + def __reduce__(self): + return (type(self), (int(self._id),)) @property def id(self): @@ -175,16 +183,33 @@ class SendChannel(_ChannelEnd): _end = 'send' - def __new__(cls, cid, *, _unbound=None): - if _unbound is None: - try: - op = _channels.get_channel_defaults(cid) - _unbound = (op,) - except ChannelNotFoundError: - _unbound = _serialize_unbound(UNBOUND) - self = super().__new__(cls, cid) - self._unbound = _unbound - return self +# def __new__(cls, cid, *, _unbound=None): +# if _unbound is None: +# try: +# op = _channels.get_channel_defaults(cid) +# _unbound = (op,) +# except ChannelNotFoundError: +# _unbound = _serialize_unbound(UNBOUND) +# self = super().__new__(cls, cid) +# self._unbound = _unbound +# return self + + def _set_unbound(self, op, items=None): + assert not hasattr(self, '_unbound') + if items is None: + items = _resolve_unbound(op) + unbound = (op, items) + self._unbound = unbound + return unbound + + @property + def unbounditems(self): + try: + _, items = self._unbound + except AttributeError: + op, _ = _channels.get_queue_defaults(self._id) + _, items = self._set_unbound(op) + return items @property def is_closed(self): @@ -192,61 +217,61 @@ class SendChannel(_ChannelEnd): return info.closed or info.closing def send(self, obj, timeout=None, *, - unbound=None, + unbounditems=None, ): """Send the object (i.e. its data) to the channel's receiving end. This blocks until the object is received. """ - if unbound is None: - unboundop, = self._unbound + if unbounditems is None: + unboundop = -1 else: - unboundop, = _serialize_unbound(unbound) + unboundop, = _serialize_unbound(unbounditems) _channels.send(self._id, obj, unboundop, timeout=timeout, blocking=True) def send_nowait(self, obj, *, - unbound=None, + unbounditems=None, ): """Send the object to the channel's receiving end. If the object is immediately received then return True (else False). Otherwise this is the same as send(). """ - if unbound is None: - unboundop, = self._unbound + if unbounditems is None: + unboundop = -1 else: - unboundop, = _serialize_unbound(unbound) + unboundop, = _serialize_unbound(unbounditems) # XXX Note that at the moment channel_send() only ever returns # None. This should be fixed when channel_send_wait() is added. # See bpo-32604 and gh-19829. return _channels.send(self._id, obj, unboundop, blocking=False) def send_buffer(self, obj, timeout=None, *, - unbound=None, + unbounditems=None, ): """Send the object's buffer to the channel's receiving end. This blocks until the object is received. """ - if unbound is None: - unboundop, = self._unbound + if unbounditems is None: + unboundop = -1 else: - unboundop, = _serialize_unbound(unbound) + unboundop, = _serialize_unbound(unbounditems) _channels.send_buffer(self._id, obj, unboundop, timeout=timeout, blocking=True) def send_buffer_nowait(self, obj, *, - unbound=None, + unbounditems=None, ): """Send the object's buffer to the channel's receiving end. If the object is immediately received then return True (else False). Otherwise this is the same as send(). """ - if unbound is None: - unboundop, = self._unbound + if unbounditems is None: + unboundop = -1 else: - unboundop, = _serialize_unbound(unbound) + unboundop, = _serialize_unbound(unbounditems) return _channels.send_buffer(self._id, obj, unboundop, blocking=False) def close(self): diff --git a/Lib/test/support/hashlib_helper.py b/Lib/test/support/hashlib_helper.py index a4e6c9220..75dc2ba75 100644 --- a/Lib/test/support/hashlib_helper.py +++ b/Lib/test/support/hashlib_helper.py @@ -1,51 +1,330 @@ import functools import hashlib +import importlib import unittest +from test.support.import_helper import import_module + try: import _hashlib except ImportError: _hashlib = None +try: + import _hmac +except ImportError: + _hmac = None + + +def requires_hashlib(): + return unittest.skipIf(_hashlib is None, "requires _hashlib") + + +def requires_builtin_hmac(): + return unittest.skipIf(_hmac is None, "requires _hmac") + + +def _missing_hash(digestname, implementation=None, *, exc=None): + parts = ["missing", implementation, f"hash algorithm: {digestname!r}"] + msg = " ".join(filter(None, parts)) + raise unittest.SkipTest(msg) from exc + + +def _openssl_availabillity(digestname, *, usedforsecurity): + try: + _hashlib.new(digestname, usedforsecurity=usedforsecurity) + except AttributeError: + assert _hashlib is None + _missing_hash(digestname, "OpenSSL") + except ValueError as exc: + _missing_hash(digestname, "OpenSSL", exc=exc) + + +def _decorate_func_or_class(func_or_class, decorator_func): + if not isinstance(func_or_class, type): + return decorator_func(func_or_class) + + decorated_class = func_or_class + setUpClass = decorated_class.__dict__.get('setUpClass') + if setUpClass is None: + def setUpClass(cls): + super(decorated_class, cls).setUpClass() + setUpClass.__qualname__ = decorated_class.__qualname__ + '.setUpClass' + setUpClass.__module__ = decorated_class.__module__ + else: + setUpClass = setUpClass.__func__ + setUpClass = classmethod(decorator_func(setUpClass)) + decorated_class.setUpClass = setUpClass + return decorated_class + def requires_hashdigest(digestname, openssl=None, usedforsecurity=True): - """Decorator raising SkipTest if a hashing algorithm is not available + """Decorator raising SkipTest if a hashing algorithm is not available. - The hashing algorithm could be missing or blocked by a strict crypto - policy. + The hashing algorithm may be missing, blocked by a strict crypto policy, + or Python may be configured with `--with-builtin-hashlib-hashes=no`. If 'openssl' is True, then the decorator checks that OpenSSL provides - the algorithm. Otherwise the check falls back to built-in - implementations. The usedforsecurity flag is passed to the constructor. + the algorithm. Otherwise the check falls back to (optional) built-in + HACL* implementations. + The usedforsecurity flag is passed to the constructor but has no effect + on HACL* implementations. + + Examples of exceptions being suppressed: ValueError: [digital envelope routines: EVP_DigestInit_ex] disabled for FIPS ValueError: unsupported hash type md4 """ - def decorator(func_or_class): - if isinstance(func_or_class, type): - setUpClass = func_or_class.__dict__.get('setUpClass') - if setUpClass is None: - def setUpClass(cls): - super(func_or_class, cls).setUpClass() - setUpClass.__qualname__ = func_or_class.__qualname__ + '.setUpClass' - setUpClass.__module__ = func_or_class.__module__ - else: - setUpClass = setUpClass.__func__ - setUpClass = classmethod(decorator(setUpClass)) - func_or_class.setUpClass = setUpClass - return func_or_class + if openssl and _hashlib is not None: + def test_availability(): + _hashlib.new(digestname, usedforsecurity=usedforsecurity) + else: + def test_availability(): + hashlib.new(digestname, usedforsecurity=usedforsecurity) - @functools.wraps(func_or_class) + def decorator_func(func): + @functools.wraps(func) def wrapper(*args, **kwargs): try: - if openssl and _hashlib is not None: - _hashlib.new(digestname, usedforsecurity=usedforsecurity) - else: - hashlib.new(digestname, usedforsecurity=usedforsecurity) - except ValueError: - raise unittest.SkipTest( - f"hash digest '{digestname}' is not available." - ) - return func_or_class(*args, **kwargs) + test_availability() + except ValueError as exc: + _missing_hash(digestname, exc=exc) + return func(*args, **kwargs) return wrapper + + def decorator(func_or_class): + return _decorate_func_or_class(func_or_class, decorator_func) return decorator + + +def requires_openssl_hashdigest(digestname, *, usedforsecurity=True): + """Decorator raising SkipTest if an OpenSSL hashing algorithm is missing. + + The hashing algorithm may be missing or blocked by a strict crypto policy. + """ + def decorator_func(func): + @requires_hashlib() # avoid checking at each call + @functools.wraps(func) + def wrapper(*args, **kwargs): + _openssl_availabillity(digestname, usedforsecurity=usedforsecurity) + return func(*args, **kwargs) + return wrapper + + def decorator(func_or_class): + return _decorate_func_or_class(func_or_class, decorator_func) + return decorator + + +def find_openssl_hashdigest_constructor(digestname, *, usedforsecurity=True): + """Find the OpenSSL hash function constructor by its name.""" + assert isinstance(digestname, str), digestname + _openssl_availabillity(digestname, usedforsecurity=usedforsecurity) + # This returns a function of the form _hashlib.openssl_ and + # not a lambda function as it is rejected by _hashlib.hmac_new(). + return getattr(_hashlib, f"openssl_{digestname}") + + +def requires_builtin_hashdigest( + module_name, digestname, *, usedforsecurity=True +): + """Decorator raising SkipTest if a HACL* hashing algorithm is missing. + + - The *module_name* is the C extension module name based on HACL*. + - The *digestname* is one of its member, e.g., 'md5'. + """ + def decorator_func(func): + @functools.wraps(func) + def wrapper(*args, **kwargs): + module = import_module(module_name) + try: + getattr(module, digestname) + except AttributeError: + fullname = f'{module_name}.{digestname}' + _missing_hash(fullname, implementation="HACL") + return func(*args, **kwargs) + return wrapper + + def decorator(func_or_class): + return _decorate_func_or_class(func_or_class, decorator_func) + return decorator + + +def find_builtin_hashdigest_constructor( + module_name, digestname, *, usedforsecurity=True +): + """Find the HACL* hash function constructor. + + - The *module_name* is the C extension module name based on HACL*. + - The *digestname* is one of its member, e.g., 'md5'. + """ + module = import_module(module_name) + try: + constructor = getattr(module, digestname) + constructor(b'', usedforsecurity=usedforsecurity) + except (AttributeError, TypeError, ValueError): + _missing_hash(f'{module_name}.{digestname}', implementation="HACL") + return constructor + + +class HashFunctionsTrait: + """Mixin trait class containing hash functions. + + This class is assumed to have all unitest.TestCase methods but should + not directly inherit from it to prevent the test suite being run on it. + + Subclasses should implement the hash functions by returning an object + that can be recognized as a valid digestmod parameter for both hashlib + and HMAC. In particular, it cannot be a lambda function as it will not + be recognized by hashlib (it will still be accepted by the pure Python + implementation of HMAC). + """ + + ALGORITHMS = [ + 'md5', 'sha1', + 'sha224', 'sha256', 'sha384', 'sha512', + 'sha3_224', 'sha3_256', 'sha3_384', 'sha3_512', + ] + + # Default 'usedforsecurity' to use when looking up a hash function. + usedforsecurity = True + + def _find_constructor(self, name): + # By default, a missing algorithm skips the test that uses it. + self.assertIn(name, self.ALGORITHMS) + self.skipTest(f"missing hash function: {name}") + + @property + def md5(self): + return self._find_constructor("md5") + + @property + def sha1(self): + return self._find_constructor("sha1") + + @property + def sha224(self): + return self._find_constructor("sha224") + + @property + def sha256(self): + return self._find_constructor("sha256") + + @property + def sha384(self): + return self._find_constructor("sha384") + + @property + def sha512(self): + return self._find_constructor("sha512") + + @property + def sha3_224(self): + return self._find_constructor("sha3_224") + + @property + def sha3_256(self): + return self._find_constructor("sha3_256") + + @property + def sha3_384(self): + return self._find_constructor("sha3_384") + + @property + def sha3_512(self): + return self._find_constructor("sha3_512") + + +class NamedHashFunctionsTrait(HashFunctionsTrait): + """Trait containing named hash functions. + + Hash functions are available if and only if they are available in hashlib. + """ + + def _find_constructor(self, name): + self.assertIn(name, self.ALGORITHMS) + return name + + +class OpenSSLHashFunctionsTrait(HashFunctionsTrait): + """Trait containing OpenSSL hash functions. + + Hash functions are available if and only if they are available in _hashlib. + """ + + def _find_constructor(self, name): + self.assertIn(name, self.ALGORITHMS) + return find_openssl_hashdigest_constructor( + name, usedforsecurity=self.usedforsecurity + ) + + +class BuiltinHashFunctionsTrait(HashFunctionsTrait): + """Trait containing HACL* hash functions. + + Hash functions are available if and only if they are available in C. + In particular, HACL* HMAC-MD5 may be available even though HACL* md5 + is not since the former is unconditionally built. + """ + + def _find_constructor_in(self, module, name): + self.assertIn(name, self.ALGORITHMS) + return find_builtin_hashdigest_constructor(module, name) + + @property + def md5(self): + return self._find_constructor_in("_md5", "md5") + + @property + def sha1(self): + return self._find_constructor_in("_sha1", "sha1") + + @property + def sha224(self): + return self._find_constructor_in("_sha2", "sha224") + + @property + def sha256(self): + return self._find_constructor_in("_sha2", "sha256") + + @property + def sha384(self): + return self._find_constructor_in("_sha2", "sha384") + + @property + def sha512(self): + return self._find_constructor_in("_sha2", "sha512") + + @property + def sha3_224(self): + return self._find_constructor_in("_sha3", "sha3_224") + + @property + def sha3_256(self): + return self._find_constructor_in("_sha3","sha3_256") + + @property + def sha3_384(self): + return self._find_constructor_in("_sha3","sha3_384") + + @property + def sha3_512(self): + return self._find_constructor_in("_sha3","sha3_512") + + +def find_gil_minsize(modules_names, default=2048): + """Get the largest GIL_MINSIZE value for the given cryptographic modules. + + The valid module names are the following: + + - _hashlib + - _md5, _sha1, _sha2, _sha3, _blake2 + - _hmac + """ + sizes = [] + for module_name in modules_names: + try: + module = importlib.import_module(module_name) + except ImportError: + continue + sizes.append(getattr(module, '_GIL_MINSIZE', default)) + return max(sizes, default=default) diff --git a/Lib/test/support/hypothesis_helper.py b/Lib/test/support/hypothesis_helper.py index a99a4963f..6e9e168f6 100644 --- a/Lib/test/support/hypothesis_helper.py +++ b/Lib/test/support/hypothesis_helper.py @@ -7,9 +7,10 @@ except ImportError: else: # Regrtest changes to use a tempdir as the working directory, so we have # to tell Hypothesis to use the original in order to persist the database. + from hypothesis.configuration import set_hypothesis_home_dir + from test.support import has_socket_support from test.support.os_helper import SAVEDCWD - from hypothesis.configuration import set_hypothesis_home_dir set_hypothesis_home_dir(os.path.join(SAVEDCWD, ".hypothesis")) diff --git a/Lib/test/support/i18n_helper.py b/Lib/test/support/i18n_helper.py index 2e304f29e..af97cdc9c 100644 --- a/Lib/test/support/i18n_helper.py +++ b/Lib/test/support/i18n_helper.py @@ -3,10 +3,10 @@ import subprocess import sys import unittest from pathlib import Path + from test.support import REPO_ROOT, TEST_HOME_DIR, requires_subprocess from test.test_tools import skip_if_missing - pygettext = Path(REPO_ROOT) / 'Tools' / 'i18n' / 'pygettext.py' msgid_pattern = re.compile(r'msgid(.*?)(?:msgid_plural|msgctxt|msgstr)', diff --git a/Lib/test/support/import_helper.py b/Lib/test/support/import_helper.py index a4ea5bc6a..d31713b86 100644 --- a/Lib/test/support/import_helper.py +++ b/Lib/test/support/import_helper.py @@ -1,14 +1,16 @@ -import contextlib import _imp +import contextlib import importlib +import importlib.machinery import importlib.util import os import shutil import sys +import textwrap import unittest import warnings -from .os_helper import unlink, temp_dir +from .os_helper import temp_dir, unlink @contextlib.contextmanager @@ -303,9 +305,138 @@ def ready_to_import(name=None, source=""): try: sys.path.insert(0, tempdir) yield name, path - finally: sys.path.remove(tempdir) + finally: if old_module is not None: sys.modules[name] = old_module else: sys.modules.pop(name, None) + + +def ensure_lazy_imports(imported_module, modules_to_block): + """Test that when imported_module is imported, none of the modules in + modules_to_block are imported as a side effect.""" + modules_to_block = frozenset(modules_to_block) + script = textwrap.dedent( + f""" + import sys + modules_to_block = {modules_to_block} + if unexpected := modules_to_block & sys.modules.keys(): + startup = ", ".join(unexpected) + raise AssertionError(f'unexpectedly imported at startup: {{startup}}') + + import {imported_module} + if unexpected := modules_to_block & sys.modules.keys(): + after = ", ".join(unexpected) + raise AssertionError(f'unexpectedly imported after importing {imported_module}: {{after}}') + """ + ) + from .script_helper import assert_python_ok + assert_python_ok("-S", "-c", script) + + +@contextlib.contextmanager +def module_restored(name): + """A context manager that restores a module to the original state.""" + missing = object() + orig = sys.modules.get(name, missing) + if orig is None: + mod = importlib.import_module(name) + else: + mod = type(sys)(name) + mod.__dict__.update(orig.__dict__) + sys.modules[name] = mod + try: + yield mod + finally: + if orig is missing: + sys.modules.pop(name, None) + else: + sys.modules[name] = orig + + +def create_module(name, loader=None, *, ispkg=False): + """Return a new, empty module.""" + spec = importlib.machinery.ModuleSpec( + name, + loader, + origin='', + is_package=ispkg, + ) + return importlib.util.module_from_spec(spec) + + +def _ensure_module(name, ispkg, addparent, clearnone): + try: + mod = orig = sys.modules[name] + except KeyError: + mod = orig = None + missing = True + else: + missing = False + if mod is not None: + # It was already imported. + return mod, orig, missing + # Otherwise, None means it was explicitly disabled. + + assert name != '__main__' + if not missing: + assert orig is None, (name, sys.modules[name]) + if not clearnone: + raise ModuleNotFoundError(name) + del sys.modules[name] + # Try normal import, then fall back to adding the module. + try: + mod = importlib.import_module(name) + except ModuleNotFoundError: + if addparent and not clearnone: + addparent = None + mod = _add_module(name, ispkg, addparent) + return mod, orig, missing + + +def _add_module(spec, ispkg, addparent): + if isinstance(spec, str): + name = spec + mod = create_module(name, ispkg=ispkg) + spec = mod.__spec__ + else: + name = spec.name + mod = importlib.util.module_from_spec(spec) + sys.modules[name] = mod + if addparent is not False and spec.parent: + _ensure_module(spec.parent, True, addparent, bool(addparent)) + return mod + + +def add_module(spec, *, parents=True): + """Return the module after creating it and adding it to sys.modules. + + If parents is True then also create any missing parents. + """ + return _add_module(spec, False, parents) + + +def add_package(spec, *, parents=True): + """Return the module after creating it and adding it to sys.modules. + + If parents is True then also create any missing parents. + """ + return _add_module(spec, True, parents) + + +def ensure_module_imported(name, *, clearnone=True): + """Return the corresponding module. + + If it was already imported then return that. Otherwise, try + importing it (optionally clear it first if None). If that fails + then create a new empty module. + + It can be helpful to combine this with ready_to_import() and/or + isolated_modules(). + """ + if sys.modules.get(name) is not None: + mod = sys.modules[name] + else: + mod, _, _ = _ensure_module(name, False, True, clearnone) + return mod diff --git a/Lib/test/support/interpreters/__init__.py b/Lib/test/support/interpreters/__init__.py deleted file mode 100644 index e067f2593..000000000 --- a/Lib/test/support/interpreters/__init__.py +++ /dev/null @@ -1,258 +0,0 @@ -"""Subinterpreters High Level Module.""" - -import threading -import weakref -import _interpreters - -# aliases: -from _interpreters import ( - InterpreterError, InterpreterNotFoundError, NotShareableError, - is_shareable, -) - - -__all__ = [ - 'get_current', 'get_main', 'create', 'list_all', 'is_shareable', - 'Interpreter', - 'InterpreterError', 'InterpreterNotFoundError', 'ExecutionFailed', - 'NotShareableError', - 'create_queue', 'Queue', 'QueueEmpty', 'QueueFull', -] - - -_queuemod = None - -def __getattr__(name): - if name in ('Queue', 'QueueEmpty', 'QueueFull', 'create_queue'): - global create_queue, Queue, QueueEmpty, QueueFull - ns = globals() - from .queues import ( - create as create_queue, - Queue, QueueEmpty, QueueFull, - ) - return ns[name] - else: - raise AttributeError(name) - - -_EXEC_FAILURE_STR = """ -{superstr} - -Uncaught in the interpreter: - -{formatted} -""".strip() - -class ExecutionFailed(InterpreterError): - """An unhandled exception happened during execution. - - This is raised from Interpreter.exec() and Interpreter.call(). - """ - - def __init__(self, excinfo): - msg = excinfo.formatted - if not msg: - if excinfo.type and excinfo.msg: - msg = f'{excinfo.type.__name__}: {excinfo.msg}' - else: - msg = excinfo.type.__name__ or excinfo.msg - super().__init__(msg) - self.excinfo = excinfo - - def __str__(self): - try: - formatted = self.excinfo.errdisplay - except Exception: - return super().__str__() - else: - return _EXEC_FAILURE_STR.format( - superstr=super().__str__(), - formatted=formatted, - ) - - -def create(): - """Return a new (idle) Python interpreter.""" - id = _interpreters.create(reqrefs=True) - return Interpreter(id, _ownsref=True) - - -def list_all(): - """Return all existing interpreters.""" - return [Interpreter(id, _whence=whence) - for id, whence in _interpreters.list_all(require_ready=True)] - - -def get_current(): - """Return the currently running interpreter.""" - id, whence = _interpreters.get_current() - return Interpreter(id, _whence=whence) - - -def get_main(): - """Return the main interpreter.""" - id, whence = _interpreters.get_main() - assert whence == _interpreters.WHENCE_RUNTIME, repr(whence) - return Interpreter(id, _whence=whence) - - -_known = weakref.WeakValueDictionary() - -class Interpreter: - """A single Python interpreter. - - Attributes: - - "id" - the unique process-global ID number for the interpreter - "whence" - indicates where the interpreter was created - - If the interpreter wasn't created by this module - then any method that modifies the interpreter will fail, - i.e. .close(), .prepare_main(), .exec(), and .call() - """ - - _WHENCE_TO_STR = { - _interpreters.WHENCE_UNKNOWN: 'unknown', - _interpreters.WHENCE_RUNTIME: 'runtime init', - _interpreters.WHENCE_LEGACY_CAPI: 'legacy C-API', - _interpreters.WHENCE_CAPI: 'C-API', - _interpreters.WHENCE_XI: 'cross-interpreter C-API', - _interpreters.WHENCE_STDLIB: '_interpreters module', - } - - def __new__(cls, id, /, _whence=None, _ownsref=None): - # There is only one instance for any given ID. - if not isinstance(id, int): - raise TypeError(f'id must be an int, got {id!r}') - id = int(id) - if _whence is None: - if _ownsref: - _whence = _interpreters.WHENCE_STDLIB - else: - _whence = _interpreters.whence(id) - assert _whence in cls._WHENCE_TO_STR, repr(_whence) - if _ownsref is None: - _ownsref = (_whence == _interpreters.WHENCE_STDLIB) - try: - self = _known[id] - assert hasattr(self, '_ownsref') - except KeyError: - self = super().__new__(cls) - _known[id] = self - self._id = id - self._whence = _whence - self._ownsref = _ownsref - if _ownsref: - # This may raise InterpreterNotFoundError: - _interpreters.incref(id) - return self - - def __repr__(self): - return f'{type(self).__name__}({self.id})' - - def __hash__(self): - return hash(self._id) - - def __del__(self): - self._decref() - - # for pickling: - def __getnewargs__(self): - return (self._id,) - - # for pickling: - def __getstate__(self): - return None - - def _decref(self): - if not self._ownsref: - return - self._ownsref = False - try: - _interpreters.decref(self._id) - except InterpreterNotFoundError: - pass - - @property - def id(self): - return self._id - - @property - def whence(self): - return self._WHENCE_TO_STR[self._whence] - - def is_running(self): - """Return whether or not the identified interpreter is running.""" - return _interpreters.is_running(self._id) - - # Everything past here is available only to interpreters created by - # interpreters.create(). - - def close(self): - """Finalize and destroy the interpreter. - - Attempting to destroy the current interpreter results - in an InterpreterError. - """ - return _interpreters.destroy(self._id, restrict=True) - - def prepare_main(self, ns=None, /, **kwargs): - """Bind the given values into the interpreter's __main__. - - The values must be shareable. - """ - ns = dict(ns, **kwargs) if ns is not None else kwargs - _interpreters.set___main___attrs(self._id, ns, restrict=True) - - def exec(self, code, /): - """Run the given source code in the interpreter. - - This is essentially the same as calling the builtin "exec" - with this interpreter, using the __dict__ of its __main__ - module as both globals and locals. - - There is no return value. - - If the code raises an unhandled exception then an ExecutionFailed - exception is raised, which summarizes the unhandled exception. - The actual exception is discarded because objects cannot be - shared between interpreters. - - This blocks the current Python thread until done. During - that time, the previous interpreter is allowed to run - in other threads. - """ - excinfo = _interpreters.exec(self._id, code, restrict=True) - if excinfo is not None: - raise ExecutionFailed(excinfo) - - def call(self, callable, /): - """Call the object in the interpreter with given args/kwargs. - - Only functions that take no arguments and have no closure - are supported. - - The return value is discarded. - - If the callable raises an exception then the error display - (including full traceback) is send back between the interpreters - and an ExecutionFailed exception is raised, much like what - happens with Interpreter.exec(). - """ - # XXX Support args and kwargs. - # XXX Support arbitrary callables. - # XXX Support returning the return value (e.g. via pickle). - excinfo = _interpreters.call(self._id, callable, restrict=True) - if excinfo is not None: - raise ExecutionFailed(excinfo) - - def call_in_thread(self, callable, /): - """Return a new thread that calls the object in the interpreter. - - The return value and any raised exception are discarded. - """ - def task(): - self.call(callable) - t = threading.Thread(target=task) - t.start() - return t diff --git a/Lib/test/support/interpreters/_crossinterp.py b/Lib/test/support/interpreters/_crossinterp.py deleted file mode 100644 index 544e197ba..000000000 --- a/Lib/test/support/interpreters/_crossinterp.py +++ /dev/null @@ -1,102 +0,0 @@ -"""Common code between queues and channels.""" - - -class ItemInterpreterDestroyed(Exception): - """Raised when trying to get an item whose interpreter was destroyed.""" - - -class classonly: - """A non-data descriptor that makes a value only visible on the class. - - This is like the "classmethod" builtin, but does not show up on - instances of the class. It may be used as a decorator. - """ - - def __init__(self, value): - self.value = value - self.getter = classmethod(value).__get__ - self.name = None - - def __set_name__(self, cls, name): - if self.name is not None: - raise TypeError('already used') - self.name = name - - def __get__(self, obj, cls): - if obj is not None: - raise AttributeError(self.name) - # called on the class - return self.getter(None, cls) - - -class UnboundItem: - """Represents a cross-interpreter item no longer bound to an interpreter. - - An item is unbound when the interpreter that added it to the - cross-interpreter container is destroyed. - """ - - __slots__ = () - - @classonly - def singleton(cls, kind, module, name='UNBOUND'): - doc = cls.__doc__.replace('cross-interpreter container', kind) - doc = doc.replace('cross-interpreter', kind) - subclass = type( - f'Unbound{kind.capitalize()}Item', - (cls,), - dict( - _MODULE=module, - _NAME=name, - __doc__=doc, - ), - ) - return object.__new__(subclass) - - _MODULE = __name__ - _NAME = 'UNBOUND' - - def __new__(cls): - raise Exception(f'use {cls._MODULE}.{cls._NAME}') - - def __repr__(self): - return f'{self._MODULE}.{self._NAME}' -# return f'interpreters.queues.UNBOUND' - - -UNBOUND = object.__new__(UnboundItem) -UNBOUND_ERROR = object() -UNBOUND_REMOVE = object() - -_UNBOUND_CONSTANT_TO_FLAG = { - UNBOUND_REMOVE: 1, - UNBOUND_ERROR: 2, - UNBOUND: 3, -} -_UNBOUND_FLAG_TO_CONSTANT = {v: k - for k, v in _UNBOUND_CONSTANT_TO_FLAG.items()} - - -def serialize_unbound(unbound): - op = unbound - try: - flag = _UNBOUND_CONSTANT_TO_FLAG[op] - except KeyError: - raise NotImplementedError(f'unsupported unbound replacement op {op!r}') - return flag, - - -def resolve_unbound(flag, exctype_destroyed): - try: - op = _UNBOUND_FLAG_TO_CONSTANT[flag] - except KeyError: - raise NotImplementedError(f'unsupported unbound replacement op {flag!r}') - if op is UNBOUND_REMOVE: - # "remove" not possible here - raise NotImplementedError - elif op is UNBOUND_ERROR: - raise exctype_destroyed("item's original interpreter destroyed") - elif op is UNBOUND: - return UNBOUND - else: - raise NotImplementedError(repr(op)) diff --git a/Lib/test/support/interpreters/queues.py b/Lib/test/support/interpreters/queues.py deleted file mode 100644 index deb8e8613..000000000 --- a/Lib/test/support/interpreters/queues.py +++ /dev/null @@ -1,313 +0,0 @@ -"""Cross-interpreter Queues High Level Module.""" - -import pickle -import queue -import time -import weakref -import _interpqueues as _queues -from . import _crossinterp - -# aliases: -from _interpqueues import ( - QueueError, QueueNotFoundError, -) -from ._crossinterp import ( - UNBOUND_ERROR, UNBOUND_REMOVE, -) - -__all__ = [ - 'UNBOUND', 'UNBOUND_ERROR', 'UNBOUND_REMOVE', - 'create', 'list_all', - 'Queue', - 'QueueError', 'QueueNotFoundError', 'QueueEmpty', 'QueueFull', - 'ItemInterpreterDestroyed', -] - - -class QueueEmpty(QueueError, queue.Empty): - """Raised from get_nowait() when the queue is empty. - - It is also raised from get() if it times out. - """ - - -class QueueFull(QueueError, queue.Full): - """Raised from put_nowait() when the queue is full. - - It is also raised from put() if it times out. - """ - - -class ItemInterpreterDestroyed(QueueError, - _crossinterp.ItemInterpreterDestroyed): - """Raised from get() and get_nowait().""" - - -_SHARED_ONLY = 0 -_PICKLED = 1 - - -UNBOUND = _crossinterp.UnboundItem.singleton('queue', __name__) - - -def _serialize_unbound(unbound): - if unbound is UNBOUND: - unbound = _crossinterp.UNBOUND - return _crossinterp.serialize_unbound(unbound) - - -def _resolve_unbound(flag): - resolved = _crossinterp.resolve_unbound(flag, ItemInterpreterDestroyed) - if resolved is _crossinterp.UNBOUND: - resolved = UNBOUND - return resolved - - -def create(maxsize=0, *, syncobj=False, unbounditems=UNBOUND): - """Return a new cross-interpreter queue. - - The queue may be used to pass data safely between interpreters. - - "syncobj" sets the default for Queue.put() - and Queue.put_nowait(). - - "unbounditems" likewise sets the default. See Queue.put() for - supported values. The default value is UNBOUND, which replaces - the unbound item. - """ - fmt = _SHARED_ONLY if syncobj else _PICKLED - unbound = _serialize_unbound(unbounditems) - unboundop, = unbound - qid = _queues.create(maxsize, fmt, unboundop) - return Queue(qid, _fmt=fmt, _unbound=unbound) - - -def list_all(): - """Return a list of all open queues.""" - return [Queue(qid, _fmt=fmt, _unbound=(unboundop,)) - for qid, fmt, unboundop in _queues.list_all()] - - -_known_queues = weakref.WeakValueDictionary() - -class Queue: - """A cross-interpreter queue.""" - - def __new__(cls, id, /, *, _fmt=None, _unbound=None): - # There is only one instance for any given ID. - if isinstance(id, int): - id = int(id) - else: - raise TypeError(f'id must be an int, got {id!r}') - if _fmt is None: - if _unbound is None: - _fmt, op = _queues.get_queue_defaults(id) - _unbound = (op,) - else: - _fmt, _ = _queues.get_queue_defaults(id) - elif _unbound is None: - _, op = _queues.get_queue_defaults(id) - _unbound = (op,) - try: - self = _known_queues[id] - except KeyError: - self = super().__new__(cls) - self._id = id - self._fmt = _fmt - self._unbound = _unbound - _known_queues[id] = self - _queues.bind(id) - return self - - def __del__(self): - try: - _queues.release(self._id) - except QueueNotFoundError: - pass - try: - del _known_queues[self._id] - except KeyError: - pass - - def __repr__(self): - return f'{type(self).__name__}({self.id})' - - def __hash__(self): - return hash(self._id) - - # for pickling: - def __getnewargs__(self): - return (self._id,) - - # for pickling: - def __getstate__(self): - return None - - @property - def id(self): - return self._id - - @property - def maxsize(self): - try: - return self._maxsize - except AttributeError: - self._maxsize = _queues.get_maxsize(self._id) - return self._maxsize - - def empty(self): - return self.qsize() == 0 - - def full(self): - return _queues.is_full(self._id) - - def qsize(self): - return _queues.get_count(self._id) - - def put(self, obj, timeout=None, *, - syncobj=None, - unbound=None, - _delay=10 / 1000, # 10 milliseconds - ): - """Add the object to the queue. - - This blocks while the queue is full. - - If "syncobj" is None (the default) then it uses the - queue's default, set with create_queue(). - - If "syncobj" is false then all objects are supported, - at the expense of worse performance. - - If "syncobj" is true then the object must be "shareable". - Examples of "shareable" objects include the builtin singletons, - str, and memoryview. One benefit is that such objects are - passed through the queue efficiently. - - The key difference, though, is conceptual: the corresponding - object returned from Queue.get() will be strictly equivalent - to the given obj. In other words, the two objects will be - effectively indistinguishable from each other, even if the - object is mutable. The received object may actually be the - same object, or a copy (immutable values only), or a proxy. - Regardless, the received object should be treated as though - the original has been shared directly, whether or not it - actually is. That's a slightly different and stronger promise - than just (initial) equality, which is all "syncobj=False" - can promise. - - "unbound" controls the behavior of Queue.get() for the given - object if the current interpreter (calling put()) is later - destroyed. - - If "unbound" is None (the default) then it uses the - queue's default, set with create_queue(), - which is usually UNBOUND. - - If "unbound" is UNBOUND_ERROR then get() will raise an - ItemInterpreterDestroyed exception if the original interpreter - has been destroyed. This does not otherwise affect the queue; - the next call to put() will work like normal, returning the next - item in the queue. - - If "unbound" is UNBOUND_REMOVE then the item will be removed - from the queue as soon as the original interpreter is destroyed. - Be aware that this will introduce an imbalance between put() - and get() calls. - - If "unbound" is UNBOUND then it is returned by get() in place - of the unbound item. - """ - if syncobj is None: - fmt = self._fmt - else: - fmt = _SHARED_ONLY if syncobj else _PICKLED - if unbound is None: - unboundop, = self._unbound - else: - unboundop, = _serialize_unbound(unbound) - if timeout is not None: - timeout = int(timeout) - if timeout < 0: - raise ValueError(f'timeout value must be non-negative') - end = time.time() + timeout - if fmt is _PICKLED: - obj = pickle.dumps(obj) - while True: - try: - _queues.put(self._id, obj, fmt, unboundop) - except QueueFull as exc: - if timeout is not None and time.time() >= end: - raise # re-raise - time.sleep(_delay) - else: - break - - def put_nowait(self, obj, *, syncobj=None, unbound=None): - if syncobj is None: - fmt = self._fmt - else: - fmt = _SHARED_ONLY if syncobj else _PICKLED - if unbound is None: - unboundop, = self._unbound - else: - unboundop, = _serialize_unbound(unbound) - if fmt is _PICKLED: - obj = pickle.dumps(obj) - _queues.put(self._id, obj, fmt, unboundop) - - def get(self, timeout=None, *, - _delay=10 / 1000, # 10 milliseconds - ): - """Return the next object from the queue. - - This blocks while the queue is empty. - - If the next item's original interpreter has been destroyed - then the "next object" is determined by the value of the - "unbound" argument to put(). - """ - if timeout is not None: - timeout = int(timeout) - if timeout < 0: - raise ValueError(f'timeout value must be non-negative') - end = time.time() + timeout - while True: - try: - obj, fmt, unboundop = _queues.get(self._id) - except QueueEmpty as exc: - if timeout is not None and time.time() >= end: - raise # re-raise - time.sleep(_delay) - else: - break - if unboundop is not None: - assert obj is None, repr(obj) - return _resolve_unbound(unboundop) - if fmt == _PICKLED: - obj = pickle.loads(obj) - else: - assert fmt == _SHARED_ONLY - return obj - - def get_nowait(self): - """Return the next object from the channel. - - If the queue is empty then raise QueueEmpty. Otherwise this - is the same as get(). - """ - try: - obj, fmt, unboundop = _queues.get(self._id) - except QueueEmpty as exc: - raise # re-raise - if unboundop is not None: - assert obj is None, repr(obj) - return _resolve_unbound(unboundop) - if fmt == _PICKLED: - obj = pickle.loads(obj) - else: - assert fmt == _SHARED_ONLY - return obj - - -_queues._register_heap_types(Queue, QueueEmpty, QueueFull) diff --git a/Lib/test/support/logging_helper.py b/Lib/test/support/logging_helper.py index 12fcca4f0..db556c7f5 100644 --- a/Lib/test/support/logging_helper.py +++ b/Lib/test/support/logging_helper.py @@ -1,5 +1,6 @@ import logging.handlers + class TestHandler(logging.handlers.BufferingHandler): def __init__(self, matcher): # BufferingHandler takes a "capacity" argument diff --git a/Lib/test/support/os_helper.py b/Lib/test/support/os_helper.py index 26c467a7a..d3d6fa632 100644 --- a/Lib/test/support/os_helper.py +++ b/Lib/test/support/os_helper.py @@ -13,7 +13,6 @@ import warnings from test import support - # Filename used for testing TESTFN_ASCII = '@test' @@ -295,6 +294,33 @@ def skip_unless_working_chmod(test): return test if ok else unittest.skip(msg)(test) +@contextlib.contextmanager +def save_mode(path, *, quiet=False): + """Context manager that restores the mode (permissions) of *path* on exit. + + Arguments: + + path: Path of the file to restore the mode of. + + quiet: if False (the default), the context manager raises an exception + on error. Otherwise, it issues only a warning and keeps the current + working directory the same. + + """ + saved_mode = os.stat(path) + try: + yield + finally: + try: + os.chmod(path, saved_mode.st_mode) + except OSError as exc: + if not quiet: + raise + warnings.warn(f'tests may fail, unable to restore the mode of ' + f'{path!r} to {saved_mode.st_mode}: {exc}', + RuntimeWarning, stacklevel=3) + + # Check whether the current effective user has the capability to override # DAC (discretionary access control). Typically user root is able to # bypass file read, write, and execute permission checks. The capability diff --git a/Lib/test/support/pty_helper.py b/Lib/test/support/pty_helper.py index 6587fd403..7e1ae9e59 100644 --- a/Lib/test/support/pty_helper.py +++ b/Lib/test/support/pty_helper.py @@ -10,6 +10,7 @@ from errno import EIO from test.support.import_helper import import_module + def run_pty(script, input=b"dummy input\r", env=None): pty = import_module('pty') output = bytearray() diff --git a/Lib/test/support/script_helper.py b/Lib/test/support/script_helper.py index 04458077d..a338f4844 100644 --- a/Lib/test/support/script_helper.py +++ b/Lib/test/support/script_helper.py @@ -3,17 +3,16 @@ import collections import importlib -import sys import os import os.path -import subprocess import py_compile - +import subprocess +import sys from importlib.util import source_from_cache + from test import support from test.support.import_helper import make_legacy_pyc - # Cached result of the expensive test performed in the function below. __cached_interp_requires_environment = None @@ -70,23 +69,25 @@ class _PythonRunResult(collections.namedtuple("_PythonRunResult", out = b'(... truncated stdout ...)' + out[-maxlen:] if len(err) > maxlen: err = b'(... truncated stderr ...)' + err[-maxlen:] - out = out.decode('ascii', 'replace').rstrip() - err = err.decode('ascii', 'replace').rstrip() - raise AssertionError("Process return code is %d\n" - "command line: %r\n" - "\n" - "stdout:\n" - "---\n" - "%s\n" - "---\n" - "\n" - "stderr:\n" - "---\n" - "%s\n" - "---" - % (self.rc, cmd_line, - out, - err)) + out = out.decode('utf8', 'replace').rstrip() + err = err.decode('utf8', 'replace').rstrip() + + exitcode = self.rc + signame = support.get_signal_name(exitcode) + if signame: + exitcode = f"{exitcode} ({signame})" + raise AssertionError(f"Process return code is {exitcode}\n" + f"command line: {cmd_line!r}\n" + f"\n" + f"stdout:\n" + f"---\n" + f"{out}\n" + f"---\n" + f"\n" + f"stderr:\n" + f"---\n" + f"{err}\n" + f"---") # Executing the interpreter in a subprocess diff --git a/Lib/test/support/smtpd.py b/Lib/test/support/smtpd.py old mode 100644 new mode 100755 index 6052232ec..cf333aaf6 --- a/Lib/test/support/smtpd.py +++ b/Lib/test/support/smtpd.py @@ -7,7 +7,7 @@ Options: --nosetuid -n - This program generally tries to setuid `nobody', unless this flag is + This program generally tries to setuid 'nobody', unless this flag is set. The setuid call will fail if this program is not run as root (in which case, use this flag). @@ -17,7 +17,7 @@ Options: --class classname -c classname - Use `classname' as the concrete SMTP proxy class. Uses `PureProxy' by + Use 'classname' as the concrete SMTP proxy class. Uses 'PureProxy' by default. --size limit @@ -39,8 +39,8 @@ Options: Version: %(__version__)s -If localhost is not given then `localhost' is used, and if localport is not -given then 8025 is used. If remotehost is not given then `localhost' is used, +If localhost is not given then 'localhost' is used, and if localport is not +given then 8025 is used. If remotehost is not given then 'localhost' is used, and if remoteport is not given, then 25 is used. """ @@ -70,16 +70,17 @@ and if remoteport is not given, then 25 is used. # - Handle more ESMTP extensions # - handle error codes from the backend smtpd -import sys -import os +import collections import errno import getopt -import time +import os import socket -import collections -from test.support import asyncore, asynchat -from warnings import warn +import sys +import time from email._header_value_parser import get_addr_spec, get_angle_addr +from warnings import warn + +from test.support import asynchat, asyncore __all__ = [ "SMTPChannel", "SMTPServer", "DebuggingServer", "PureProxy", @@ -633,7 +634,8 @@ class SMTPServer(asyncore.dispatcher): " be set to True at the same time") asyncore.dispatcher.__init__(self, map=map) try: - gai_results = socket.getaddrinfo(*localaddr, + family = 0 if socket.has_ipv6 else socket.AF_INET + gai_results = socket.getaddrinfo(*localaddr, family=family, type=socket.SOCK_STREAM) self.create_socket(gai_results[0][0], gai_results[0][1]) # try to re-use a server port if possible @@ -672,9 +674,9 @@ class SMTPServer(asyncore.dispatcher): message to. data is a string containing the entire full text of the message, - headers (if supplied) and all. It has been `de-transparencied' + headers (if supplied) and all. It has been 'de-transparencied' according to RFC 821, Section 4.5.2. In other words, a line - containing a `.' followed by other text has had the leading dot + containing a '.' followed by other text has had the leading dot removed. kwargs is a dictionary containing additional information. It is @@ -685,7 +687,7 @@ class SMTPServer(asyncore.dispatcher): ['BODY=8BITMIME', 'SMTPUTF8']. 'rcpt_options': same, for the rcpt command. - This function should return None for a normal `250 Ok' response; + This function should return None for a normal '250 Ok' response; otherwise, it should return the desired response string in RFC 821 format. diff --git a/Lib/test/support/socket_helper.py b/Lib/test/support/socket_helper.py index a41e487f3..655ffbea0 100644 --- a/Lib/test/support/socket_helper.py +++ b/Lib/test/support/socket_helper.py @@ -2,8 +2,8 @@ import contextlib import errno import os.path import socket -import sys import subprocess +import sys import tempfile import unittest diff --git a/Lib/test/support/strace_helper.py b/Lib/test/support/strace_helper.py new file mode 100644 index 000000000..abc93dee2 --- /dev/null +++ b/Lib/test/support/strace_helper.py @@ -0,0 +1,210 @@ +import os +import re +import sys +import textwrap +import unittest +from dataclasses import dataclass +from functools import cache + +from test import support +from test.support.script_helper import run_python_until_end + +_strace_binary = "/usr/bin/strace" +_syscall_regex = re.compile( + r"(?P[^(]*)\((?P[^)]*)\)\s*[=]\s*(?P.+)") +_returncode_regex = re.compile( + br"\+\+\+ exited with (?P\d+) \+\+\+") + + +@dataclass +class StraceEvent: + syscall: str + args: list[str] + returncode: str + + +@dataclass +class StraceResult: + strace_returncode: int + python_returncode: int + + """The event messages generated by strace. This is very similar to the + stderr strace produces with returncode marker section removed.""" + event_bytes: bytes + stdout: bytes + stderr: bytes + + def events(self): + """Parse event_bytes data into system calls for easier processing. + + This assumes the program under inspection doesn't print any non-utf8 + strings which would mix into the strace output.""" + decoded_events = self.event_bytes.decode('utf-8', 'surrogateescape') + matches = [ + _syscall_regex.match(event) + for event in decoded_events.splitlines() + ] + return [ + StraceEvent(match["syscall"], + [arg.strip() for arg in (match["args"].split(","))], + match["returncode"]) for match in matches if match + ] + + def sections(self): + """Find all "MARK " writes and use them to make groups of events. + + This is useful to avoid variable / overhead events, like those at + interpreter startup or when opening a file so a test can verify just + the small case under study.""" + current_section = "__startup" + sections = {current_section: []} + for event in self.events(): + if event.syscall == 'write' and len( + event.args) > 2 and event.args[1].startswith("\"MARK "): + # Found a new section, don't include the write in the section + # but all events until next mark should be in that section + current_section = event.args[1].split( + " ", 1)[1].removesuffix('\\n"') + if current_section not in sections: + sections[current_section] = list() + else: + sections[current_section].append(event) + + return sections + +def _filter_memory_call(call): + # mmap can operate on a fd or "MAP_ANONYMOUS" which gives a block of memory. + # Ignore "MAP_ANONYMOUS + the "MAP_ANON" alias. + if call.syscall == "mmap" and "MAP_ANON" in call.args[3]: + return True + + if call.syscall in ("munmap", "mprotect"): + return True + + return False + + +def filter_memory(syscalls): + """Filter out memory allocation calls from File I/O calls. + + Some calls (mmap, munmap, etc) can be used on files or to just get a block + of memory. Use this function to filter out the memory related calls from + other calls.""" + + return [call for call in syscalls if not _filter_memory_call(call)] + + +@support.requires_subprocess() +def strace_python(code, strace_flags, check=True): + """Run strace and return the trace. + + Sets strace_returncode and python_returncode to `-1` on error.""" + res = None + + def _make_error(reason, details): + return StraceResult( + strace_returncode=-1, + python_returncode=-1, + event_bytes= f"error({reason},details={details!r}) = -1".encode('utf-8'), + stdout=res.out if res else b"", + stderr=res.err if res else b"") + + # Run strace, and get out the raw text + try: + res, cmd_line = run_python_until_end( + "-c", + textwrap.dedent(code), + __run_using_command=[_strace_binary] + strace_flags, + ) + except OSError as err: + return _make_error("Caught OSError", err) + + if check and res.rc: + res.fail(cmd_line) + + # Get out program returncode + stripped = res.err.strip() + output = stripped.rsplit(b"\n", 1) + if len(output) != 2: + return _make_error("Expected strace events and exit code line", + stripped[-50:]) + + returncode_match = _returncode_regex.match(output[1]) + if not returncode_match: + return _make_error("Expected to find returncode in last line.", + output[1][:50]) + + python_returncode = int(returncode_match["returncode"]) + if check and python_returncode: + res.fail(cmd_line) + + return StraceResult(strace_returncode=res.rc, + python_returncode=python_returncode, + event_bytes=output[0], + stdout=res.out, + stderr=res.err) + + +def get_events(code, strace_flags, prelude, cleanup): + # NOTE: The flush is currently required to prevent the prints from getting + # buffered and done all at once at exit + prelude = textwrap.dedent(prelude) + code = textwrap.dedent(code) + cleanup = textwrap.dedent(cleanup) + to_run = f""" +print("MARK prelude", flush=True) +{prelude} +print("MARK code", flush=True) +{code} +print("MARK cleanup", flush=True) +{cleanup} +print("MARK __shutdown", flush=True) + """ + trace = strace_python(to_run, strace_flags) + all_sections = trace.sections() + return all_sections['code'] + + +def get_syscalls(code, strace_flags, prelude="", cleanup="", + ignore_memory=True): + """Get the syscalls which a given chunk of python code generates""" + events = get_events(code, strace_flags, prelude=prelude, cleanup=cleanup) + + if ignore_memory: + events = filter_memory(events) + + return [ev.syscall for ev in events] + + +# Moderately expensive (spawns a subprocess), so share results when possible. +@cache +def _can_strace(): + res = strace_python("import sys; sys.exit(0)", + # --trace option needs strace 5.5 (gh-133741) + ["--trace=%process"], + check=False) + if res.strace_returncode == 0 and res.python_returncode == 0: + assert res.events(), "Should have parsed multiple calls" + return True + return False + + +def requires_strace(): + if sys.platform != "linux": + return unittest.skip("Linux only, requires strace.") + + if "LD_PRELOAD" in os.environ: + # Distribution packaging (ex. Debian `fakeroot` and Gentoo `sandbox`) + # use LD_PRELOAD to intercept system calls, which changes the overall + # set of system calls which breaks tests expecting a specific set of + # system calls). + return unittest.skip("Not supported when LD_PRELOAD is intercepting system calls.") + + if support.check_sanitizer(address=True, memory=True): + return unittest.skip("LeakSanitizer does not work under ptrace (strace, gdb, etc)") + + return unittest.skipUnless(_can_strace(), "Requires working strace") + + +__all__ = ["filter_memory", "get_events", "get_syscalls", "requires_strace", + "strace_python", "StraceEvent", "StraceResult"] diff --git a/Lib/test/support/testcase.py b/Lib/test/support/testcase.py index fd32457d1..e617b19b6 100644 --- a/Lib/test/support/testcase.py +++ b/Lib/test/support/testcase.py @@ -1,6 +1,7 @@ from math import copysign, isnan +# XXX: RUSTPYTHON: removed in 3.14 class ExtraAssertions: def assertIsSubclass(self, cls, superclass, msg=None): diff --git a/Lib/test/support/testresult.py b/Lib/test/support/testresult.py deleted file mode 100644 index de23fdd59..000000000 --- a/Lib/test/support/testresult.py +++ /dev/null @@ -1,191 +0,0 @@ -'''Test runner and result class for the regression test suite. - -''' - -import functools -import io -import sys -import time -import traceback -import unittest -from test import support - -class RegressionTestResult(unittest.TextTestResult): - USE_XML = False - - def __init__(self, stream, descriptions, verbosity): - super().__init__(stream=stream, descriptions=descriptions, - verbosity=2 if verbosity else 0) - self.buffer = True - if self.USE_XML: - from xml.etree import ElementTree as ET - from datetime import datetime, UTC - self.__ET = ET - self.__suite = ET.Element('testsuite') - self.__suite.set('start', - datetime.now(UTC) - .replace(tzinfo=None) - .isoformat(' ')) - self.__e = None - self.__start_time = None - - @classmethod - def __getId(cls, test): - try: - test_id = test.id - except AttributeError: - return str(test) - try: - return test_id() - except TypeError: - return str(test_id) - return repr(test) - - def startTest(self, test): - super().startTest(test) - if self.USE_XML: - self.__e = e = self.__ET.SubElement(self.__suite, 'testcase') - self.__start_time = time.perf_counter() - - def _add_result(self, test, capture=False, **args): - if not self.USE_XML: - return - e = self.__e - self.__e = None - if e is None: - return - ET = self.__ET - - e.set('name', args.pop('name', self.__getId(test))) - e.set('status', args.pop('status', 'run')) - e.set('result', args.pop('result', 'completed')) - if self.__start_time: - e.set('time', f'{time.perf_counter() - self.__start_time:0.6f}') - - if capture: - if self._stdout_buffer is not None: - stdout = self._stdout_buffer.getvalue().rstrip() - ET.SubElement(e, 'system-out').text = stdout - if self._stderr_buffer is not None: - stderr = self._stderr_buffer.getvalue().rstrip() - ET.SubElement(e, 'system-err').text = stderr - - for k, v in args.items(): - if not k or not v: - continue - e2 = ET.SubElement(e, k) - if hasattr(v, 'items'): - for k2, v2 in v.items(): - if k2: - e2.set(k2, str(v2)) - else: - e2.text = str(v2) - else: - e2.text = str(v) - - @classmethod - def __makeErrorDict(cls, err_type, err_value, err_tb): - if isinstance(err_type, type): - if err_type.__module__ == 'builtins': - typename = err_type.__name__ - else: - typename = f'{err_type.__module__}.{err_type.__name__}' - else: - typename = repr(err_type) - - msg = traceback.format_exception(err_type, err_value, None) - tb = traceback.format_exception(err_type, err_value, err_tb) - - return { - 'type': typename, - 'message': ''.join(msg), - '': ''.join(tb), - } - - def addError(self, test, err): - self._add_result(test, True, error=self.__makeErrorDict(*err)) - super().addError(test, err) - - def addExpectedFailure(self, test, err): - self._add_result(test, True, output=self.__makeErrorDict(*err)) - super().addExpectedFailure(test, err) - - def addFailure(self, test, err): - self._add_result(test, True, failure=self.__makeErrorDict(*err)) - super().addFailure(test, err) - if support.failfast: - self.stop() - - def addSkip(self, test, reason): - self._add_result(test, skipped=reason) - super().addSkip(test, reason) - - def addSuccess(self, test): - self._add_result(test) - super().addSuccess(test) - - def addUnexpectedSuccess(self, test): - self._add_result(test, outcome='UNEXPECTED_SUCCESS') - super().addUnexpectedSuccess(test) - - def get_xml_element(self): - if not self.USE_XML: - raise ValueError("USE_XML is false") - e = self.__suite - e.set('tests', str(self.testsRun)) - e.set('errors', str(len(self.errors))) - e.set('failures', str(len(self.failures))) - return e - -class QuietRegressionTestRunner: - def __init__(self, stream, buffer=False): - self.result = RegressionTestResult(stream, None, 0) - self.result.buffer = buffer - - def run(self, test): - test(self.result) - return self.result - -def get_test_runner_class(verbosity, buffer=False): - if verbosity: - return functools.partial(unittest.TextTestRunner, - resultclass=RegressionTestResult, - buffer=buffer, - verbosity=verbosity) - return functools.partial(QuietRegressionTestRunner, buffer=buffer) - -def get_test_runner(stream, verbosity, capture_output=False): - return get_test_runner_class(verbosity, capture_output)(stream) - -if __name__ == '__main__': - import xml.etree.ElementTree as ET - RegressionTestResult.USE_XML = True - - class TestTests(unittest.TestCase): - def test_pass(self): - pass - - def test_pass_slow(self): - time.sleep(1.0) - - def test_fail(self): - print('stdout', file=sys.stdout) - print('stderr', file=sys.stderr) - self.fail('failure message') - - def test_error(self): - print('stdout', file=sys.stdout) - print('stderr', file=sys.stderr) - raise RuntimeError('error message') - - suite = unittest.TestSuite() - suite.addTest(unittest.TestLoader().loadTestsFromTestCase(TestTests)) - stream = io.StringIO() - runner_cls = get_test_runner_class(sum(a == '-v' for a in sys.argv)) - runner = runner_cls(sys.stdout) - result = runner.run(suite) - print('Output:', stream.getvalue()) - print('XML: ', end='') - for s in ET.tostringlist(result.get_xml_element()): - print(s.decode(), end='') - print() diff --git a/Lib/test/support/threading_helper.py b/Lib/test/support/threading_helper.py index afa25a76f..9b2b8f2df 100644 --- a/Lib/test/support/threading_helper.py +++ b/Lib/test/support/threading_helper.py @@ -8,7 +8,6 @@ import unittest from test import support - #======================================================================= # Threading support to prevent reporting refleaks when running regrtest.py -R @@ -248,3 +247,27 @@ def requires_working_threading(*, module=False): raise unittest.SkipTest(msg) else: return unittest.skipUnless(can_start_thread, msg) + + +def run_concurrently(worker_func, nthreads, args=(), kwargs={}): + """ + Run the worker function concurrently in multiple threads. + """ + barrier = threading.Barrier(nthreads) + + def wrapper_func(*args, **kwargs): + # Wait for all threads to reach this point before proceeding. + barrier.wait() + worker_func(*args, **kwargs) + + with catch_threading_exception() as cm: + workers = [ + threading.Thread(target=wrapper_func, args=args, kwargs=kwargs) + for _ in range(nthreads) + ] + with start_threads(workers): + pass + + # If a worker thread raises an exception, re-raise it. + if cm.exc_value is not None: + raise cm.exc_value diff --git a/Lib/test/support/venv.py b/Lib/test/support/venv.py index 78e6a51ec..b60f6097e 100644 --- a/Lib/test/support/venv.py +++ b/Lib/test/support/venv.py @@ -1,8 +1,8 @@ import contextlib import logging import os -import subprocess import shlex +import subprocess import sys import sysconfig import tempfile @@ -68,3 +68,14 @@ class VirtualEnvironment: raise else: return result + + +class VirtualEnvironmentMixin: + def venv(self, name=None, **venv_create_args): + venv_name = self.id() + if name: + venv_name += f'-{name}' + return VirtualEnvironment.from_tmpdir( + prefix=f'{venv_name}-venv-', + **venv_create_args, + ) diff --git a/Lib/test/support/warnings_helper.py b/Lib/test/support/warnings_helper.py index c1bf05623..5f6f14afd 100644 --- a/Lib/test/support/warnings_helper.py +++ b/Lib/test/support/warnings_helper.py @@ -23,8 +23,7 @@ def check_syntax_warning(testcase, statement, errtext='', testcase.assertEqual(len(warns), 1, warns) warn, = warns - testcase.assertTrue(issubclass(warn.category, SyntaxWarning), - warn.category) + testcase.assertIsSubclass(warn.category, SyntaxWarning) if errtext: testcase.assertRegex(str(warn.message), errtext) testcase.assertEqual(warn.filename, '') @@ -160,11 +159,12 @@ def _filterwarnings(filters, quiet=False): registry = frame.f_globals.get('__warningregistry__') if registry: registry.clear() - with warnings.catch_warnings(record=True) as w: - # Set filter "always" to record all warnings. Because - # test_warnings swap the module, we need to look up in - # the sys.modules dictionary. - sys.modules['warnings'].simplefilter("always") + # Because test_warnings swap the module, we need to look up in the + # sys.modules dictionary. + wmod = sys.modules['warnings'] + with wmod.catch_warnings(record=True) as w: + # Set filter "always" to record all warnings. + wmod.simplefilter("always") yield WarningsRecorder(w) # Filter the recorded warnings reraise = list(w) diff --git a/Lib/test/test_support.py b/Lib/test/test_support.py index d66afdc83..813a639d1 100644 --- a/Lib/test/test_support.py +++ b/Lib/test/test_support.py @@ -5,6 +5,7 @@ import io import logging import os import shutil +import signal import socket import stat import subprocess @@ -16,12 +17,13 @@ import unittest import warnings from test import support -from test.support import import_helper -from test.support import os_helper -from test.support import script_helper -from test.support import socket_helper -from test.support import warnings_helper -from test.support.testcase import ExtraAssertions +from test.support import ( + import_helper, + os_helper, + script_helper, + socket_helper, + warnings_helper, +) TESTFN = os_helper.TESTFN @@ -51,26 +53,26 @@ def _caplog(): root_logger.removeHandler(handler) -class TestSupport(unittest.TestCase, ExtraAssertions): +class TestSupport(unittest.TestCase): @classmethod def setUpClass(cls): - orig_filter_len = len(warnings.filters) + orig_filter_len = len(warnings._get_filters()) cls._warnings_helper_token = support.ignore_deprecations_from( "test.support.warnings_helper", like=".*used in test_support.*" ) cls._test_support_token = support.ignore_deprecations_from( __name__, like=".*You should NOT be seeing this.*" ) - assert len(warnings.filters) == orig_filter_len + 2 + assert len(warnings._get_filters()) == orig_filter_len + 2 @classmethod def tearDownClass(cls): - orig_filter_len = len(warnings.filters) + orig_filter_len = len(warnings._get_filters()) support.clear_ignored_deprecations( cls._warnings_helper_token, cls._test_support_token, ) - assert len(warnings.filters) == orig_filter_len - 2 + assert len(warnings._get_filters()) == orig_filter_len - 2 def test_ignored_deprecations_are_silent(self): """Test support.ignore_deprecations_from() silences warnings""" @@ -98,7 +100,7 @@ class TestSupport(unittest.TestCase, ExtraAssertions): self.assertEqual(support.get_original_stdout(), sys.stdout) def test_unload(self): - import sched + import sched # noqa: F401 self.assertIn("sched", sys.modules) import_helper.unload("sched") self.assertNotIn("sched", sys.modules) @@ -407,10 +409,10 @@ class TestSupport(unittest.TestCase, ExtraAssertions): with support.swap_attr(obj, "y", 5) as y: self.assertEqual(obj.y, 5) self.assertIsNone(y) - self.assertFalse(hasattr(obj, 'y')) + self.assertNotHasAttr(obj, 'y') with support.swap_attr(obj, "y", 5): del obj.y - self.assertFalse(hasattr(obj, 'y')) + self.assertNotHasAttr(obj, 'y') def test_swap_item(self): D = {"x":1} @@ -469,7 +471,6 @@ class TestSupport(unittest.TestCase, ExtraAssertions): extra = { 'TextTestResult', 'installHandler', - 'IsolatedAsyncioTestCase', } not_exported = {'load_tests', "TestProgram", "BaseTestSuite"} support.check__all__(self, @@ -562,6 +563,7 @@ class TestSupport(unittest.TestCase, ExtraAssertions): ['-Wignore', '-X', 'dev'], ['-X', 'faulthandler'], ['-X', 'importtime'], + ['-X', 'importtime=2'], ['-X', 'showrefcount'], ['-X', 'tracemalloc'], ['-X', 'tracemalloc=3'], @@ -586,7 +588,6 @@ class TestSupport(unittest.TestCase, ExtraAssertions): self.check_options(opts, 'optim_args_from_interpreter_flags') @unittest.skipIf(support.is_apple_mobile, "Unstable on Apple Mobile") - @unittest.skipIf(support.is_emscripten, "Unstable in Emscripten") @unittest.skipIf(support.is_wasi, "Unavailable on WASI") def test_fd_count(self): # We cannot test the absolute value of fd_count(): on old Linux kernel @@ -614,17 +615,13 @@ class TestSupport(unittest.TestCase, ExtraAssertions): self.check_print_warning("a\nb", 'Warning -- a\nWarning -- b\n') - # TODO: RUSTPYTHON - strftime extension not fully supported on non-Windows - @unittest.skipUnless(sys.platform == "win32" or support.is_emscripten, - "strftime extension not fully supported on non-Windows") def test_has_strftime_extensions(self): - if support.is_emscripten or sys.platform == "win32": + if sys.platform == "win32": self.assertFalse(support.has_strftime_extensions) else: self.assertTrue(support.has_strftime_extensions) - # TODO: RUSTPYTHON - _testinternalcapi module not available - @unittest.expectedFailure + @unittest.expectedFailure # TODO: RUSTPYTHON; - _testinternalcapi module not available def test_get_recursion_depth(self): # test support.get_recursion_depth() code = textwrap.dedent(""" @@ -668,8 +665,7 @@ class TestSupport(unittest.TestCase, ExtraAssertions): """) script_helper.assert_python_ok("-c", code) - # TODO: RUSTPYTHON - stack overflow in debug mode with deep recursion - @unittest.skip("TODO: RUSTPYTHON - causes segfault in debug builds") + @unittest.skip('TODO: RUSTPYTHON; - causes segfault in debug builds') def test_recursion(self): # Test infinite_recursion() and get_recursion_available() functions. def recursive_function(depth): @@ -778,9 +774,31 @@ class TestSupport(unittest.TestCase, ExtraAssertions): self.assertEqual(support.copy_python_src_ignore(path, os.listdir(path)), ignored) + def test_get_signal_name(self): + for exitcode, expected in ( + (-int(signal.SIGINT), 'SIGINT'), + (-int(signal.SIGSEGV), 'SIGSEGV'), + (128 + int(signal.SIGABRT), 'SIGABRT'), + (3221225477, "STATUS_ACCESS_VIOLATION"), + (0xC00000FD, "STATUS_STACK_OVERFLOW"), + ): + self.assertEqual(support.get_signal_name(exitcode), expected, + exitcode) + def test_linked_to_musl(self): linked = support.linked_to_musl() - self.assertIsInstance(linked, bool) + self.assertIsNotNone(linked) + if support.is_wasm32: + self.assertTrue(linked) + # The value is cached, so make sure it returns the same value again. + self.assertIs(linked, support.linked_to_musl()) + # The musl version is either triple or just a major version number. + if linked: + self.assertIsInstance(linked, tuple) + self.assertIn(len(linked), (1, 3)) + for v in linked: + self.assertIsInstance(v, int) + # XXX -follows a list of untested API # make_legacy_pyc