mirror of
https://github.com/RustPython/RustPython.git
synced 2026-10-28 10:44:36 +08:00
Update Python 3.15 test harness helpers and discovery (#9364)
Assisted-by: OpenAI Codex: model version unavailable
This commit is contained in:
1 parent
edda30d2db
commit
0dca682e45
3 files changed
+262
-6
No files matched your search
Vendored
+4
-1
@@ -21,13 +21,16 @@ SPLITTESTDIRS: set[TestName] = {
|
||||
"test_asyncio",
|
||||
"test_concurrent_futures",
|
||||
"test_doctests",
|
||||
"test_free_threading",
|
||||
"test_future_stmt",
|
||||
"test_gdb",
|
||||
"test_inspect",
|
||||
"test_pydoc",
|
||||
"test_io",
|
||||
"test_multiprocessing_fork",
|
||||
"test_multiprocessing_forkserver",
|
||||
"test_multiprocessing_spawn",
|
||||
"test_os",
|
||||
"test_pydoc",
|
||||
}
|
||||
|
||||
|
||||
|
||||
Vendored
+13
-3
@@ -658,6 +658,7 @@ def requires_working_socket(*, module=False):
|
||||
else:
|
||||
return unittest.skipUnless(has_socket_support, msg)
|
||||
|
||||
|
||||
@functools.cache
|
||||
def has_remote_subprocess_debugging():
|
||||
"""Check if we have permissions to debug subprocesses remotely.
|
||||
@@ -1245,7 +1246,6 @@ class _MemoryWatchdog:
|
||||
self.started = False
|
||||
|
||||
def start(self):
|
||||
import warnings
|
||||
try:
|
||||
f = open(self.procfile, 'r')
|
||||
except OSError as e:
|
||||
@@ -1281,6 +1281,7 @@ def bigmemtest(size, memuse, dry_run=True):
|
||||
test doesn't support dummy runs when -M is not specified.
|
||||
"""
|
||||
def decorator(f):
|
||||
@functools.wraps(f)
|
||||
def wrapper(self):
|
||||
size = wrapper.size
|
||||
memuse = wrapper.memuse
|
||||
@@ -1333,6 +1334,7 @@ def nomemtest(f):
|
||||
|
||||
def bigaddrspacetest(f):
|
||||
"""Decorator for tests that fill the address space."""
|
||||
@functools.wraps(f)
|
||||
def wrapper(self):
|
||||
if max_memuse < MAX_Py_ssize_t:
|
||||
if MAX_Py_ssize_t >= 2**63 - 1 and max_memuse >= 2**31:
|
||||
@@ -1414,7 +1416,7 @@ def no_tracing(func):
|
||||
sys.settrace(original_trace)
|
||||
|
||||
coverage_wrapper = trace_wrapper
|
||||
if 'test.cov' in sys.modules: # -Xpresite=test.cov used
|
||||
if 'test.cov' in sys.modules: # -Xpresite=test.cov:enable used
|
||||
cov = sys.monitoring.COVERAGE_ID
|
||||
@functools.wraps(func)
|
||||
def coverage_wrapper(*args, **kwargs):
|
||||
@@ -1438,6 +1440,7 @@ def no_rerun(reason):
|
||||
def deco(func):
|
||||
assert not isinstance(func, type), func
|
||||
_has_run = False
|
||||
@functools.wraps(func)
|
||||
def wrapper(self):
|
||||
nonlocal _has_run
|
||||
if _has_run:
|
||||
@@ -3003,7 +3006,7 @@ def iter_builtin_types():
|
||||
# Fall back to making a best-effort guess.
|
||||
if hasattr(object, '__flags__'):
|
||||
# Look for any type object with the Py_TPFLAGS_STATIC_BUILTIN flag set.
|
||||
import datetime
|
||||
import datetime # noqa: F401
|
||||
seen = set()
|
||||
for cls, subs in walk_class_hierarchy(object):
|
||||
if cls in seen:
|
||||
@@ -3139,6 +3142,13 @@ def force_color(color: bool):
|
||||
import _colorize
|
||||
from .os_helper import EnvironmentVarGuard
|
||||
|
||||
if color:
|
||||
try:
|
||||
import _pyrepl # noqa: F401
|
||||
except ModuleNotFoundError:
|
||||
# Can't force enable color without _pyrepl, so just skip.
|
||||
raise unittest.SkipTest("_pyrepl is missing")
|
||||
|
||||
with (
|
||||
swap_attr(_colorize, "can_colorize", lambda *, file=None: color),
|
||||
EnvironmentVarGuard() as env,
|
||||
|
||||
Vendored
+245
-2
@@ -1,6 +1,8 @@
|
||||
import contextlib
|
||||
import errno
|
||||
import importlib
|
||||
import itertools
|
||||
import inspect
|
||||
import io
|
||||
import logging
|
||||
import os
|
||||
@@ -18,6 +20,7 @@ import warnings
|
||||
|
||||
from test import support
|
||||
from test.support import isolation
|
||||
from test.support import hashlib_helper
|
||||
from test.support import import_helper
|
||||
from test.support import os_helper
|
||||
from test.support import script_helper
|
||||
@@ -94,9 +97,15 @@ class TestSupport(unittest.TestCase):
|
||||
self.test_get_attribute)
|
||||
self.assertRaises(unittest.SkipTest, support.get_attribute, self, "foo")
|
||||
|
||||
@unittest.skip("failing buildbots")
|
||||
@unittest.skipIf(support.is_android or support.is_apple_mobile,
|
||||
'Mobile platforms redirect stdout to system log')
|
||||
def test_get_original_stdout(self):
|
||||
self.assertEqual(support.get_original_stdout(), sys.stdout)
|
||||
if isinstance(sys.stdout, io.StringIO):
|
||||
# gh-55258: When --junit-xml is used, stdout is a StringIO:
|
||||
# use sys.__stdout__ in this case.
|
||||
self.assertEqual(support.get_original_stdout(), sys.__stdout__)
|
||||
else:
|
||||
self.assertEqual(support.get_original_stdout(), sys.stdout)
|
||||
|
||||
def test_unload(self):
|
||||
import sched # noqa: F401
|
||||
@@ -483,6 +492,7 @@ class TestSupport(unittest.TestCase):
|
||||
|
||||
self.assertRaises(AssertionError, support.check__all__, self, unittest)
|
||||
|
||||
@warnings_helper.ignore_fork_in_thread_deprecation_warnings()
|
||||
@unittest.skipUnless(hasattr(os, 'waitpid') and hasattr(os, 'WNOHANG'),
|
||||
'need os.waitpid() and os.WNOHANG')
|
||||
@support.requires_fork()
|
||||
@@ -834,6 +844,239 @@ class TestSupport(unittest.TestCase):
|
||||
# SuppressCrashReport
|
||||
|
||||
|
||||
@hashlib_helper.requires_builtin_hashes()
|
||||
class TestHashlibSupport(unittest.TestCase):
|
||||
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
super().setUpClass()
|
||||
cls.hashlib = import_helper.import_module("hashlib")
|
||||
cls.hmac = import_helper.import_module("hmac")
|
||||
|
||||
# All C extension modules must be present since blocking
|
||||
# the built-in implementation while allowing OpenSSL or vice-versa
|
||||
# may result in failures depending on the exposed built-in hashes.
|
||||
cls._hashlib = import_helper.import_module("_hashlib")
|
||||
cls._hmac = import_helper.import_module("_hmac")
|
||||
cls._md5 = import_helper.import_module("_md5")
|
||||
|
||||
def skip_if_fips_mode(self):
|
||||
if self._hashlib.get_fips_mode():
|
||||
self.skipTest("disabled in FIPS mode")
|
||||
|
||||
def skip_if_not_fips_mode(self):
|
||||
if not self._hashlib.get_fips_mode():
|
||||
self.skipTest("requires FIPS mode")
|
||||
|
||||
def check_context(self, disabled=True):
|
||||
if disabled:
|
||||
return self.assertRaises(ValueError)
|
||||
return contextlib.nullcontext()
|
||||
|
||||
def try_import_attribute(self, fullname, default=None):
|
||||
if fullname is None:
|
||||
return default
|
||||
assert fullname.count('.') == 1, fullname
|
||||
module_name, attribute = fullname.split('.', maxsplit=1)
|
||||
try:
|
||||
module = importlib.import_module(module_name)
|
||||
except ImportError:
|
||||
return default
|
||||
try:
|
||||
return getattr(module, attribute, default)
|
||||
except TypeError:
|
||||
return default
|
||||
|
||||
def fetch_hash_function(self, name, implementation):
|
||||
info = hashlib_helper.get_hash_func_info(name)
|
||||
match hashlib_helper.Implementation(implementation):
|
||||
case hashlib_helper.Implementation.hashlib:
|
||||
method_name = info.hashlib.member_name
|
||||
assert isinstance(method_name, str), method_name
|
||||
return getattr(self.hashlib, method_name)
|
||||
case hashlib_helper.Implementation.openssl:
|
||||
method_name = info.openssl.member_name
|
||||
assert isinstance(method_name, str | None), method_name
|
||||
return getattr(self._hashlib, method_name or "", None)
|
||||
fullname = info[implementation].fullname
|
||||
return self.try_import_attribute(fullname)
|
||||
|
||||
def fetch_hmac_function(self, name):
|
||||
target = hashlib_helper.get_hmac_item_info(name)
|
||||
return target.import_member()
|
||||
|
||||
def check_openssl_hash(self, name, *, disabled=True):
|
||||
"""Check that OpenSSL HASH interface is enabled/disabled."""
|
||||
with self.check_context(disabled):
|
||||
_ = self._hashlib.new(name)
|
||||
if do_hash := self.fetch_hash_function(name, "openssl"):
|
||||
self.assertStartsWith(do_hash.__name__, 'openssl_')
|
||||
with self.check_context(disabled):
|
||||
_ = do_hash(b"")
|
||||
|
||||
def check_openssl_hmac(self, name, *, disabled=True):
|
||||
"""Check that OpenSSL HMAC interface is enabled/disabled."""
|
||||
if name in hashlib_helper.NON_HMAC_DIGEST_NAMES:
|
||||
# HMAC-BLAKE and HMAC-SHAKE raise a ValueError as they are not
|
||||
# supported at all (they do not make any sense in practice).
|
||||
with self.assertRaises(ValueError):
|
||||
self._hashlib.hmac_digest(b"", b"", name)
|
||||
else:
|
||||
with self.check_context(disabled):
|
||||
_ = self._hashlib.hmac_digest(b"", b"", name)
|
||||
# OpenSSL does not provide one-shot explicit HMAC functions
|
||||
|
||||
def check_builtin_hash(self, name, *, disabled=True):
|
||||
"""Check that HACL* HASH interface is enabled/disabled."""
|
||||
if do_hash := self.fetch_hash_function(name, "builtin"):
|
||||
self.assertEqual(do_hash.__name__, name)
|
||||
with self.check_context(disabled):
|
||||
_ = do_hash(b"")
|
||||
|
||||
def check_builtin_hmac(self, name, *, disabled=True):
|
||||
"""Check that HACL* HMAC interface is enabled/disabled."""
|
||||
if name in hashlib_helper.NON_HMAC_DIGEST_NAMES:
|
||||
# HMAC-BLAKE and HMAC-SHAKE raise a ValueError as they are not
|
||||
# supported at all (they do not make any sense in practice).
|
||||
with self.assertRaises(ValueError):
|
||||
self._hmac.compute_digest(b"", b"", name)
|
||||
else:
|
||||
with self.check_context(disabled):
|
||||
_ = self._hmac.compute_digest(b"", b"", name)
|
||||
|
||||
with self.check_context(disabled):
|
||||
_ = self._hmac.new(b"", b"", name)
|
||||
|
||||
if do_hmac := self.fetch_hmac_function(name):
|
||||
self.assertStartsWith(do_hmac.__name__, 'compute_')
|
||||
with self.check_context(disabled):
|
||||
_ = do_hmac(b"", b"")
|
||||
else:
|
||||
self.assertIn(name, hashlib_helper.NON_HMAC_DIGEST_NAMES)
|
||||
|
||||
@support.subTests(
|
||||
('name', 'allow_openssl', 'allow_builtin'),
|
||||
itertools.product(
|
||||
hashlib_helper.CANONICAL_DIGEST_NAMES,
|
||||
[True, False],
|
||||
[True, False],
|
||||
)
|
||||
)
|
||||
def test_disable_hash(self, name, allow_openssl, allow_builtin):
|
||||
# In FIPS mode, the function may be available but would still need
|
||||
# to raise a ValueError, so we will test the helper separately.
|
||||
self.skip_if_fips_mode()
|
||||
flags = dict(allow_openssl=allow_openssl, allow_builtin=allow_builtin)
|
||||
is_fully_disabled = not allow_builtin and not allow_openssl
|
||||
|
||||
with hashlib_helper.block_algorithm(name, **flags):
|
||||
# OpenSSL's blake2s and blake2b are unknown names
|
||||
# when only the OpenSSL interface is available.
|
||||
if allow_openssl and not allow_builtin:
|
||||
aliases = {'blake2s': 'blake2s256', 'blake2b': 'blake2b512'}
|
||||
name_for_hashlib_new = aliases.get(name, name)
|
||||
else:
|
||||
name_for_hashlib_new = name
|
||||
|
||||
with self.check_context(is_fully_disabled):
|
||||
_ = self.hashlib.new(name_for_hashlib_new)
|
||||
|
||||
# Since _hashlib is present, explicit blake2b/blake2s constructors
|
||||
# use the built-in implementation, while others (since we are not
|
||||
# in FIPS mode and since _hashlib exists) use the OpenSSL function.
|
||||
with self.check_context(is_fully_disabled):
|
||||
_ = getattr(self.hashlib, name)()
|
||||
|
||||
self.check_openssl_hash(name, disabled=not allow_openssl)
|
||||
self.check_builtin_hash(name, disabled=not allow_builtin)
|
||||
|
||||
if name not in hashlib_helper.NON_HMAC_DIGEST_NAMES:
|
||||
with self.check_context(is_fully_disabled):
|
||||
_ = self.hmac.new(b"", b"", name)
|
||||
with self.check_context(is_fully_disabled):
|
||||
_ = self.hmac.HMAC(b"", b"", name)
|
||||
with self.check_context(is_fully_disabled):
|
||||
_ = self.hmac.digest(b"", b"", name)
|
||||
|
||||
self.check_openssl_hmac(name, disabled=not allow_openssl)
|
||||
self.check_builtin_hmac(name, disabled=not allow_builtin)
|
||||
|
||||
@hashlib_helper.block_algorithm("md5")
|
||||
def test_disable_hash_md5_in_fips_mode(self):
|
||||
self.skip_if_not_fips_mode()
|
||||
|
||||
self.assertRaises(ValueError, self.hashlib.new, "md5")
|
||||
self.assertRaises(ValueError, self._hashlib.new, "md5")
|
||||
self.assertRaises(ValueError, self.hashlib.md5)
|
||||
self.assertRaises(ValueError, self._hashlib.openssl_md5)
|
||||
|
||||
kwargs = dict(usedforsecurity=True)
|
||||
self.assertRaises(ValueError, self.hashlib.new, "md5", **kwargs)
|
||||
self.assertRaises(ValueError, self._hashlib.new, "md5", **kwargs)
|
||||
self.assertRaises(ValueError, self.hashlib.md5, **kwargs)
|
||||
self.assertRaises(ValueError, self._hashlib.openssl_md5, **kwargs)
|
||||
|
||||
@hashlib_helper.block_algorithm("md5", allow_openssl=True)
|
||||
def test_disable_hash_md5_in_fips_mode_allow_openssl(self):
|
||||
self.skip_if_not_fips_mode()
|
||||
# Allow the OpenSSL interface to be used but not the HACL* one.
|
||||
# hashlib.new("md5") is dispatched to hashlib.openssl_md5()
|
||||
self.assertRaises(ValueError, self.hashlib.new, "md5")
|
||||
# dispatched to hashlib.openssl_md5() in FIPS mode
|
||||
h2 = self.hashlib.new("md5", usedforsecurity=False)
|
||||
self.assertIsInstance(h2, self._hashlib.HASH)
|
||||
|
||||
# block_algorithm() does not mock hashlib.md5 and _hashlib.openssl_md5
|
||||
self.assertNotHasAttr(self.hashlib.md5, "__wrapped__")
|
||||
self.assertNotHasAttr(self._hashlib.openssl_md5, "__wrapped__")
|
||||
|
||||
hashlib_md5 = inspect.unwrap(self.hashlib.md5)
|
||||
self.assertIs(hashlib_md5, self._hashlib.openssl_md5)
|
||||
self.assertRaises(ValueError, self.hashlib.md5)
|
||||
# allow MD5 to be used in FIPS mode if usedforsecurity=False
|
||||
h3 = self.hashlib.md5(usedforsecurity=False)
|
||||
self.assertIsInstance(h3, self._hashlib.HASH)
|
||||
|
||||
@hashlib_helper.block_algorithm("md5", allow_builtin=True)
|
||||
def test_disable_hash_md5_in_fips_mode_allow_builtin(self):
|
||||
self.skip_if_not_fips_mode()
|
||||
# Allow the HACL* interface to be used but not the OpenSSL one.
|
||||
h1 = self.hashlib.new("md5") # dispatched to _md5.md5()
|
||||
self.assertNotIsInstance(h1, self._hashlib.HASH)
|
||||
h2 = self.hashlib.new("md5", usedforsecurity=False)
|
||||
self.assertIsInstance(h2, type(h1))
|
||||
|
||||
# block_algorithm() mocks hashlib.md5 and _hashlib.openssl_md5
|
||||
self.assertHasAttr(self.hashlib.md5, "__wrapped__")
|
||||
self.assertHasAttr(self._hashlib.openssl_md5, "__wrapped__")
|
||||
|
||||
hashlib_md5 = inspect.unwrap(self.hashlib.md5)
|
||||
openssl_md5 = inspect.unwrap(self._hashlib.openssl_md5)
|
||||
self.assertIs(hashlib_md5, openssl_md5)
|
||||
self.assertRaises(ValueError, self.hashlib.md5)
|
||||
self.assertRaises(ValueError, self.hashlib.md5,
|
||||
usedforsecurity=False)
|
||||
|
||||
@hashlib_helper.block_algorithm("md5",
|
||||
allow_openssl=True,
|
||||
allow_builtin=True)
|
||||
def test_disable_hash_md5_in_fips_mode_allow_all(self):
|
||||
self.skip_if_not_fips_mode()
|
||||
# hashlib.new() isn't blocked as it falls back to _md5.md5
|
||||
self.assertIsInstance(self.hashlib.new("md5"), self._md5.MD5Type)
|
||||
self.assertRaises(ValueError, self._hashlib.new, "md5")
|
||||
h = self._hashlib.new("md5", usedforsecurity=False)
|
||||
self.assertIsInstance(h, self._hashlib.HASH)
|
||||
|
||||
self.assertNotHasAttr(self.hashlib.md5, "__wrapped__")
|
||||
self.assertNotHasAttr(self._hashlib.openssl_md5, "__wrapped__")
|
||||
|
||||
self.assertIs(self.hashlib.md5, self._hashlib.openssl_md5)
|
||||
self.assertRaises(ValueError, self.hashlib.md5)
|
||||
h = self.hashlib.md5(usedforsecurity=False)
|
||||
self.assertIsInstance(h, self._hashlib.HASH)
|
||||
|
||||
|
||||
class TestIsolated(unittest.TestCase):
|
||||
# Drive the sample tests in test._isolated_sample (which really spawn
|
||||
# subprocesses through @isolation.runInSubprocess()) under a private
|
||||
|
||||
Reference in new issue
Block a user