Update test.support from 3.14.2

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

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

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
 

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})

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)

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

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_<name> 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)

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

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)',

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='<import_helper>',
+        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

Lib/test/support/interpreters/__init__.py

@@ -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

Lib/test/support/interpreters/_crossinterp.py

@@ -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))

Lib/test/support/interpreters/queues.py

@@ -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)

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

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

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()

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

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.
 

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
 

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<syscall>[^(]*)\((?P<args>[^)]*)\)\s*[=]\s*(?P<returncode>.+)")
+_returncode_regex = re.compile(
+    br"\+\+\+ exited with (?P<returncode>\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 <X>" 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"]

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

Lib/test/support/testresult.py

@@ -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()

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

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,
+        )

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, '<testcase>')
@@ -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)

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