Merge pull request #6768 from youknowone/support

Update support to 3.14.2
This commit is contained in:
Jeong, YunWon authored and GitHub committed 2026-01-18 20:36:04 +09:00
commit e0479fe197
67 files changed
+2239 -1512

No files matched your search

+1
View File
@@ -4417,6 +4417,7 @@ class _TestSharedMemory(BaseTestCase):
sms.close()
@unittest.skip("TODO: RUSTPYTHON; flaky")
@unittest.skipIf(os.name != "posix", "not feasible in non-posix platforms")
def test_shared_memory_SharedMemoryServer_ignores_sigint(self):
# bpo-36368: protect SharedMemoryManager server process from
+491 -290
View File
File diff suppressed because it is too large. Load diff
+1 -1
View File
@@ -1,6 +1,6 @@
from enum import Enum
import functools
import unittest
from enum import Enum
__all__ = [
"given",
+4
View File
@@ -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
+1 -1
View File
@@ -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
+17 -7
View File
@@ -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})
+3 -2
View File
@@ -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)
@@ -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):
+307 -28
View File
@@ -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)
+2 -1
View File
@@ -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"))
+1 -1
View File
@@ -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)',
+133 -2
View File
@@ -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
@@ -309,3 +311,132 @@ def ready_to_import(name=None, source=""):
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
-258
View File
@@ -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
-102
View File
@@ -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))
-313
View File
@@ -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)
+1
View File
@@ -1,5 +1,6 @@
import logging.handlers
class TestHandler(logging.handlers.BufferingHandler):
def __init__(self, matcher):
# BufferingHandler takes a "capacity" argument
+27 -1
View File
@@ -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
+1
View File
@@ -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()
+22 -21
View File
@@ -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
Vendored Regular → Executable
+16 -14
View File
@@ -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.
+1 -1
View File
@@ -2,8 +2,8 @@ import contextlib
import errno
import os.path
import socket
import sys
import subprocess
import sys
import tempfile
import unittest
+210
View File
@@ -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"]
+1
View File
@@ -1,6 +1,7 @@
from math import copysign, isnan
# XXX: RUSTPYTHON: removed in 3.14
class ExtraAssertions:
def assertIsSubclass(self, cls, superclass, msg=None):
-191
View File
@@ -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()
+24 -1
View File
@@ -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
+12 -1
View File
@@ -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,
)
+7 -7
View File
@@ -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)
-1
View File
@@ -24,7 +24,6 @@ def supports_virtual_terminal():
class TestTheme(unittest.TestCase):
@unittest.expectedFailure # TODO: RUSTPYTHON
def test_attributes(self):
# only theme configurations attributes by default
for field in dataclasses.fields(_colorize.Theme):
-1
View File
@@ -6824,7 +6824,6 @@ class TestInvalidNargs(TestCase):
class TestImportStar(TestCase):
@unittest.expectedFailure # TODO: RUSTPYTHON
def test(self):
for name in argparse.__all__:
self.assertHasAttr(argparse, name)
-2
View File
@@ -3023,7 +3023,6 @@ class NodeTransformerTests(ASTTestMixin, BaseNodeVisitorCases, unittest.TestCase
self.assertASTTransformation(YieldRemover, code, expected)
@unittest.expectedFailure # TODO: RUSTPYTHON; <class 'object'> is not <class 'NoneType'>
def test_node_return_list(self):
code = """
class DSL(Base, kw1=True): ...
@@ -3064,7 +3063,6 @@ class NodeTransformerTests(ASTTestMixin, BaseNodeVisitorCases, unittest.TestCase
self.assertASTTransformation(PrintToLog, code, expected)
@unittest.expectedFailure # TODO: RUSTPYTHON; <class 'object'> is not <class 'NoneType'>
def test_node_replace(self):
code = """
def func(arg):
-1
View File
@@ -1090,7 +1090,6 @@ class CommandLineTestCase(unittest.TestCase):
output = run('--months', '1', '2004')
self.assertIn(conv('\nMo Tu We Th Fr Sa Su\n'), output)
@unittest.expectedFailure # TODO: RUSTPYTHON
def test_option_type(self):
self.assertFailure('-t')
self.assertFailure('--type')
+6 -6
View File
@@ -681,7 +681,7 @@ class IntStrDigitLimitsTests(unittest.TestCase):
digits = 78_268
with (
support.adjust_int_max_str_digits(digits),
support.CPUStopwatch() as sw_convert):
support.Stopwatch() as sw_convert):
huge_decimal = str(huge_int)
self.assertEqual(len(huge_decimal), digits)
# Ensuring that we chose a slow enough conversion to measure.
@@ -696,7 +696,7 @@ class IntStrDigitLimitsTests(unittest.TestCase):
with support.adjust_int_max_str_digits(int(.995 * digits)):
with (
self.assertRaises(ValueError) as err,
support.CPUStopwatch() as sw_fail_huge):
support.Stopwatch() as sw_fail_huge):
str(huge_int)
self.assertIn('conversion', str(err.exception))
self.assertLessEqual(sw_fail_huge.seconds, sw_convert.seconds/2)
@@ -706,7 +706,7 @@ class IntStrDigitLimitsTests(unittest.TestCase):
extra_huge_int = int(f'0x{"c"*500_000}', base=16) # 602060 digits.
with (
self.assertRaises(ValueError) as err,
support.CPUStopwatch() as sw_fail_extra_huge):
support.Stopwatch() as sw_fail_extra_huge):
# If not limited, 8 seconds said Zen based cloud VM.
str(extra_huge_int)
self.assertIn('conversion', str(err.exception))
@@ -722,7 +722,7 @@ class IntStrDigitLimitsTests(unittest.TestCase):
huge = '8'*digits
with (
support.adjust_int_max_str_digits(digits),
support.CPUStopwatch() as sw_convert):
support.Stopwatch() as sw_convert):
int(huge)
# Ensuring that we chose a slow enough conversion to measure.
# It takes 0.1 seconds on a Zen based cloud VM in an opt build.
@@ -734,7 +734,7 @@ class IntStrDigitLimitsTests(unittest.TestCase):
with support.adjust_int_max_str_digits(digits - 1):
with (
self.assertRaises(ValueError) as err,
support.CPUStopwatch() as sw_fail_huge):
support.Stopwatch() as sw_fail_huge):
int(huge)
self.assertIn('conversion', str(err.exception))
self.assertLessEqual(sw_fail_huge.seconds, sw_convert.seconds/2)
@@ -744,7 +744,7 @@ class IntStrDigitLimitsTests(unittest.TestCase):
extra_huge = '7'*1_200_000
with (
self.assertRaises(ValueError) as err,
support.CPUStopwatch() as sw_fail_extra_huge):
support.Stopwatch() as sw_fail_extra_huge):
# If not limited, 8 seconds in the Zen based cloud VM.
int(extra_huge)
self.assertIn('conversion', str(err.exception))
-2
View File
@@ -99,8 +99,6 @@ class TestFail:
'keys must be str, int, float, bool or None, not tuple'):
self.dumps(data)
# TODO: RUSTPYTHON
@unittest.expectedFailure
def test_not_serializable(self):
import sys
with self.assertRaisesRegex(TypeError,
-1
View File
@@ -165,7 +165,6 @@ class TestMain(unittest.TestCase):
self.assertEqual(process.stdout, self.jsonlines_expect)
self.assertEqual(process.stderr, '')
@unittest.expectedFailure # TODO: RUSTPYTHON
def test_help_flag(self):
rc, out, err = assert_python_ok('-m', self.module, '-h',
PYTHON_COLORS='0')
+2 -2
View File
@@ -1,7 +1,7 @@
from test.support import (gc_collect, bigmemtest, _2G,
cpython_only, captured_stdout,
check_disallow_instantiation, is_emscripten, is_wasi,
warnings_helper, SHORT_TIMEOUT, CPUStopwatch, requires_resource)
warnings_helper, SHORT_TIMEOUT, Stopwatch, requires_resource)
import locale
import re
import string
@@ -2487,7 +2487,7 @@ class ReTests(unittest.TestCase):
@requires_resource('cpu')
def test_search_anchor_at_beginning(self):
s = 'x'*10**7
with CPUStopwatch() as stopwatch:
with Stopwatch() as stopwatch:
for p in r'\Ay', r'^y':
self.assertIsNone(re.search(p, s))
self.assertEqual(re.split(p, s), [s])
+2 -2
View File
@@ -24,7 +24,7 @@ import unittest
from xml.etree import ElementTree
from test import support
from test.support import os_helper, without_optimizer
from test.support import os_helper, requires_jit_disabled
from test.libregrtest import cmdline
from test.libregrtest import main
from test.libregrtest import setup
@@ -1245,7 +1245,7 @@ class ArgsTestCase(BaseTestCase):
stats=TestStats(4, 1),
forever=True)
@without_optimizer
@requires_jit_disabled
def check_leak(self, code, what, *, run_workers=False):
test = self.create_test('huntrleaks', code=code)
+45 -25
View File
@@ -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}
@@ -458,6 +460,7 @@ class TestSupport(unittest.TestCase, ExtraAssertions):
self.OtherClass, self.RefClass, ignore=ignore)
self.assertEqual(set(), missing_items)
@unittest.expectedFailure # TODO: RUSTPYTHON
def test_check__all__(self):
extra = {'tempdir'}
not_exported = {'template'}
@@ -469,7 +472,6 @@ class TestSupport(unittest.TestCase, ExtraAssertions):
extra = {
'TextTestResult',
'installHandler',
'IsolatedAsyncioTestCase',
}
not_exported = {'load_tests', "TestProgram", "BaseTestSuite"}
support.check__all__(self,
@@ -562,6 +564,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 +589,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 +616,14 @@ 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")
@unittest.expectedFailureIf(sys.platform != "win32", "TODO: RUSTPYTHON; no has_strftime_extensions yet")
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 +667,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 +776,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
+1
View File
@@ -1,4 +1,5 @@
import os.path
from test.support import load_package_tests
+2 -1
View File
@@ -1,4 +1,5 @@
from . import load_tests
import unittest
from . import load_tests
unittest.main()
+1
View File
@@ -14,6 +14,7 @@ import sys
import unittest
import warnings
def warnfun():
warnings.warn('rw', RuntimeWarning)
+3 -2
View File
@@ -1,10 +1,11 @@
import datetime
import unittest
import warnings
import weakref
import unittest
from test.support import gc_collect
from itertools import product
from test.support import gc_collect
class Test_Assertions(unittest.TestCase):
def test_AlmostEqual(self):
+13 -3
View File
@@ -1,7 +1,9 @@
import asyncio
import contextvars
import unittest
from test import support
from test.support import force_not_colorized
support.requires_working_socket(module=True)
@@ -11,7 +13,9 @@ class MyException(Exception):
def tearDownModule():
asyncio.set_event_loop_policy(None)
# XXX: RUSTPYTHON; asyncio.events._set_event_loop_policy is not implemented
# asyncio.events._set_event_loop_policy(None)
pass
class TestCM:
@@ -253,6 +257,7 @@ class TestAsyncCase(unittest.TestCase):
test.doCleanups()
self.assertEqual(events, ['asyncSetUp', 'test', 'asyncTearDown', 'cleanup'])
@force_not_colorized
def test_exception_in_tear_clean_up(self):
class Test(unittest.IsolatedAsyncioTestCase):
async def asyncSetUp(self):
@@ -314,18 +319,21 @@ class TestAsyncCase(unittest.TestCase):
self.assertIn('It is deprecated to return a value that is not None', str(w.warning))
self.assertIn('test1', str(w.warning))
self.assertEqual(w.filename, __file__)
self.assertIn("returned 'int'", str(w.warning))
with self.assertWarns(DeprecationWarning) as w:
Test('test2').run()
self.assertIn('It is deprecated to return a value that is not None', str(w.warning))
self.assertIn('test2', str(w.warning))
self.assertEqual(w.filename, __file__)
self.assertIn("returned 'async_generator'", str(w.warning))
with self.assertWarns(DeprecationWarning) as w:
Test('test3').run()
self.assertIn('It is deprecated to return a value that is not None', str(w.warning))
self.assertIn('test3', str(w.warning))
self.assertEqual(w.filename, __file__)
self.assertIn(f'returned {Nothing.__name__!r}', str(w.warning))
def test_cleanups_interleave_order(self):
events = []
@@ -470,6 +478,7 @@ class TestAsyncCase(unittest.TestCase):
test.doCleanups()
self.assertEqual(events, ['asyncSetUp', 'test', 'cleanup'])
@unittest.expectedFailure
def test_setup_get_event_loop(self):
# See https://github.com/python/cpython/issues/95736
# Make sure the default event loop is not used
@@ -477,7 +486,7 @@ class TestAsyncCase(unittest.TestCase):
class TestCase1(unittest.IsolatedAsyncioTestCase):
def setUp(self):
asyncio.get_event_loop_policy().get_event_loop()
asyncio.events._get_event_loop_policy().get_event_loop()
async def test_demo1(self):
pass
@@ -486,8 +495,9 @@ class TestAsyncCase(unittest.TestCase):
result = test.run()
self.assertTrue(result.wasSuccessful())
@unittest.expectedFailure # TODO: RUSTPYTHON; asyncio.events._set_event_loop_policy is not implemented
def test_loop_factory(self):
asyncio.set_event_loop_policy(None)
asyncio.events._set_event_loop_policy(None)
class TestCase1(unittest.IsolatedAsyncioTestCase):
loop_factory = asyncio.EventLoop
+2 -2
View File
@@ -1,10 +1,10 @@
import gc
import io
import os
import sys
import signal
import weakref
import sys
import unittest
import weakref
from test import support
+349 -22
View File
@@ -1,26 +1,27 @@
import contextlib
import difflib
import pprint
import inspect
import logging
import pickle
import pprint
import re
import sys
import logging
import types
import unittest
import warnings
import weakref
import inspect
import types
from collections import UserString
from copy import deepcopy
from test import support
import unittest
from test.test_unittest.support import (
TestEquality, TestHashing, LoggingResult, LegacyLoggingResult,
ResultWithNoStartTestRunStopTestRun
)
from test.support import captured_stderr, gc_collect
from test.test_unittest.support import (
LegacyLoggingResult,
LoggingResult,
ResultWithNoStartTestRunStopTestRun,
TestEquality,
TestHashing,
)
log_foo = logging.getLogger('foo')
log_foobar = logging.getLogger('foo.bar')
@@ -54,6 +55,10 @@ class Test(object):
self.events.append('tearDown')
class List(list):
pass
class Test_TestCase(unittest.TestCase, TestEquality, TestHashing):
### Set up attributes used by inherited tests
@@ -85,7 +90,7 @@ class Test_TestCase(unittest.TestCase, TestEquality, TestHashing):
def runTest(self): raise MyException()
def test(self): pass
self.assertEqual(Test().id()[-13:], '.Test.runTest')
self.assertEndsWith(Test().id(), '.Test.runTest')
# test that TestCase can be instantiated with no args
# primarily for use at the interactive interpreter
@@ -106,7 +111,7 @@ class Test_TestCase(unittest.TestCase, TestEquality, TestHashing):
def runTest(self): raise MyException()
def test(self): pass
self.assertEqual(Test('test').id()[-10:], '.Test.test')
self.assertEndsWith(Test('test').id(), '.Test.test')
# "class TestCase([methodName])"
# ...
@@ -325,18 +330,40 @@ class Test_TestCase(unittest.TestCase, TestEquality, TestHashing):
self.assertIn('It is deprecated to return a value that is not None', str(w.warning))
self.assertIn('test1', str(w.warning))
self.assertEqual(w.filename, __file__)
self.assertIn("returned 'int'", str(w.warning))
with self.assertWarns(DeprecationWarning) as w:
Foo('test2').run()
self.assertIn('It is deprecated to return a value that is not None', str(w.warning))
self.assertIn('test2', str(w.warning))
self.assertEqual(w.filename, __file__)
self.assertIn("returned 'generator'", str(w.warning))
with self.assertWarns(DeprecationWarning) as w:
Foo('test3').run()
self.assertIn('It is deprecated to return a value that is not None', str(w.warning))
self.assertIn('test3', str(w.warning))
self.assertEqual(w.filename, __file__)
self.assertIn(f'returned {Nothing.__name__!r}', str(w.warning))
def test_deprecation_of_return_val_from_test_async_method(self):
class Foo(unittest.TestCase):
async def test1(self):
return 1
with self.assertWarns(DeprecationWarning) as w:
warnings.filterwarnings('ignore',
'coroutine .* was never awaited', RuntimeWarning)
Foo('test1').run()
support.gc_collect()
self.assertIn('It is deprecated to return a value that is not None', str(w.warning))
self.assertIn('test1', str(w.warning))
self.assertEqual(w.filename, __file__)
self.assertIn("returned 'coroutine'", str(w.warning))
self.assertIn(
'Maybe you forgot to use IsolatedAsyncioTestCase as the base class?',
str(w.warning),
)
def _check_call_order__subtests(self, result, events, expected_events):
class Foo(Test.LoggingTestCase):
@@ -678,16 +705,136 @@ class Test_TestCase(unittest.TestCase, TestEquality, TestHashing):
self.assertRaises(self.failureException, self.assertIsNot, thing, thing)
def testAssertIsInstance(self):
thing = []
thing = List()
self.assertIsInstance(thing, list)
self.assertRaises(self.failureException, self.assertIsInstance,
thing, dict)
self.assertIsInstance(thing, (int, list))
with self.assertRaises(self.failureException) as cm:
self.assertIsInstance(thing, int)
self.assertEqual(str(cm.exception),
"[] is not an instance of <class 'int'>")
with self.assertRaises(self.failureException) as cm:
self.assertIsInstance(thing, (int, float))
self.assertEqual(str(cm.exception),
"[] is not an instance of any of (<class 'int'>, <class 'float'>)")
with self.assertRaises(self.failureException) as cm:
self.assertIsInstance(thing, int, 'ababahalamaha')
self.assertIn('ababahalamaha', str(cm.exception))
with self.assertRaises(self.failureException) as cm:
self.assertIsInstance(thing, int, msg='ababahalamaha')
self.assertIn('ababahalamaha', str(cm.exception))
def testAssertNotIsInstance(self):
thing = []
self.assertNotIsInstance(thing, dict)
self.assertRaises(self.failureException, self.assertNotIsInstance,
thing, list)
thing = List()
self.assertNotIsInstance(thing, int)
self.assertNotIsInstance(thing, (int, float))
with self.assertRaises(self.failureException) as cm:
self.assertNotIsInstance(thing, list)
self.assertEqual(str(cm.exception),
"[] is an instance of <class 'list'>")
with self.assertRaises(self.failureException) as cm:
self.assertNotIsInstance(thing, (int, list))
self.assertEqual(str(cm.exception),
"[] is an instance of <class 'list'>")
with self.assertRaises(self.failureException) as cm:
self.assertNotIsInstance(thing, list, 'ababahalamaha')
self.assertIn('ababahalamaha', str(cm.exception))
with self.assertRaises(self.failureException) as cm:
self.assertNotIsInstance(thing, list, msg='ababahalamaha')
self.assertIn('ababahalamaha', str(cm.exception))
def testAssertIsSubclass(self):
self.assertIsSubclass(List, list)
self.assertIsSubclass(List, (int, list))
with self.assertRaises(self.failureException) as cm:
self.assertIsSubclass(List, int)
self.assertEqual(str(cm.exception),
f"{List!r} is not a subclass of <class 'int'>")
with self.assertRaises(self.failureException) as cm:
self.assertIsSubclass(List, (int, float))
self.assertEqual(str(cm.exception),
f"{List!r} is not a subclass of any of (<class 'int'>, <class 'float'>)")
with self.assertRaises(self.failureException) as cm:
self.assertIsSubclass(1, int)
self.assertEqual(str(cm.exception), "1 is not a class")
with self.assertRaises(self.failureException) as cm:
self.assertIsSubclass(List, int, 'ababahalamaha')
self.assertIn('ababahalamaha', str(cm.exception))
with self.assertRaises(self.failureException) as cm:
self.assertIsSubclass(List, int, msg='ababahalamaha')
self.assertIn('ababahalamaha', str(cm.exception))
def testAssertNotIsSubclass(self):
self.assertNotIsSubclass(List, int)
self.assertNotIsSubclass(List, (int, float))
with self.assertRaises(self.failureException) as cm:
self.assertNotIsSubclass(List, list)
self.assertEqual(str(cm.exception),
f"{List!r} is a subclass of <class 'list'>")
with self.assertRaises(self.failureException) as cm:
self.assertNotIsSubclass(List, (int, list))
self.assertEqual(str(cm.exception),
f"{List!r} is a subclass of <class 'list'>")
with self.assertRaises(self.failureException) as cm:
self.assertNotIsSubclass(1, int)
self.assertEqual(str(cm.exception), "1 is not a class")
with self.assertRaises(self.failureException) as cm:
self.assertNotIsSubclass(List, list, 'ababahalamaha')
self.assertIn('ababahalamaha', str(cm.exception))
with self.assertRaises(self.failureException) as cm:
self.assertNotIsSubclass(List, list, msg='ababahalamaha')
self.assertIn('ababahalamaha', str(cm.exception))
def testAssertHasAttr(self):
a = List()
a.x = 1
self.assertHasAttr(a, 'x')
with self.assertRaises(self.failureException) as cm:
self.assertHasAttr(a, 'y')
self.assertEqual(str(cm.exception),
"'List' object has no attribute 'y'")
with self.assertRaises(self.failureException) as cm:
self.assertHasAttr(List, 'spam')
self.assertEqual(str(cm.exception),
"type object 'List' has no attribute 'spam'")
with self.assertRaises(self.failureException) as cm:
self.assertHasAttr(sys, 'nonexistent')
self.assertEqual(str(cm.exception),
"module 'sys' has no attribute 'nonexistent'")
with self.assertRaises(self.failureException) as cm:
self.assertHasAttr(a, 'y', 'ababahalamaha')
self.assertIn('ababahalamaha', str(cm.exception))
with self.assertRaises(self.failureException) as cm:
self.assertHasAttr(a, 'y', msg='ababahalamaha')
self.assertIn('ababahalamaha', str(cm.exception))
def testAssertNotHasAttr(self):
a = List()
a.x = 1
self.assertNotHasAttr(a, 'y')
with self.assertRaises(self.failureException) as cm:
self.assertNotHasAttr(a, 'x')
self.assertEqual(str(cm.exception),
"'List' object has unexpected attribute 'x'")
with self.assertRaises(self.failureException) as cm:
self.assertNotHasAttr(List, 'append')
self.assertEqual(str(cm.exception),
"type object 'List' has unexpected attribute 'append'")
with self.assertRaises(self.failureException) as cm:
self.assertNotHasAttr(sys, 'modules')
self.assertEqual(str(cm.exception),
"module 'sys' has unexpected attribute 'modules'")
with self.assertRaises(self.failureException) as cm:
self.assertNotHasAttr(a, 'x', 'ababahalamaha')
self.assertIn('ababahalamaha', str(cm.exception))
with self.assertRaises(self.failureException) as cm:
self.assertNotHasAttr(a, 'x', msg='ababahalamaha')
self.assertIn('ababahalamaha', str(cm.exception))
def testAssertIn(self):
animals = {'monkey': 'banana', 'cow': 'grass', 'seal': 'fish'}
@@ -1842,6 +1989,186 @@ test case
pass
self.assertIsNone(value)
def testAssertStartsWith(self):
self.assertStartsWith('ababahalamaha', 'ababa')
self.assertStartsWith('ababahalamaha', ('x', 'ababa', 'y'))
self.assertStartsWith(UserString('ababahalamaha'), 'ababa')
self.assertStartsWith(UserString('ababahalamaha'), ('x', 'ababa', 'y'))
self.assertStartsWith(bytearray(b'ababahalamaha'), b'ababa')
self.assertStartsWith(bytearray(b'ababahalamaha'), (b'x', b'ababa', b'y'))
self.assertStartsWith(b'ababahalamaha', bytearray(b'ababa'))
self.assertStartsWith(b'ababahalamaha',
(bytearray(b'x'), bytearray(b'ababa'), bytearray(b'y')))
with self.assertRaises(self.failureException) as cm:
self.assertStartsWith('ababahalamaha', 'amaha')
self.assertEqual(str(cm.exception),
"'ababahalamaha' doesn't start with 'amaha'")
with self.assertRaises(self.failureException) as cm:
self.assertStartsWith('ababahalamaha', ('x', 'y'))
self.assertEqual(str(cm.exception),
"'ababahalamaha' doesn't start with any of ('x', 'y')")
with self.assertRaises(self.failureException) as cm:
self.assertStartsWith(b'ababahalamaha', 'ababa')
self.assertEqual(str(cm.exception), 'Expected str, not bytes')
with self.assertRaises(self.failureException) as cm:
self.assertStartsWith(b'ababahalamaha', ('amaha', 'ababa'))
self.assertEqual(str(cm.exception), 'Expected str, not bytes')
with self.assertRaises(self.failureException) as cm:
self.assertStartsWith([], 'ababa')
self.assertEqual(str(cm.exception), 'Expected str, not list')
with self.assertRaises(self.failureException) as cm:
self.assertStartsWith('ababahalamaha', b'ababa')
self.assertEqual(str(cm.exception), 'Expected bytes, not str')
with self.assertRaises(self.failureException) as cm:
self.assertStartsWith('ababahalamaha', (b'amaha', b'ababa'))
self.assertEqual(str(cm.exception), 'Expected bytes, not str')
with self.assertRaises(TypeError):
self.assertStartsWith('ababahalamaha', ord('a'))
with self.assertRaises(self.failureException) as cm:
self.assertStartsWith('ababahalamaha', 'amaha', 'abracadabra')
self.assertIn('ababahalamaha', str(cm.exception))
with self.assertRaises(self.failureException) as cm:
self.assertStartsWith('ababahalamaha', 'amaha', msg='abracadabra')
self.assertIn('ababahalamaha', str(cm.exception))
def testAssertNotStartsWith(self):
self.assertNotStartsWith('ababahalamaha', 'amaha')
self.assertNotStartsWith('ababahalamaha', ('x', 'amaha', 'y'))
self.assertNotStartsWith(UserString('ababahalamaha'), 'amaha')
self.assertNotStartsWith(UserString('ababahalamaha'), ('x', 'amaha', 'y'))
self.assertNotStartsWith(bytearray(b'ababahalamaha'), b'amaha')
self.assertNotStartsWith(bytearray(b'ababahalamaha'), (b'x', b'amaha', b'y'))
self.assertNotStartsWith(b'ababahalamaha', bytearray(b'amaha'))
self.assertNotStartsWith(b'ababahalamaha',
(bytearray(b'x'), bytearray(b'amaha'), bytearray(b'y')))
with self.assertRaises(self.failureException) as cm:
self.assertNotStartsWith('ababahalamaha', 'ababa')
self.assertEqual(str(cm.exception),
"'ababahalamaha' starts with 'ababa'")
with self.assertRaises(self.failureException) as cm:
self.assertNotStartsWith('ababahalamaha', ('x', 'ababa', 'y'))
self.assertEqual(str(cm.exception),
"'ababahalamaha' starts with 'ababa'")
with self.assertRaises(self.failureException) as cm:
self.assertNotStartsWith(b'ababahalamaha', 'ababa')
self.assertEqual(str(cm.exception), 'Expected str, not bytes')
with self.assertRaises(self.failureException) as cm:
self.assertNotStartsWith(b'ababahalamaha', ('amaha', 'ababa'))
self.assertEqual(str(cm.exception), 'Expected str, not bytes')
with self.assertRaises(self.failureException) as cm:
self.assertNotStartsWith([], 'ababa')
self.assertEqual(str(cm.exception), 'Expected str, not list')
with self.assertRaises(self.failureException) as cm:
self.assertNotStartsWith('ababahalamaha', b'ababa')
self.assertEqual(str(cm.exception), 'Expected bytes, not str')
with self.assertRaises(self.failureException) as cm:
self.assertNotStartsWith('ababahalamaha', (b'amaha', b'ababa'))
self.assertEqual(str(cm.exception), 'Expected bytes, not str')
with self.assertRaises(TypeError):
self.assertNotStartsWith('ababahalamaha', ord('a'))
with self.assertRaises(self.failureException) as cm:
self.assertNotStartsWith('ababahalamaha', 'ababa', 'abracadabra')
self.assertIn('ababahalamaha', str(cm.exception))
with self.assertRaises(self.failureException) as cm:
self.assertNotStartsWith('ababahalamaha', 'ababa', msg='abracadabra')
self.assertIn('ababahalamaha', str(cm.exception))
def testAssertEndsWith(self):
self.assertEndsWith('ababahalamaha', 'amaha')
self.assertEndsWith('ababahalamaha', ('x', 'amaha', 'y'))
self.assertEndsWith(UserString('ababahalamaha'), 'amaha')
self.assertEndsWith(UserString('ababahalamaha'), ('x', 'amaha', 'y'))
self.assertEndsWith(bytearray(b'ababahalamaha'), b'amaha')
self.assertEndsWith(bytearray(b'ababahalamaha'), (b'x', b'amaha', b'y'))
self.assertEndsWith(b'ababahalamaha', bytearray(b'amaha'))
self.assertEndsWith(b'ababahalamaha',
(bytearray(b'x'), bytearray(b'amaha'), bytearray(b'y')))
with self.assertRaises(self.failureException) as cm:
self.assertEndsWith('ababahalamaha', 'ababa')
self.assertEqual(str(cm.exception),
"'ababahalamaha' doesn't end with 'ababa'")
with self.assertRaises(self.failureException) as cm:
self.assertEndsWith('ababahalamaha', ('x', 'y'))
self.assertEqual(str(cm.exception),
"'ababahalamaha' doesn't end with any of ('x', 'y')")
with self.assertRaises(self.failureException) as cm:
self.assertEndsWith(b'ababahalamaha', 'amaha')
self.assertEqual(str(cm.exception), 'Expected str, not bytes')
with self.assertRaises(self.failureException) as cm:
self.assertEndsWith(b'ababahalamaha', ('ababa', 'amaha'))
self.assertEqual(str(cm.exception), 'Expected str, not bytes')
with self.assertRaises(self.failureException) as cm:
self.assertEndsWith([], 'amaha')
self.assertEqual(str(cm.exception), 'Expected str, not list')
with self.assertRaises(self.failureException) as cm:
self.assertEndsWith('ababahalamaha', b'amaha')
self.assertEqual(str(cm.exception), 'Expected bytes, not str')
with self.assertRaises(self.failureException) as cm:
self.assertEndsWith('ababahalamaha', (b'ababa', b'amaha'))
self.assertEqual(str(cm.exception), 'Expected bytes, not str')
with self.assertRaises(TypeError):
self.assertEndsWith('ababahalamaha', ord('a'))
with self.assertRaises(self.failureException) as cm:
self.assertEndsWith('ababahalamaha', 'ababa', 'abracadabra')
self.assertIn('ababahalamaha', str(cm.exception))
with self.assertRaises(self.failureException) as cm:
self.assertEndsWith('ababahalamaha', 'ababa', msg='abracadabra')
self.assertIn('ababahalamaha', str(cm.exception))
def testAssertNotEndsWith(self):
self.assertNotEndsWith('ababahalamaha', 'ababa')
self.assertNotEndsWith('ababahalamaha', ('x', 'ababa', 'y'))
self.assertNotEndsWith(UserString('ababahalamaha'), 'ababa')
self.assertNotEndsWith(UserString('ababahalamaha'), ('x', 'ababa', 'y'))
self.assertNotEndsWith(bytearray(b'ababahalamaha'), b'ababa')
self.assertNotEndsWith(bytearray(b'ababahalamaha'), (b'x', b'ababa', b'y'))
self.assertNotEndsWith(b'ababahalamaha', bytearray(b'ababa'))
self.assertNotEndsWith(b'ababahalamaha',
(bytearray(b'x'), bytearray(b'ababa'), bytearray(b'y')))
with self.assertRaises(self.failureException) as cm:
self.assertNotEndsWith('ababahalamaha', 'amaha')
self.assertEqual(str(cm.exception),
"'ababahalamaha' ends with 'amaha'")
with self.assertRaises(self.failureException) as cm:
self.assertNotEndsWith('ababahalamaha', ('x', 'amaha', 'y'))
self.assertEqual(str(cm.exception),
"'ababahalamaha' ends with 'amaha'")
with self.assertRaises(self.failureException) as cm:
self.assertNotEndsWith(b'ababahalamaha', 'amaha')
self.assertEqual(str(cm.exception), 'Expected str, not bytes')
with self.assertRaises(self.failureException) as cm:
self.assertNotEndsWith(b'ababahalamaha', ('ababa', 'amaha'))
self.assertEqual(str(cm.exception), 'Expected str, not bytes')
with self.assertRaises(self.failureException) as cm:
self.assertNotEndsWith([], 'amaha')
self.assertEqual(str(cm.exception), 'Expected str, not list')
with self.assertRaises(self.failureException) as cm:
self.assertNotEndsWith('ababahalamaha', b'amaha')
self.assertEqual(str(cm.exception), 'Expected bytes, not str')
with self.assertRaises(self.failureException) as cm:
self.assertNotEndsWith('ababahalamaha', (b'ababa', b'amaha'))
self.assertEqual(str(cm.exception), 'Expected bytes, not str')
with self.assertRaises(TypeError):
self.assertNotEndsWith('ababahalamaha', ord('a'))
with self.assertRaises(self.failureException) as cm:
self.assertNotEndsWith('ababahalamaha', 'amaha', 'abracadabra')
self.assertIn('ababahalamaha', str(cm.exception))
with self.assertRaises(self.failureException) as cm:
self.assertNotEndsWith('ababahalamaha', 'amaha', msg='abracadabra')
self.assertIn('ababahalamaha', str(cm.exception))
def testDeprecatedFailMethods(self):
"""Test that the deprecated fail* methods get removed in 3.12"""
deprecated_names = [
+58 -7
View File
@@ -1,15 +1,17 @@
import os.path
from os.path import abspath
import pickle
import re
import sys
import types
import pickle
from test import support
from test.support import import_helper
import unittest
import unittest.mock
from importlib._bootstrap_external import NamespaceLoader
from os.path import abspath
import test.test_unittest
from test import support
from test.support import import_helper
from test.test_importlib import util as test_util
class TestableTestProgram(unittest.TestProgram):
@@ -395,7 +397,7 @@ class TestDiscovery(unittest.TestCase):
self.addCleanup(restore_isdir)
_find_tests_args = []
def _find_tests(start_dir, pattern):
def _find_tests(start_dir, pattern, namespace=None):
_find_tests_args.append((start_dir, pattern))
return ['tests']
loader._find_tests = _find_tests
@@ -815,7 +817,7 @@ class TestDiscovery(unittest.TestCase):
expectedPath = os.path.abspath(os.path.dirname(test.test_unittest.__file__))
self.wasRun = False
def _find_tests(start_dir, pattern):
def _find_tests(start_dir, pattern, namespace=None):
self.wasRun = True
self.assertEqual(start_dir, expectedPath)
return tests
@@ -848,6 +850,55 @@ class TestDiscovery(unittest.TestCase):
'Can not use builtin modules '
'as dotted module names')
def test_discovery_from_dotted_namespace_packages(self):
loader = unittest.TestLoader()
package = types.ModuleType('package')
package.__name__ = "tests"
package.__path__ = ['/a', '/b']
package.__file__ = None
package.__spec__ = types.SimpleNamespace(
name=package.__name__,
loader=NamespaceLoader(package.__name__, package.__path__, None),
submodule_search_locations=['/a', '/b']
)
def _import(packagename, *args, **kwargs):
sys.modules[packagename] = package
return package
_find_tests_args = []
def _find_tests(start_dir, pattern, namespace=None):
_find_tests_args.append((start_dir, pattern))
return ['%s/tests' % start_dir]
loader._find_tests = _find_tests
loader.suiteClass = list
with unittest.mock.patch('builtins.__import__', _import):
# Since loader.discover() can modify sys.path, restore it when done.
with import_helper.DirsOnSysPath():
# Make sure to remove 'package' from sys.modules when done.
with test_util.uncache('package'):
suite = loader.discover('package')
self.assertEqual(suite, ['/a/tests', '/b/tests'])
@unittest.expectedFailure # TODO: RUSTPYTHON
def test_discovery_start_dir_is_namespace(self):
"""Subdirectory discovery not affected if start_dir is a namespace pkg."""
loader = unittest.TestLoader()
with (
import_helper.DirsOnSysPath(os.path.join(os.path.dirname(__file__))),
test_util.uncache('namespace_test_pkg')
):
suite = loader.discover('namespace_test_pkg')
self.assertEqual(
{list(suite)[0]._tests[0].__module__ for suite in suite._tests if list(suite)},
# files under namespace_test_pkg.noop not discovered.
{'namespace_test_pkg.test_foo', 'namespace_test_pkg.bar.test_bar'},
)
def test_discovery_failed_discovery(self):
from test.test_importlib import util
+2 -2
View File
@@ -1,9 +1,9 @@
import functools
import sys
import types
import unittest
class Test_TestLoader(unittest.TestCase):
### Basic object tests
@@ -76,7 +76,7 @@ class Test_TestLoader(unittest.TestCase):
loader = unittest.TestLoader()
# This has to be false for the test to succeed
self.assertFalse('runTest'.startswith(loader.testMethodPrefix))
self.assertNotStartsWith('runTest', loader.testMethodPrefix)
suite = loader.loadTestsFromTestCase(Foo)
self.assertIsInstance(suite, loader.suiteClass)
+8 -8
View File
@@ -1,9 +1,10 @@
import os
import sys
import subprocess
from test import support
import sys
import unittest
import test.test_unittest
from test import support
from test.test_unittest.test_result import BufferedWriter
@@ -135,14 +136,14 @@ class Test_TestProgram(unittest.TestCase):
argv=["foobar"],
testRunner=unittest.TextTestRunner(stream=stream),
testLoader=self.TestLoader(self.FooBar))
self.assertTrue(hasattr(program, 'result'))
self.assertHasAttr(program, 'result')
out = stream.getvalue()
self.assertIn('\nFAIL: testFail ', out)
self.assertIn('\nERROR: testError ', out)
self.assertIn('\nUNEXPECTED SUCCESS: testUnexpectedSuccess ', out)
expected = ('\n\nFAILED (failures=1, errors=1, skipped=1, '
'expected failures=1, unexpected successes=1)\n')
self.assertTrue(out.endswith(expected))
self.assertEndsWith(out, expected)
def test_Exit(self):
stream = BufferedWriter()
@@ -159,7 +160,7 @@ class Test_TestProgram(unittest.TestCase):
self.assertIn('\nUNEXPECTED SUCCESS: testUnexpectedSuccess ', out)
expected = ('\n\nFAILED (failures=1, errors=1, skipped=1, '
'expected failures=1, unexpected successes=1)\n')
self.assertTrue(out.endswith(expected))
self.assertEndsWith(out, expected)
def test_ExitAsDefault(self):
stream = BufferedWriter()
@@ -174,7 +175,7 @@ class Test_TestProgram(unittest.TestCase):
self.assertIn('\nUNEXPECTED SUCCESS: testUnexpectedSuccess ', out)
expected = ('\n\nFAILED (failures=1, errors=1, skipped=1, '
'expected failures=1, unexpected successes=1)\n')
self.assertTrue(out.endswith(expected))
self.assertEndsWith(out, expected)
def test_ExitSkippedSuite(self):
stream = BufferedWriter()
@@ -186,7 +187,7 @@ class Test_TestProgram(unittest.TestCase):
self.assertEqual(cm.exception.code, 0)
out = stream.getvalue()
expected = '\n\nOK (skipped=1)\n'
self.assertTrue(out.endswith(expected))
self.assertEndsWith(out, expected)
def test_ExitEmptySuite(self):
stream = BufferedWriter()
@@ -506,7 +507,6 @@ class TestCommandLineArgs(unittest.TestCase):
self.assertEqual(program.testNamePatterns, ['*foo*', '*bar*', '*pat*'])
@unittest.expectedFailureIf(sys.platform != 'win32', 'TODO: RUSTPYTHON')
def testSelectedTestNamesFunctionalTest(self):
def run_unittest(args):
# Use -E to ignore PYTHONSAFEPATH env var
+15 -11
View File
@@ -4,8 +4,12 @@ import textwrap
import traceback
import unittest
from unittest.util import strclass
from test.support import warnings_helper
from test.support import captured_stdout, force_not_colorized_test_class
from test.support import (
captured_stdout,
force_not_colorized_test_class,
warnings_helper,
)
from test.test_unittest.support import BufferedWriter
@@ -13,7 +17,7 @@ class MockTraceback(object):
class TracebackException:
def __init__(self, *args, **kwargs):
self.capture_locals = kwargs.get('capture_locals', False)
def format(self):
def format(self, **kwargs):
result = ['A traceback']
if self.capture_locals:
result.append('locals')
@@ -186,7 +190,7 @@ class Test_TestResult(unittest.TestCase):
test = Foo('test_1')
try:
test.fail("foo")
except:
except AssertionError:
exc_info_tuple = sys.exc_info()
result = unittest.TestResult()
@@ -214,7 +218,7 @@ class Test_TestResult(unittest.TestCase):
def get_exc_info():
try:
test.fail("foo")
except:
except AssertionError:
return sys.exc_info()
exc_info_tuple = get_exc_info()
@@ -241,9 +245,9 @@ class Test_TestResult(unittest.TestCase):
try:
try:
test.fail("foo")
except:
except AssertionError:
raise ValueError(42)
except:
except ValueError:
return sys.exc_info()
exc_info_tuple = get_exc_info()
@@ -271,7 +275,7 @@ class Test_TestResult(unittest.TestCase):
loop.__cause__ = loop
loop.__context__ = loop
raise loop
except:
except Exception:
return sys.exc_info()
exc_info_tuple = get_exc_info()
@@ -300,7 +304,7 @@ class Test_TestResult(unittest.TestCase):
ex1.__cause__ = ex2
ex2.__context__ = ex1
raise C
except:
except Exception:
return sys.exc_info()
exc_info_tuple = get_exc_info()
@@ -345,7 +349,7 @@ class Test_TestResult(unittest.TestCase):
test = Foo('test_1')
try:
raise TypeError()
except:
except TypeError:
exc_info_tuple = sys.exc_info()
result = unittest.TestResult()
@@ -454,7 +458,7 @@ class Test_TestResult(unittest.TestCase):
self.assertTrue(result.failfast)
result = runner.run(test)
stream.flush()
self.assertTrue(stream.getvalue().endswith('\n\nOK\n'))
self.assertEndsWith(stream.getvalue(), '\n\nOK\n')
@force_not_colorized_test_class
+2 -3
View File
@@ -1,13 +1,12 @@
import io
import os
import sys
import pickle
import subprocess
from test import support
import sys
import unittest
from unittest.case import _Outcome
from test import support
from test.test_unittest.support import (
BufferedWriter,
LoggingResult,
-1
View File
@@ -1,6 +1,5 @@
import io
import sys
import unittest
+3
View File
@@ -1,5 +1,6 @@
import unittest
from test.support import force_not_colorized
from test.test_unittest.support import LoggingResult
@@ -293,6 +294,7 @@ class Test_TestSkipping(unittest.TestCase):
self.assertFalse(result.unexpectedSuccesses)
self.assertTrue(result.wasSuccessful())
@force_not_colorized
def test_expected_failure_and_fail_in_cleanup(self):
class Foo(unittest.TestCase):
@unittest.expectedFailure
@@ -372,6 +374,7 @@ class Test_TestSkipping(unittest.TestCase):
self.assertEqual(result.unexpectedSuccesses, [test])
self.assertFalse(result.wasSuccessful())
@force_not_colorized
def test_unexpected_success_and_fail_in_cleanup(self):
class Foo(unittest.TestCase):
@unittest.expectedFailure
+2 -3
View File
@@ -1,10 +1,9 @@
import unittest
import gc
import sys
import unittest
import weakref
from test.test_unittest.support import LoggingResult, TestEquality
from test.test_unittest.support import LoggingResult, TestEquality
### Support code for Test_TestSuite
################################################################
+5 -1
View File
@@ -1,5 +1,9 @@
import unittest
from unittest.util import safe_repr, sorted_list_difference, unorderable_list_difference
from unittest.util import (
safe_repr,
sorted_list_difference,
unorderable_list_difference,
)
class TestUtil(unittest.TestCase):
-2
View File
@@ -101,7 +101,6 @@ class PublicAPITests(BaseTest):
public API.
"""
@unittest.expectedFailure # TODO: RUSTPYTHON 'PyPublicAPITests' object has no attribute 'assertHasAttr'
def test_module_all_attribute(self):
self.assertHasAttr(self.module, '__all__')
target_api = ["warn", "warn_explicit", "showwarning",
@@ -851,7 +850,6 @@ class PyWarnTests(WarnTests, unittest.TestCase):
# As an early adopter, we sanity check the
# test.import_helper.import_fresh_module utility function
@unittest.expectedFailure # TODO: RUSTPYTHON 'PyWarnTests' object has no attribute 'assertHasAttr'
def test_pure_python(self):
self.assertIsNot(original_warnings, self.module)
self.assertHasAttr(self.module.warn, '__code__')
+24 -9
View File
@@ -27,7 +27,7 @@ Further information is available in the bundled documentation, and from
http://docs.python.org/library/unittest.html
Copyright (c) 1999-2003 Steve Purcell
Copyright (c) 2003-2010 Python Software Foundation
Copyright (c) 2003 Python Software Foundation
This module is free software, and you may redistribute it and/or modify
it under the same terms as Python itself, so long as this copyright message
and disclaimer are retained in their original form.
@@ -53,15 +53,30 @@ __all__ = ['TestResult', 'TestCase', 'IsolatedAsyncioTestCase', 'TestSuite',
__unittest = True
from .result import TestResult
from .case import (addModuleCleanup, TestCase, FunctionTestCase, SkipTest, skip,
skipIf, skipUnless, expectedFailure, doModuleCleanups,
enterModuleContext)
from .suite import BaseTestSuite, TestSuite
from .case import (
FunctionTestCase,
SkipTest,
TestCase,
addModuleCleanup,
doModuleCleanups,
enterModuleContext,
expectedFailure,
skip,
skipIf,
skipUnless,
)
from .loader import TestLoader, defaultTestLoader
from .main import TestProgram, main
from .runner import TextTestRunner, TextTestResult
from .signals import installHandler, registerResult, removeResult, removeHandler
from .main import TestProgram, main # noqa: F401
from .result import TestResult
from .runner import TextTestResult, TextTestRunner
from .signals import (
installHandler,
registerResult,
removeHandler,
removeResult,
)
from .suite import BaseTestSuite, TestSuite # noqa: F401
# IsolatedAsyncioTestCase will be imported lazily.
+1
View File
@@ -1,6 +1,7 @@
"""Main entry point"""
import sys
if sys.argv[0].endswith("__main__.py"):
import os.path
# We change sys.argv[0] to make help message more useful
+1 -2
View File
@@ -1,9 +1,8 @@
import logging
import collections
import logging
from .case import _BaseTestCaseContext
_LoggingWatcher = collections.namedtuple("_LoggingWatcher",
["records", "output"])
+18 -6
View File
@@ -75,9 +75,17 @@ class IsolatedAsyncioTestCase(TestCase):
enter = cls.__aenter__
exit = cls.__aexit__
except AttributeError:
raise TypeError(f"'{cls.__module__}.{cls.__qualname__}' object does "
f"not support the asynchronous context manager protocol"
) from None
msg = (f"'{cls.__module__}.{cls.__qualname__}' object does "
"not support the asynchronous context manager protocol")
try:
cls.__enter__
cls.__exit__
except AttributeError:
pass
else:
msg += (" but it supports the context manager protocol. "
"Did you mean to use enterContext()?")
raise TypeError(msg) from None
result = await enter(cm)
self.addAsyncCleanup(exit, cm, None, None, None)
return result
@@ -91,9 +99,13 @@ class IsolatedAsyncioTestCase(TestCase):
self._callAsync(self.asyncSetUp)
def _callTestMethod(self, method):
if self._callMaybeAsync(method) is not None:
warnings.warn(f'It is deprecated to return a value that is not None from a '
f'test case ({method})', DeprecationWarning, stacklevel=4)
result = self._callMaybeAsync(method)
if result is not None:
msg = (
f'It is deprecated to return a value that is not None '
f'from a test case ({method} returned {type(result).__name__!r})',
)
warnings.warn(msg, DeprecationWarning, stacklevel=4)
def _callTearDown(self):
self._callAsync(self.asyncTearDown)
+171 -16
View File
@@ -1,20 +1,25 @@
"""Test case implementation"""
import sys
import functools
import difflib
import pprint
import re
import warnings
import collections
import contextlib
import traceback
import difflib
import functools
import pprint
import re
import sys
import time
import traceback
import types
import warnings
from . import result
from .util import (strclass, safe_repr, _count_diff_all_purpose,
_count_diff_hashable, _common_shorten_repr)
from .util import (
_common_shorten_repr,
_count_diff_all_purpose,
_count_diff_hashable,
safe_repr,
strclass,
)
__unittest = True
@@ -111,8 +116,17 @@ def _enter_context(cm, addcleanup):
enter = cls.__enter__
exit = cls.__exit__
except AttributeError:
raise TypeError(f"'{cls.__module__}.{cls.__qualname__}' object does "
f"not support the context manager protocol") from None
msg = (f"'{cls.__module__}.{cls.__qualname__}' object does "
"not support the context manager protocol")
try:
cls.__aenter__
cls.__aexit__
except AttributeError:
pass
else:
msg += (" but it supports the asynchronous context manager "
"protocol. Did you mean to use enterAsyncContext()?")
raise TypeError(msg) from None
result = enter(cm)
addcleanup(exit, cm, None, None, None)
return result
@@ -603,9 +617,18 @@ class TestCase(object):
self.setUp()
def _callTestMethod(self, method):
if method() is not None:
warnings.warn(f'It is deprecated to return a value that is not None from a '
f'test case ({method})', DeprecationWarning, stacklevel=3)
result = method()
if result is not None:
import inspect
msg = (
f'It is deprecated to return a value that is not None '
f'from a test case ({method} returned {type(result).__name__!r})'
)
if inspect.iscoroutine(result):
msg += (
'. Maybe you forgot to use IsolatedAsyncioTestCase as the base class?'
)
warnings.warn(msg, DeprecationWarning, stacklevel=3)
def _callTearDown(self):
self.tearDown()
@@ -1312,13 +1335,71 @@ class TestCase(object):
"""Same as self.assertTrue(isinstance(obj, cls)), with a nicer
default message."""
if not isinstance(obj, cls):
standardMsg = '%s is not an instance of %r' % (safe_repr(obj), cls)
if isinstance(cls, tuple):
standardMsg = f'{safe_repr(obj)} is not an instance of any of {cls!r}'
else:
standardMsg = f'{safe_repr(obj)} is not an instance of {cls!r}'
self.fail(self._formatMessage(msg, standardMsg))
def assertNotIsInstance(self, obj, cls, msg=None):
"""Included for symmetry with assertIsInstance."""
if isinstance(obj, cls):
standardMsg = '%s is an instance of %r' % (safe_repr(obj), cls)
if isinstance(cls, tuple):
for x in cls:
if isinstance(obj, x):
cls = x
break
standardMsg = f'{safe_repr(obj)} is an instance of {cls!r}'
self.fail(self._formatMessage(msg, standardMsg))
def assertIsSubclass(self, cls, superclass, msg=None):
try:
if issubclass(cls, superclass):
return
except TypeError:
if not isinstance(cls, type):
self.fail(self._formatMessage(msg, f'{cls!r} is not a class'))
raise
if isinstance(superclass, tuple):
standardMsg = f'{cls!r} is not a subclass of any of {superclass!r}'
else:
standardMsg = f'{cls!r} is not a subclass of {superclass!r}'
self.fail(self._formatMessage(msg, standardMsg))
def assertNotIsSubclass(self, cls, superclass, msg=None):
try:
if not issubclass(cls, superclass):
return
except TypeError:
if not isinstance(cls, type):
self.fail(self._formatMessage(msg, f'{cls!r} is not a class'))
raise
if isinstance(superclass, tuple):
for x in superclass:
if issubclass(cls, x):
superclass = x
break
standardMsg = f'{cls!r} is a subclass of {superclass!r}'
self.fail(self._formatMessage(msg, standardMsg))
def assertHasAttr(self, obj, name, msg=None):
if not hasattr(obj, name):
if isinstance(obj, types.ModuleType):
standardMsg = f'module {obj.__name__!r} has no attribute {name!r}'
elif isinstance(obj, type):
standardMsg = f'type object {obj.__name__!r} has no attribute {name!r}'
else:
standardMsg = f'{type(obj).__name__!r} object has no attribute {name!r}'
self.fail(self._formatMessage(msg, standardMsg))
def assertNotHasAttr(self, obj, name, msg=None):
if hasattr(obj, name):
if isinstance(obj, types.ModuleType):
standardMsg = f'module {obj.__name__!r} has unexpected attribute {name!r}'
elif isinstance(obj, type):
standardMsg = f'type object {obj.__name__!r} has unexpected attribute {name!r}'
else:
standardMsg = f'{type(obj).__name__!r} object has unexpected attribute {name!r}'
self.fail(self._formatMessage(msg, standardMsg))
def assertRaisesRegex(self, expected_exception, expected_regex,
@@ -1382,6 +1463,80 @@ class TestCase(object):
msg = self._formatMessage(msg, standardMsg)
raise self.failureException(msg)
def _tail_type_check(self, s, tails, msg):
if not isinstance(tails, tuple):
tails = (tails,)
for tail in tails:
if isinstance(tail, str):
if not isinstance(s, str):
self.fail(self._formatMessage(msg,
f'Expected str, not {type(s).__name__}'))
elif isinstance(tail, (bytes, bytearray)):
if not isinstance(s, (bytes, bytearray)):
self.fail(self._formatMessage(msg,
f'Expected bytes, not {type(s).__name__}'))
def assertStartsWith(self, s, prefix, msg=None):
try:
if s.startswith(prefix):
return
except (AttributeError, TypeError):
self._tail_type_check(s, prefix, msg)
raise
a = safe_repr(s, short=True)
b = safe_repr(prefix)
if isinstance(prefix, tuple):
standardMsg = f"{a} doesn't start with any of {b}"
else:
standardMsg = f"{a} doesn't start with {b}"
self.fail(self._formatMessage(msg, standardMsg))
def assertNotStartsWith(self, s, prefix, msg=None):
try:
if not s.startswith(prefix):
return
except (AttributeError, TypeError):
self._tail_type_check(s, prefix, msg)
raise
if isinstance(prefix, tuple):
for x in prefix:
if s.startswith(x):
prefix = x
break
a = safe_repr(s, short=True)
b = safe_repr(prefix)
self.fail(self._formatMessage(msg, f"{a} starts with {b}"))
def assertEndsWith(self, s, suffix, msg=None):
try:
if s.endswith(suffix):
return
except (AttributeError, TypeError):
self._tail_type_check(s, suffix, msg)
raise
a = safe_repr(s, short=True)
b = safe_repr(suffix)
if isinstance(suffix, tuple):
standardMsg = f"{a} doesn't end with any of {b}"
else:
standardMsg = f"{a} doesn't end with {b}"
self.fail(self._formatMessage(msg, standardMsg))
def assertNotEndsWith(self, s, suffix, msg=None):
try:
if not s.endswith(suffix):
return
except (AttributeError, TypeError):
self._tail_type_check(s, suffix, msg)
raise
if isinstance(suffix, tuple):
for x in suffix:
if s.endswith(x):
suffix = x
break
a = safe_repr(s, short=True)
b = safe_repr(suffix)
self.fail(self._formatMessage(msg, f"{a} ends with {b}"))
class FunctionTestCase(TestCase):
+46 -16
View File
@@ -1,12 +1,11 @@
"""Loading unittests."""
import functools
import os
import re
import sys
import traceback
import types
import functools
from fnmatch import fnmatch, fnmatchcase
from . import case, suite, util
@@ -274,6 +273,8 @@ class TestLoader(object):
self._top_level_dir = top_level_dir
is_not_importable = False
is_namespace = False
tests = []
if os.path.isdir(os.path.abspath(start_dir)):
start_dir = os.path.abspath(start_dir)
if start_dir != top_level_dir:
@@ -286,12 +287,25 @@ class TestLoader(object):
is_not_importable = True
else:
the_module = sys.modules[start_dir]
top_part = start_dir.split('.')[0]
try:
start_dir = os.path.abspath(
os.path.dirname((the_module.__file__)))
except AttributeError:
if the_module.__name__ in sys.builtin_module_names:
if not hasattr(the_module, "__file__") or the_module.__file__ is None:
# look for namespace packages
try:
spec = the_module.__spec__
except AttributeError:
spec = None
if spec and spec.submodule_search_locations is not None:
is_namespace = True
for path in the_module.__path__:
if (not set_implicit_top and
not path.startswith(top_level_dir)):
continue
self._top_level_dir = \
(path.split(the_module.__name__
.replace(".", os.path.sep))[0])
tests.extend(self._find_tests(path, pattern, namespace=True))
elif the_module.__name__ in sys.builtin_module_names:
# builtin module
raise TypeError('Can not use builtin modules '
'as dotted module names') from None
@@ -300,14 +314,27 @@ class TestLoader(object):
f"don't know how to discover from {the_module!r}"
) from None
else:
top_part = start_dir.split('.')[0]
start_dir = os.path.abspath(os.path.dirname((the_module.__file__)))
if set_implicit_top:
self._top_level_dir = self._get_directory_containing_module(top_part)
if not is_namespace:
if sys.modules[top_part].__file__ is None:
self._top_level_dir = os.path.dirname(the_module.__file__)
if self._top_level_dir not in sys.path:
sys.path.insert(0, self._top_level_dir)
else:
self._top_level_dir = \
self._get_directory_containing_module(top_part)
sys.path.remove(top_level_dir)
if is_not_importable:
raise ImportError('Start directory is not importable: %r' % start_dir)
tests = list(self._find_tests(start_dir, pattern))
if not is_namespace:
tests = list(self._find_tests(start_dir, pattern))
self._top_level_dir = original_top_level_dir
return self.suiteClass(tests)
@@ -343,7 +370,7 @@ class TestLoader(object):
# override this method to use alternative matching strategy
return fnmatch(path, pattern)
def _find_tests(self, start_dir, pattern):
def _find_tests(self, start_dir, pattern, namespace=False):
"""Used by discovery. Yields test suites it loads."""
# Handle the __init__ in this package
name = self._get_name_from_path(start_dir)
@@ -352,7 +379,8 @@ class TestLoader(object):
if name != '.' and name not in self._loading_packages:
# name is in self._loading_packages while we have called into
# loadTestsFromModule with name.
tests, should_recurse = self._find_test_path(start_dir, pattern)
tests, should_recurse = self._find_test_path(
start_dir, pattern, namespace)
if tests is not None:
yield tests
if not should_recurse:
@@ -363,7 +391,8 @@ class TestLoader(object):
paths = sorted(os.listdir(start_dir))
for path in paths:
full_path = os.path.join(start_dir, path)
tests, should_recurse = self._find_test_path(full_path, pattern)
tests, should_recurse = self._find_test_path(
full_path, pattern, False)
if tests is not None:
yield tests
if should_recurse:
@@ -371,11 +400,11 @@ class TestLoader(object):
name = self._get_name_from_path(full_path)
self._loading_packages.add(name)
try:
yield from self._find_tests(full_path, pattern)
yield from self._find_tests(full_path, pattern, False)
finally:
self._loading_packages.discard(name)
def _find_test_path(self, full_path, pattern):
def _find_test_path(self, full_path, pattern, namespace=False):
"""Used by discovery.
Loads tests from a single file, or a directories' __init__.py when
@@ -419,7 +448,8 @@ class TestLoader(object):
msg % (mod_name, module_dir, expected_dir))
return self.loadTestsFromModule(module, pattern=pattern), False
elif os.path.isdir(full_path):
if not os.path.isfile(os.path.join(full_path, '__init__.py')):
if (not namespace and
not os.path.isfile(os.path.join(full_path, '__init__.py'))):
return None, False
load_tests = None
+3 -3
View File
@@ -1,8 +1,8 @@
"""Unittest main program"""
import sys
import argparse
import os
import sys
from . import loader, runner
from .signals import installHandler
@@ -197,7 +197,7 @@ class TestProgram(object):
return parser
def _getMainArgParser(self, parent):
parser = argparse.ArgumentParser(parents=[parent])
parser = argparse.ArgumentParser(parents=[parent], color=True)
parser.prog = self.progName
parser.print_help = self._print_help
@@ -208,7 +208,7 @@ class TestProgram(object):
return parser
def _getDiscoveryArgParser(self, parent):
parser = argparse.ArgumentParser(parents=[parent])
parser = argparse.ArgumentParser(parents=[parent], color=True)
parser.prog = '%s discover' % self.progName
parser.epilog = ('For test discovery all test modules must be '
'importable from the top level directory of the '
+35 -17
View File
@@ -25,19 +25,20 @@ __all__ = (
import asyncio
import builtins
import contextlib
import io
import inspect
import io
import pkgutil
import pprint
import sys
import builtins
import pkgutil
from asyncio import iscoroutinefunction
import threading
from types import CodeType, ModuleType, MethodType
from unittest.util import safe_repr
from functools import wraps, partial
from dataclasses import fields, is_dataclass
from functools import partial, wraps
from inspect import iscoroutinefunction
from threading import RLock
from types import CodeType, MethodType, ModuleType
from unittest.util import safe_repr
class InvalidSpecError(Exception):
@@ -568,6 +569,11 @@ class NonCallableMock(Base):
__dict__['_mock_methods'] = spec
__dict__['_spec_asyncs'] = _spec_asyncs
def _mock_extend_spec_methods(self, spec_methods):
methods = self.__dict__.get('_mock_methods') or []
methods.extend(spec_methods)
self.__dict__['_mock_methods'] = methods
def __get_return_value(self):
ret = self._mock_return_value
if self._mock_delegate is not None:
@@ -1766,7 +1772,7 @@ def patch(
the patch is undone.
If `new` is omitted, then the target is replaced with an
`AsyncMock if the patched object is an async function or a
`AsyncMock` if the patched object is an async function or a
`MagicMock` otherwise. If `patch` is used as a decorator and `new` is
omitted, the created mock is passed in as an extra argument to the
decorated function. If `patch` is used as a context manager the created
@@ -1840,7 +1846,8 @@ def patch(
class _patch_dict(object):
"""
Patch a dictionary, or dictionary like object, and restore the dictionary
to its original state after the test.
to its original state after the test, where the restored dictionary is
a copy of the dictionary as it was before the test.
`in_dict` can be a dictionary or a mapping like container. If it is a
mapping then it must at least support getting, setting and deleting items
@@ -2176,8 +2183,6 @@ class MagicMixin(Base):
if getattr(self, "_mock_methods", None) is not None:
these_magics = orig_magics.intersection(self._mock_methods)
remove_magics = set()
remove_magics = orig_magics - these_magics
for entry in remove_magics:
@@ -2477,7 +2482,7 @@ class AsyncMock(AsyncMockMixin, AsyncMagicMixin, Mock):
recognized as an async function, and the result of a call is an awaitable:
>>> mock = AsyncMock()
>>> iscoroutinefunction(mock)
>>> inspect.iscoroutinefunction(mock)
True
>>> inspect.isawaitable(mock())
True
@@ -2767,6 +2772,16 @@ def create_autospec(spec, spec_set=False, instance=False, _parent=None,
f'[object={spec!r}]')
is_async_func = _is_async_func(spec)
_kwargs = {'spec': spec}
entries = [(entry, _missing) for entry in dir(spec)]
if is_type and instance and is_dataclass(spec):
is_dataclass_spec = True
dataclass_fields = fields(spec)
entries.extend((f.name, f.type) for f in dataclass_fields)
dataclass_spec_list = [f.name for f in dataclass_fields]
else:
is_dataclass_spec = False
if spec_set:
_kwargs = {'spec_set': spec}
elif spec is None:
@@ -2802,6 +2817,8 @@ def create_autospec(spec, spec_set=False, instance=False, _parent=None,
mock = Klass(parent=_parent, _new_parent=_parent, _new_name=_new_name,
name=_name, **_kwargs)
if is_dataclass_spec:
mock._mock_extend_spec_methods(dataclass_spec_list)
if isinstance(spec, FunctionTypes):
# should only happen at the top level because we don't
@@ -2823,7 +2840,7 @@ def create_autospec(spec, spec_set=False, instance=False, _parent=None,
_name='()', _parent=mock,
wraps=wrapped)
for entry in dir(spec):
for entry, original in entries:
if _is_magic(entry):
# MagicMock already does the useful magic methods for us
continue
@@ -2837,10 +2854,11 @@ def create_autospec(spec, spec_set=False, instance=False, _parent=None,
# AttributeError on being fetched?
# we could be resilient against it, or catch and propagate the
# exception when the attribute is fetched from the mock
try:
original = getattr(spec, entry)
except AttributeError:
continue
if original is _missing:
try:
original = getattr(spec, entry)
except AttributeError:
continue
child_kwargs = {'spec': original}
# Wrap child attributes also.
+5 -2
View File
@@ -3,9 +3,9 @@
import io
import sys
import traceback
from functools import wraps
from . import util
from functools import wraps
__unittest = True
@@ -189,7 +189,10 @@ class TestResult(object):
tb_e = traceback.TracebackException(
exctype, value, tb,
capture_locals=self.tb_locals, compact=True)
msgLines = list(tb_e.format())
from _colorize import can_colorize
colorize = hasattr(self, "stream") and can_colorize(file=self.stream)
msgLines = list(tb_e.format(colorize=colorize))
if self.buffer:
output = sys.stdout.getvalue()
+61 -40
View File
@@ -4,6 +4,8 @@ import sys
import time
import warnings
from _colorize import get_theme
from . import result
from .case import _SubTest
from .signals import registerResult
@@ -13,18 +15,18 @@ __unittest = True
class _WritelnDecorator(object):
"""Used to decorate file-like objects with a handy 'writeln' method"""
def __init__(self,stream):
def __init__(self, stream):
self.stream = stream
def __getattr__(self, attr):
if attr in ('stream', '__getstate__'):
raise AttributeError(attr)
return getattr(self.stream,attr)
return getattr(self.stream, attr)
def writeln(self, arg=None):
if arg:
self.write(arg)
self.write('\n') # text-mode streams translate to \r\n if needed
self.write('\n') # text-mode streams translate to \r\n if needed
class TextTestResult(result.TestResult):
@@ -43,6 +45,7 @@ class TextTestResult(result.TestResult):
self.showAll = verbosity > 1
self.dots = verbosity == 1
self.descriptions = descriptions
self._theme = get_theme(tty_file=stream).unittest
self._newline = True
self.durations = durations
@@ -76,86 +79,100 @@ class TextTestResult(result.TestResult):
def addSubTest(self, test, subtest, err):
if err is not None:
t = self._theme
if self.showAll:
if issubclass(err[0], subtest.failureException):
self._write_status(subtest, "FAIL")
self._write_status(subtest, f"{t.fail}FAIL{t.reset}")
else:
self._write_status(subtest, "ERROR")
self._write_status(subtest, f"{t.fail}ERROR{t.reset}")
elif self.dots:
if issubclass(err[0], subtest.failureException):
self.stream.write('F')
self.stream.write(f"{t.fail}F{t.reset}")
else:
self.stream.write('E')
self.stream.write(f"{t.fail}E{t.reset}")
self.stream.flush()
super(TextTestResult, self).addSubTest(test, subtest, err)
def addSuccess(self, test):
super(TextTestResult, self).addSuccess(test)
t = self._theme
if self.showAll:
self._write_status(test, "ok")
self._write_status(test, f"{t.passed}ok{t.reset}")
elif self.dots:
self.stream.write('.')
self.stream.write(f"{t.passed}.{t.reset}")
self.stream.flush()
def addError(self, test, err):
super(TextTestResult, self).addError(test, err)
t = self._theme
if self.showAll:
self._write_status(test, "ERROR")
self._write_status(test, f"{t.fail}ERROR{t.reset}")
elif self.dots:
self.stream.write('E')
self.stream.write(f"{t.fail}E{t.reset}")
self.stream.flush()
def addFailure(self, test, err):
super(TextTestResult, self).addFailure(test, err)
t = self._theme
if self.showAll:
self._write_status(test, "FAIL")
self._write_status(test, f"{t.fail}FAIL{t.reset}")
elif self.dots:
self.stream.write('F')
self.stream.write(f"{t.fail}F{t.reset}")
self.stream.flush()
def addSkip(self, test, reason):
super(TextTestResult, self).addSkip(test, reason)
t = self._theme
if self.showAll:
self._write_status(test, "skipped {0!r}".format(reason))
self._write_status(test, f"{t.warn}skipped{t.reset} {reason!r}")
elif self.dots:
self.stream.write("s")
self.stream.write(f"{t.warn}s{t.reset}")
self.stream.flush()
def addExpectedFailure(self, test, err):
super(TextTestResult, self).addExpectedFailure(test, err)
t = self._theme
if self.showAll:
self.stream.writeln("expected failure")
self.stream.writeln(f"{t.warn}expected failure{t.reset}")
self.stream.flush()
elif self.dots:
self.stream.write("x")
self.stream.write(f"{t.warn}x{t.reset}")
self.stream.flush()
def addUnexpectedSuccess(self, test):
super(TextTestResult, self).addUnexpectedSuccess(test)
t = self._theme
if self.showAll:
self.stream.writeln("unexpected success")
self.stream.writeln(f"{t.fail}unexpected success{t.reset}")
self.stream.flush()
elif self.dots:
self.stream.write("u")
self.stream.write(f"{t.fail}u{t.reset}")
self.stream.flush()
def printErrors(self):
t = self._theme
if self.dots or self.showAll:
self.stream.writeln()
self.stream.flush()
self.printErrorList('ERROR', self.errors)
self.printErrorList('FAIL', self.failures)
unexpectedSuccesses = getattr(self, 'unexpectedSuccesses', ())
self.printErrorList(f"{t.fail}ERROR{t.reset}", self.errors)
self.printErrorList(f"{t.fail}FAIL{t.reset}", self.failures)
unexpectedSuccesses = getattr(self, "unexpectedSuccesses", ())
if unexpectedSuccesses:
self.stream.writeln(self.separator1)
for test in unexpectedSuccesses:
self.stream.writeln(f"UNEXPECTED SUCCESS: {self.getDescription(test)}")
self.stream.writeln(
f"{t.fail}UNEXPECTED SUCCESS{t.fail_info}: "
f"{self.getDescription(test)}{t.reset}"
)
self.stream.flush()
def printErrorList(self, flavour, errors):
t = self._theme
for test, err in errors:
self.stream.writeln(self.separator1)
self.stream.writeln("%s: %s" % (flavour,self.getDescription(test)))
self.stream.writeln(
f"{flavour}{t.fail_info}: {self.getDescription(test)}{t.reset}"
)
self.stream.writeln(self.separator2)
self.stream.writeln("%s" % err)
self.stream.flush()
@@ -232,7 +249,7 @@ class TextTestRunner(object):
if self.warnings:
# if self.warnings is set, use it to filter all the warnings
warnings.simplefilter(self.warnings)
startTime = time.perf_counter()
start_time = time.perf_counter()
startTestRun = getattr(result, 'startTestRun', None)
if startTestRun is not None:
startTestRun()
@@ -242,8 +259,8 @@ class TextTestRunner(object):
stopTestRun = getattr(result, 'stopTestRun', None)
if stopTestRun is not None:
stopTestRun()
stopTime = time.perf_counter()
timeTaken = stopTime - startTime
stop_time = time.perf_counter()
time_taken = stop_time - start_time
result.printErrors()
if self.durations is not None:
self._printDurations(result)
@@ -253,10 +270,10 @@ class TextTestRunner(object):
run = result.testsRun
self.stream.writeln("Ran %d test%s in %.3fs" %
(run, run != 1 and "s" or "", timeTaken))
(run, run != 1 and "s" or "", time_taken))
self.stream.writeln()
expectedFails = unexpectedSuccesses = skipped = 0
expected_fails = unexpected_successes = skipped = 0
try:
results = map(len, (result.expectedFailures,
result.unexpectedSuccesses,
@@ -264,26 +281,30 @@ class TextTestRunner(object):
except AttributeError:
pass
else:
expectedFails, unexpectedSuccesses, skipped = results
expected_fails, unexpected_successes, skipped = results
infos = []
t = get_theme(tty_file=self.stream).unittest
if not result.wasSuccessful():
self.stream.write("FAILED")
self.stream.write(f"{t.fail_info}FAILED{t.reset}")
failed, errored = len(result.failures), len(result.errors)
if failed:
infos.append("failures=%d" % failed)
infos.append(f"{t.fail_info}failures={failed}{t.reset}")
if errored:
infos.append("errors=%d" % errored)
infos.append(f"{t.fail_info}errors={errored}{t.reset}")
elif run == 0 and not skipped:
self.stream.write("NO TESTS RAN")
self.stream.write(f"{t.warn}NO TESTS RAN{t.reset}")
else:
self.stream.write("OK")
self.stream.write(f"{t.passed}OK{t.reset}")
if skipped:
infos.append("skipped=%d" % skipped)
if expectedFails:
infos.append("expected failures=%d" % expectedFails)
if unexpectedSuccesses:
infos.append("unexpected successes=%d" % unexpectedSuccesses)
infos.append(f"{t.warn}skipped={skipped}{t.reset}")
if expected_fails:
infos.append(f"{t.warn}expected failures={expected_fails}{t.reset}")
if unexpected_successes:
infos.append(
f"{t.fail}unexpected successes={unexpected_successes}{t.reset}"
)
if infos:
self.stream.writeln(" (%s)" % (", ".join(infos),))
else:
-1
View File
@@ -1,6 +1,5 @@
import signal
import weakref
from functools import wraps
__unittest = True
+1 -2
View File
@@ -2,8 +2,7 @@
import sys
from . import case
from . import util
from . import case, util
__unittest = True
+1 -1
View File
@@ -1,6 +1,6 @@
"""Various utility functions."""
from collections import namedtuple, Counter
from collections import Counter, namedtuple
from os.path import commonprefix
__unittest = True