diff options
Diffstat (limited to 'Lib/test/support/__init__.py')
| -rw-r--r-- | Lib/test/support/__init__.py | 2290 |
1 files changed, 531 insertions, 1759 deletions
diff --git a/Lib/test/support/__init__.py b/Lib/test/support/__init__.py index 215bab8..ccc11c1 100644 --- a/Lib/test/support/__init__.py +++ b/Lib/test/support/__init__.py @@ -1,170 +1,51 @@ """Supporting definitions for the Python regression tests.""" if __name__ != 'test.support': - raise ImportError('support must be imported from the test package') + raise ImportError('test.support must be imported from the test package') -import asyncio.events -import collections.abc import contextlib import errno -import faulthandler import fnmatch import functools import gc -import glob -import hashlib -import importlib -import importlib.util -import locale -import logging.handlers -import nntplib +import socket +import stat +import sys import os import platform -import re import shutil -import socket -import stat +import warnings +import unittest +import importlib +import UserDict +import re +import time import struct -import subprocess -import sys import sysconfig -import tempfile -import _thread -import threading -import time import types -import unittest -import urllib.error -import warnings - -from .testresult import get_test_runner - -try: - import multiprocessing.process -except ImportError: - multiprocessing = None - -try: - import zlib -except ImportError: - zlib = None - -try: - import gzip -except ImportError: - gzip = None - -try: - import bz2 -except ImportError: - bz2 = None - -try: - import lzma -except ImportError: - lzma = None try: - import resource + import thread except ImportError: - resource = None - -try: - import _hashlib -except ImportError: - _hashlib = None - -__all__ = [ - # globals - "PIPE_MAX_SIZE", "verbose", "max_memuse", "use_resources", "failfast", - # exceptions - "Error", "TestFailed", "TestDidNotRun", "ResourceDenied", - # imports - "import_module", "import_fresh_module", "CleanImport", - # modules - "unload", "forget", - # io - "record_original_stdout", "get_original_stdout", "captured_stdout", - "captured_stdin", "captured_stderr", - # filesystem - "TESTFN", "SAVEDCWD", "unlink", "rmtree", "temp_cwd", "findfile", - "create_empty_file", "can_symlink", "fs_is_case_insensitive", - # unittest - "is_resource_enabled", "requires", "requires_freebsd_version", - "requires_linux_version", "requires_mac_ver", "requires_hashdigest", - "check_syntax_error", "check_syntax_warning", - "TransientResource", "time_out", "socket_peer_reset", "ioerror_peer_reset", - "transient_internet", "BasicTestRunner", "run_unittest", "run_doctest", - "skip_unless_symlink", "requires_gzip", "requires_bz2", "requires_lzma", - "bigmemtest", "bigaddrspacetest", "cpython_only", "get_attribute", - "requires_IEEE_754", "skip_unless_xattr", "requires_zlib", - "anticipate_failure", "load_package_tests", "detect_api_mismatch", - "check__all__", "skip_unless_bind_unix_socket", "skip_if_buggy_ucrt_strfptime", - "ignore_warnings", - # sys - "is_jython", "is_android", "check_impl_detail", "unix_shell", - "setswitchinterval", - # network - "HOST", "IPV6_ENABLED", "find_unused_port", "bind_port", "open_urlresource", - "bind_unix_socket", - # processes - 'temp_umask', "reap_children", - # logging - "TestHandler", - # threads - "threading_setup", "threading_cleanup", "reap_threads", "start_threads", - # miscellaneous - "check_warnings", "check_no_resource_warning", "check_no_warnings", - "EnvironmentVarGuard", - "run_with_locale", "swap_item", - "swap_attr", "Matcher", "set_memlimit", "SuppressCrashReport", "sortdict", - "run_with_tz", "PGO", "missing_compiler_executable", "fd_count", - "ALWAYS_EQ", "NEVER_EQ", "LARGEST", "SMALLEST", - "LOOPBACK_TIMEOUT", "INTERNET_TIMEOUT", "SHORT_TIMEOUT", "LONG_TIMEOUT", - ] - - -# Timeout in seconds for tests using a network server listening on the network -# local loopback interface like 127.0.0.1. -# -# The timeout is long enough to prevent test failure: it takes into account -# that the client and the server can run in different threads or even different -# processes. -# -# The timeout should be long enough for connect(), recv() and send() methods -# of socket.socket. -LOOPBACK_TIMEOUT = 5.0 -if sys.platform == 'win32' and platform.machine() == 'ARM': - # bpo-37553: test_socket.SendfileUsingSendTest is taking longer than 2 - # seconds on Windows ARM32 buildbot - LOOPBACK_TIMEOUT = 10 - -# Timeout in seconds for network requests going to the Internet. The timeout is -# short enough to prevent a test to wait for too long if the Internet request -# is blocked for whatever reason. -# -# Usually, a timeout using INTERNET_TIMEOUT should not mark a test as failed, -# but skip the test instead: see transient_internet(). -INTERNET_TIMEOUT = 60.0 - -# Timeout in seconds to mark a test as failed if the test takes "too long". -# -# The timeout value depends on the regrtest --timeout command line option. -# -# If a test using SHORT_TIMEOUT starts to fail randomly on slow buildbots, use -# LONG_TIMEOUT instead. -SHORT_TIMEOUT = 30.0 - -# Timeout in seconds to detect when a test hangs. -# -# It is long enough to reduce the risk of test failure on the slowest Python -# buildbots. It should not be used to mark a test as failed if the test takes -# "too long". The timeout value depends on the regrtest --timeout command line -# option. -LONG_TIMEOUT = 5 * 60.0 - -_NOT_SET = object() - + thread = None + +__all__ = ["Error", "TestFailed", "TestDidNotRun", "ResourceDenied", "import_module", + "verbose", "use_resources", "max_memuse", "record_original_stdout", + "get_original_stdout", "unload", "unlink", "rmtree", "forget", + "is_resource_enabled", "requires", "requires_mac_ver", + "find_unused_port", "bind_port", + "fcmp", "have_unicode", "is_jython", "TESTFN", "HOST", "FUZZ", + "SAVEDCWD", "temp_cwd", "findfile", "sortdict", "check_syntax_error", + "open_urlresource", "check_warnings", "check_py3k_warnings", + "CleanImport", "EnvironmentVarGuard", "captured_output", + "captured_stdout", "TransientResource", "transient_internet", + "run_with_locale", "set_memlimit", "bigmemtest", "bigaddrspacetest", + "BasicTestRunner", "run_unittest", "run_doctest", "threading_setup", + "threading_cleanup", "reap_threads", "start_threads", "cpython_only", + "check_impl_detail", "get_attribute", "py3k_bytes", + "import_fresh_module", "threading_cleanup", "reap_children", + "strip_python_stderr", "IPV6_ENABLED", "run_with_tz", + "SuppressCrashReport"] class Error(Exception): """Base class for regression test exceptions.""" @@ -179,7 +60,7 @@ class ResourceDenied(unittest.SkipTest): """Test skipped because it requested a disallowed resource. This is raised when a test calls requires() for a resource that - has not be enabled. It is used to distinguish between expected + has not been enabled. It is used to distinguish between expected and unexpected skips. """ @@ -188,8 +69,7 @@ def _ignore_deprecated_imports(ignore=True): """Context manager to suppress package and module deprecation warnings when importing them. - If ignore is False, this context manager has no effect. - """ + If ignore is False, this context manager has no effect.""" if ignore: with warnings.catch_warnings(): warnings.filterwarnings("ignore", ".+ (module|package)", @@ -199,45 +79,23 @@ def _ignore_deprecated_imports(ignore=True): yield -def ignore_warnings(*, category): - """Decorator to suppress deprecation warnings. - - Use of context managers to hide warnings make diffs - more noisy and tools like 'git blame' less useful. - """ - def decorator(test): - @functools.wraps(test) - def wrapper(self, *args, **kwargs): - with warnings.catch_warnings(): - warnings.simplefilter('ignore', category=category) - return test(self, *args, **kwargs) - return wrapper - return decorator - - -def import_module(name, deprecated=False, *, required_on=()): +def import_module(name, deprecated=False): """Import and return the module to be tested, raising SkipTest if it is not available. If deprecated is True, any module or package deprecation messages - will be suppressed. If a module is required on a platform but optional for - others, set required_on to an iterable of platform prefixes which will be - compared against sys.platform. - """ + will be suppressed.""" with _ignore_deprecated_imports(deprecated): try: return importlib.import_module(name) - except ImportError as msg: - if sys.platform.startswith(tuple(required_on)): - raise + except ImportError, msg: raise unittest.SkipTest(str(msg)) def _save_and_remove_module(name, orig_modules): """Helper function to save and remove a module from sys.modules - Raise ImportError if the module can't be imported. - """ + Raise ImportError if the module can't be imported.""" # try to import the module and raise an error if it can't be imported if name not in sys.modules: __import__(name) @@ -250,8 +108,7 @@ def _save_and_remove_module(name, orig_modules): def _save_and_block_module(name, orig_modules): """Helper function to save and block a module in sys.modules - Return True if the module was in sys.modules, False otherwise. - """ + Return True if the module was in sys.modules, False otherwise.""" saved = True try: orig_modules[name] = sys.modules[name] @@ -261,63 +118,21 @@ def _save_and_block_module(name, orig_modules): return saved -def anticipate_failure(condition): - """Decorator to mark a test that is known to be broken in some cases - - Any use of this decorator should have a comment identifying the - associated tracker issue. - """ - if condition: - return unittest.expectedFailure - return lambda f: f - -def load_package_tests(pkg_dir, loader, standard_tests, pattern): - """Generic load_tests implementation for simple test packages. - - Most packages can implement load_tests using this function as follows: - - def load_tests(*args): - return load_package_tests(os.path.dirname(__file__), *args) - """ - if pattern is None: - pattern = "test*" - top_dir = os.path.dirname( # Lib - os.path.dirname( # test - os.path.dirname(__file__))) # support - package_tests = loader.discover(start_dir=pkg_dir, - top_level_dir=top_dir, - pattern=pattern) - standard_tests.addTests(package_tests) - return standard_tests - - def import_fresh_module(name, fresh=(), blocked=(), deprecated=False): - """Import and return a module, deliberately bypassing sys.modules. + """Imports and returns a module, deliberately bypassing the sys.modules cache + and importing a fresh copy of the module. Once the import is complete, + the sys.modules cache is restored to its original state. - This function imports and returns a fresh copy of the named Python module - by removing the named module from sys.modules before doing the import. - Note that unlike reload, the original module is not affected by - this operation. + Modules named in fresh are also imported anew if needed by the import. + If one of these modules can't be imported, None is returned. - *fresh* is an iterable of additional module names that are also removed - from the sys.modules cache before doing the import. + Importing of modules named in blocked is prevented while the fresh import + takes place. - *blocked* is an iterable of module names that are replaced with None - in the module cache during the import to ensure that attempts to import - them raise ImportError. - - The named module and any modules named in the *fresh* and *blocked* - parameters are saved before starting the import and then reinserted into - sys.modules when the fresh import is complete. - - Module and package deprecation messages are suppressed during this import - if *deprecated* is True. - - This function will raise ImportError if the named module cannot be - imported. - """ - # NOTE: test_heapq, test_json and test_warnings include extra sanity checks - # to make sure that this utility function is working as expected + If deprecated is True, any module or package deprecation messages + will be suppressed.""" + # NOTE: test_heapq, test_json, and test_warnings include extra sanity + # checks to make sure that this utility function is working as expected with _ignore_deprecated_imports(deprecated): # Keep track of modules saved for later restoration as well # as those which just need a blocking entry removed @@ -346,16 +161,26 @@ def get_attribute(obj, name): try: attribute = getattr(obj, name) except AttributeError: - raise unittest.SkipTest("object %r has no attribute %r" % (obj, name)) + if isinstance(obj, types.ModuleType): + msg = "module %r has no attribute %r" % (obj.__name__, name) + elif isinstance(obj, types.ClassType): + msg = "class %s has no attribute %r" % (obj.__name__, name) + elif isinstance(obj, types.InstanceType): + msg = "%s instance has no attribute %r" % (obj.__class__.__name__, name) + elif isinstance(obj, type): + msg = "type object %r has no attribute %r" % (obj.__name__, name) + else: + msg = "%r object has no attribute %r" % (type(obj).__name__, name) + raise unittest.SkipTest(msg) else: return attribute + verbose = 1 # Flag set to 0 by regrtest.py use_resources = None # Flag set to [] by regrtest.py max_memuse = 0 # Disable bigmem tests (they will still be run with # small sizes, to make sure they work.) real_max_memuse = 0 -junit_xml_list = None # list of testsuite XML elements failfast = False # _original_stdout is meant to hold stdout at the time regrtest began. @@ -378,7 +203,7 @@ def unload(name): def _force_run(path, func, *args): try: return func(*args) - except OSError as err: + except EnvironmentError as err: if verbose >= 2: print('%s: %s' % (err.__class__.__name__, err)) print('re-run %s%r' % (func.__name__, args)) @@ -429,33 +254,13 @@ if sys.platform.startswith("win"): def _rmtree_inner(path): for name in _force_run(path, os.listdir, path): fullname = os.path.join(path, name) - try: - mode = os.lstat(fullname).st_mode - except OSError as exc: - print("support.rmtree(): os.lstat(%r) failed with %s" % (fullname, exc), - file=sys.__stderr__) - mode = 0 - if stat.S_ISDIR(mode): + if os.path.isdir(fullname): _waitfor(_rmtree_inner, fullname, waitall=True) _force_run(fullname, os.rmdir, fullname) else: _force_run(fullname, os.unlink, fullname) _waitfor(_rmtree_inner, path, waitall=True) _waitfor(lambda p: _force_run(p, os.rmdir, p), path) - - def _longpath(path): - try: - import ctypes - except ImportError: - # No ctypes means we can't expands paths. - pass - else: - buffer = ctypes.create_unicode_buffer(len(path) * 2) - length = ctypes.windll.kernel32.GetLongPathNameW(path, buffer, - len(buffer)) - if length: - return buffer[:length] - return path else: _unlink = os.unlink _rmdir = os.rmdir @@ -464,7 +269,7 @@ else: try: shutil.rmtree(path) return - except OSError: + except EnvironmentError: pass def _rmtree_inner(path): @@ -472,7 +277,7 @@ else: fullname = os.path.join(path, name) try: mode = os.lstat(fullname).st_mode - except OSError: + except EnvironmentError: mode = 0 if stat.S_ISDIR(mode): _rmtree_inner(fullname) @@ -482,63 +287,46 @@ else: _rmtree_inner(path) os.rmdir(path) - def _longpath(path): - return path - def unlink(filename): try: _unlink(filename) - except (FileNotFoundError, NotADirectoryError): - pass + except OSError as exc: + if exc.errno not in (errno.ENOENT, errno.ENOTDIR): + raise def rmdir(dirname): try: _rmdir(dirname) - except FileNotFoundError: - pass + except OSError as error: + # The directory need not exist. + if error.errno != errno.ENOENT: + raise def rmtree(path): try: _rmtree(path) - except FileNotFoundError: - pass - -def make_legacy_pyc(source): - """Move a PEP 3147/488 pyc file to its legacy pyc location. - - :param source: The file system path to the source file. The source file - does not need to exist, however the PEP 3147/488 pyc file must exist. - :return: The file system path to the legacy pyc file. - """ - pyc_file = importlib.util.cache_from_source(source) - up_one = os.path.dirname(os.path.abspath(source)) - legacy_pyc = os.path.join(up_one, source + 'c') - os.rename(pyc_file, legacy_pyc) - return legacy_pyc + except OSError, e: + # Unix returns ENOENT, Windows returns ESRCH. + if e.errno not in (errno.ENOENT, errno.ESRCH): + raise def forget(modname): - """'Forget' a module was ever imported. - - This removes the module from sys.modules and deletes any PEP 3147/488 or - legacy .pyc files. - """ + '''"Forget" a module was ever imported by removing it from sys.modules and + deleting any .pyc and .pyo files.''' unload(modname) for dirname in sys.path: - source = os.path.join(dirname, modname + '.py') - # It doesn't matter if they exist or not, unlink all possible - # combinations of PEP 3147/488 and legacy pyc files. - unlink(source + 'c') - for opt in ('', 1, 2): - unlink(importlib.util.cache_from_source(source, optimization=opt)) + unlink(os.path.join(dirname, modname + os.extsep + 'pyc')) + # Deleting the .pyo file cannot be within the 'try' for the .pyc since + # the chance exists that there is no .pyc (and thus the 'try' statement + # is exited) but there is a .pyo file. + unlink(os.path.join(dirname, modname + os.extsep + 'pyo')) # Check whether a gui is actually available def _is_gui_available(): if hasattr(_is_gui_available, 'result'): return _is_gui_available.result reason = None - if sys.platform.startswith('win') and platform.win32_is_iot(): - reason = "gui is not available on Windows IoT Core" - elif sys.platform.startswith('win'): + if sys.platform.startswith('win'): # if Python is running as a service (such as the buildbot service), # gui interaction may be disallowed import ctypes @@ -591,7 +379,7 @@ def _is_gui_available(): # check on every platform whether tkinter can actually do anything if not reason: try: - from tkinter import Tk + from Tkinter import Tk root = Tk() root.withdraw() root.update() @@ -620,56 +408,11 @@ def requires(resource, msg=None): """Raise ResourceDenied if the specified resource is not available.""" if not is_resource_enabled(resource): if msg is None: - msg = "Use of the %r resource not enabled" % resource + msg = "Use of the `%s' resource not enabled" % resource raise ResourceDenied(msg) if resource == 'gui' and not _is_gui_available(): raise ResourceDenied(_is_gui_available.reason) -def _requires_unix_version(sysname, min_version): - """Decorator raising SkipTest if the OS is `sysname` and the version is less - than `min_version`. - - For example, @_requires_unix_version('FreeBSD', (7, 2)) raises SkipTest if - the FreeBSD version is less than 7.2. - """ - def decorator(func): - @functools.wraps(func) - def wrapper(*args, **kw): - if platform.system() == sysname: - version_txt = platform.release().split('-', 1)[0] - try: - version = tuple(map(int, version_txt.split('.'))) - except ValueError: - pass - else: - if version < min_version: - min_version_txt = '.'.join(map(str, min_version)) - raise unittest.SkipTest( - "%s version %s or higher required, not %s" - % (sysname, min_version_txt, version_txt)) - return func(*args, **kw) - wrapper.min_version = min_version - return wrapper - return decorator - -def requires_freebsd_version(*min_version): - """Decorator raising SkipTest if the OS is FreeBSD and the FreeBSD version is - less than `min_version`. - - For example, @requires_freebsd_version(7, 2) raises SkipTest if the FreeBSD - version is less than 7.2. - """ - return _requires_unix_version('FreeBSD', min_version) - -def requires_linux_version(*min_version): - """Decorator raising SkipTest if the OS is Linux and the Linux version is - less than `min_version`. - - For example, @requires_linux_version(2, 6, 32) raises SkipTest if the Linux - version is less than 2.6.32. - """ - return _requires_unix_version('Linux', min_version) - def requires_mac_ver(*min_version): """Decorator raising SkipTest if the OS is Mac OS X and the OS X version if less than min_version. @@ -698,38 +441,9 @@ def requires_mac_ver(*min_version): return decorator -def requires_hashdigest(digestname, openssl=None, usedforsecurity=True): - """Decorator raising SkipTest if a hashing algorithm is not available - - The hashing algorithm could be missing or blocked by a strict crypto - policy. - - 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. - - ValueError: [digital envelope routines: EVP_DigestInit_ex] disabled for FIPS - ValueError: unsupported hash type md4 - """ - def decorator(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(*args, **kwargs) - return wrapper - return decorator - - -HOST = "localhost" -HOSTv4 = "127.0.0.1" +# Don't use "localhost", since resolving it uses the DNS under recent +# Windows versions (see issue #18792). +HOST = "127.0.0.1" HOSTv6 = "::1" @@ -760,7 +474,7 @@ def find_unused_port(family=socket.AF_INET, socktype=socket.SOCK_STREAM): the SO_REUSEADDR socket option having different semantics on Windows versus Unix/Linux. On Unix, you can't have two AF_INET SOCK_STREAM sockets bind, listen and then accept connections on identical host/ports. An EADDRINUSE - OSError will be raised at some point (depending on the platform and + socket.error will be raised at some point (depending on the platform and the order bind and listen were called on each socket). However, on Windows, if SO_REUSEADDR is set on the sockets, no EADDRINUSE @@ -786,11 +500,10 @@ def find_unused_port(family=socket.AF_INET, socktype=socket.SOCK_STREAM): port returned to us by the OS won't immediately be dished back out to some other process when we close and delete our temporary socket but before our calling code has a chance to bind the returned port. We can deal with this - issue if/when we come across it. - """ - - with socket.socket(family, socktype) as tempsock: - port = bind_port(tempsock) + issue if/when we come across it.""" + tempsock = socket.socket(family, socktype) + port = bind_port(tempsock) + tempsock.close() del tempsock return port @@ -808,7 +521,6 @@ def bind_port(sock, host=HOST): on Windows), it will be set on the socket. This will prevent anyone else from bind()'ing to our host/port for the duration of the test. """ - if sock.family == socket.AF_INET and sock.type == socket.SOCK_STREAM: if hasattr(socket, 'SO_REUSEADDR'): if sock.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR) == 1: @@ -819,7 +531,7 @@ def bind_port(sock, host=HOST): if sock.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT) == 1: raise TestFailed("tests should never set the SO_REUSEPORT " \ "socket option on TCP/IP sockets!") - except OSError: + except EnvironmentError: # Python's socket module was compiled using modern headers # thus defining SO_REUSEPORT but this process is running # under an older kernel that does not support SO_REUSEPORT. @@ -831,15 +543,6 @@ def bind_port(sock, host=HOST): port = sock.getsockname()[1] return port -def bind_unix_socket(sock, addr): - """Bind a unix socket, raising SkipTest if PermissionError is raised.""" - assert sock.family == socket.AF_UNIX - try: - sock.bind(addr) - except PermissionError: - sock.close() - raise unittest.SkipTest('cannot bind AF_UNIX sockets') - def _is_ipv6_enabled(): """Check whether IPv6 is enabled on this host.""" if socket.has_ipv6: @@ -848,7 +551,7 @@ def _is_ipv6_enabled(): sock = socket.socket(socket.AF_INET6, socket.SOCK_STREAM) sock.bind((HOSTv6, 0)) return True - except OSError: + except socket.error: pass finally: if sock: @@ -863,13 +566,32 @@ def system_must_validate_cert(f): def dec(*args, **kwargs): try: f(*args, **kwargs) - except OSError as e: + except IOError as e: if "CERTIFICATE_VERIFY_FAILED" in str(e): raise unittest.SkipTest("system does not contain " "necessary certificates") raise return dec +FUZZ = 1e-6 + +def fcmp(x, y): # fuzzy comparison function + if isinstance(x, float) or isinstance(y, float): + try: + fuzz = (abs(x) + abs(y)) * FUZZ + if abs(x-y) <= fuzz: + return 0 + except: + pass + elif type(x) == type(y) and isinstance(x, (tuple, list)): + for i in range(min(len(x), len(y))): + outcome = fcmp(x[i], y[i]) + if outcome != 0: + return outcome + return (len(x) > len(y)) - (len(x) < len(y)) + return (x > y) - (x < y) + + # A constant likely larger than the underlying OS pipe buffer size, to # make writes blocking. # Windows limit seems to be around 512 B, and many Unix kernels have a @@ -884,34 +606,115 @@ PIPE_MAX_SIZE = 4 * 1024 * 1024 + 1 # for a discussion of this number). SOCK_MAX_SIZE = 16 * 1024 * 1024 + 1 -# decorator for skipping tests on non-IEEE 754 platforms -requires_IEEE_754 = unittest.skipUnless( - float.__getformat__("double").startswith("IEEE"), - "test requires IEEE 754 doubles") - -requires_zlib = unittest.skipUnless(zlib, 'requires zlib') - -requires_gzip = unittest.skipUnless(gzip, 'requires gzip') - -requires_bz2 = unittest.skipUnless(bz2, 'requires bz2') +is_jython = sys.platform.startswith('java') -requires_lzma = unittest.skipUnless(lzma, 'requires lzma') +try: + unicode + have_unicode = True +except NameError: + have_unicode = False -is_jython = sys.platform.startswith('java') +requires_unicode = unittest.skipUnless(have_unicode, 'no unicode support') -is_android = hasattr(sys, 'getandroidapilevel') +def u(s): + return unicode(s, 'unicode-escape') -if sys.platform != 'win32': - unix_shell = '/system/bin/sh' if is_android else '/bin/sh' -else: - unix_shell = None +# FS_NONASCII: non-ASCII Unicode character encodable by +# sys.getfilesystemencoding(), or None if there is no such character. +FS_NONASCII = None +if have_unicode: + for character in ( + # First try printable and common characters to have a readable filename. + # For each character, the encoding list are just example of encodings able + # to encode the character (the list is not exhaustive). + + # U+00E6 (Latin Small Letter Ae): cp1252, iso-8859-1 + unichr(0x00E6), + # U+0130 (Latin Capital Letter I With Dot Above): cp1254, iso8859_3 + unichr(0x0130), + # U+0141 (Latin Capital Letter L With Stroke): cp1250, cp1257 + unichr(0x0141), + # U+03C6 (Greek Small Letter Phi): cp1253 + unichr(0x03C6), + # U+041A (Cyrillic Capital Letter Ka): cp1251 + unichr(0x041A), + # U+05D0 (Hebrew Letter Alef): Encodable to cp424 + unichr(0x05D0), + # U+060C (Arabic Comma): cp864, cp1006, iso8859_6, mac_arabic + unichr(0x060C), + # U+062A (Arabic Letter Teh): cp720 + unichr(0x062A), + # U+0E01 (Thai Character Ko Kai): cp874 + unichr(0x0E01), + + # Then try more "special" characters. "special" because they may be + # interpreted or displayed differently depending on the exact locale + # encoding and the font. + + # U+00A0 (No-Break Space) + unichr(0x00A0), + # U+20AC (Euro Sign) + unichr(0x20AC), + ): + try: + # In Windows, 'mbcs' is used, and encode() returns '?' + # for characters missing in the ANSI codepage + if character.encode(sys.getfilesystemencoding())\ + .decode(sys.getfilesystemencoding())\ + != character: + raise UnicodeError + except UnicodeError: + pass + else: + FS_NONASCII = character + break # Filename used for testing if os.name == 'java': # Jython disallows @ in module names TESTFN = '$test' +elif os.name == 'riscos': + TESTFN = 'testfile' else: TESTFN = '@test' + # Unicode name only used if TEST_FN_ENCODING exists for the platform. + if have_unicode: + # Assuming sys.getfilesystemencoding()!=sys.getdefaultencoding() + # TESTFN_UNICODE is a filename that can be encoded using the + # file system encoding, but *not* with the default (ascii) encoding + if isinstance('', unicode): + # python -U + # XXX perhaps unicode() should accept Unicode strings? + TESTFN_UNICODE = "@test-\xe0\xf2" + else: + # 2 latin characters. + TESTFN_UNICODE = unicode("@test-\xe0\xf2", "latin-1") + TESTFN_ENCODING = sys.getfilesystemencoding() + # TESTFN_UNENCODABLE is a filename that should *not* be + # able to be encoded by *either* the default or filesystem encoding. + # This test really only makes sense on Windows NT platforms + # which have special Unicode support in posixmodule. + if (not hasattr(sys, "getwindowsversion") or + sys.getwindowsversion()[3] < 2): # 0=win32s or 1=9x/ME + TESTFN_UNENCODABLE = None + else: + # Japanese characters (I think - from bug 846133) + TESTFN_UNENCODABLE = eval('u"@test-\u5171\u6709\u3055\u308c\u308b"') + try: + # XXX - Note - should be using TESTFN_ENCODING here - but for + # Windows, "mbcs" currently always operates as if in + # errors=ignore' mode - hence we get '?' characters rather than + # the exception. 'Latin1' operates as we expect - ie, fails. + # See [ 850997 ] mbcs encoding ignores errors + TESTFN_UNENCODABLE.encode("Latin1") + except UnicodeEncodeError: + pass + else: + print \ + 'WARNING: The filename %r CAN be encoded by the filesystem. ' \ + 'Unicode filename tests may not be effective' \ + % TESTFN_UNENCODABLE + # Disambiguate TESTFN for parallel testing, while letting it remain a valid # module name. @@ -921,143 +724,9 @@ TESTFN = "{}_{}_tmp".format(TESTFN, os.getpid()) # The URL must use clear-text HTTP: no redirection to encrypted HTTPS. TEST_HTTP_URL = "http://www.pythontest.net" -# FS_NONASCII: non-ASCII character encodable by os.fsencode(), -# or None if there is no such character. -FS_NONASCII = None -for character in ( - # First try printable and common characters to have a readable filename. - # For each character, the encoding list are just example of encodings able - # to encode the character (the list is not exhaustive). - - # U+00E6 (Latin Small Letter Ae): cp1252, iso-8859-1 - '\u00E6', - # U+0130 (Latin Capital Letter I With Dot Above): cp1254, iso8859_3 - '\u0130', - # U+0141 (Latin Capital Letter L With Stroke): cp1250, cp1257 - '\u0141', - # U+03C6 (Greek Small Letter Phi): cp1253 - '\u03C6', - # U+041A (Cyrillic Capital Letter Ka): cp1251 - '\u041A', - # U+05D0 (Hebrew Letter Alef): Encodable to cp424 - '\u05D0', - # U+060C (Arabic Comma): cp864, cp1006, iso8859_6, mac_arabic - '\u060C', - # U+062A (Arabic Letter Teh): cp720 - '\u062A', - # U+0E01 (Thai Character Ko Kai): cp874 - '\u0E01', - - # Then try more "special" characters. "special" because they may be - # interpreted or displayed differently depending on the exact locale - # encoding and the font. - - # U+00A0 (No-Break Space) - '\u00A0', - # U+20AC (Euro Sign) - '\u20AC', -): - try: - # If Python is set up to use the legacy 'mbcs' in Windows, - # 'replace' error mode is used, and encode() returns b'?' - # for characters missing in the ANSI codepage - if os.fsdecode(os.fsencode(character)) != character: - raise UnicodeError - except UnicodeError: - pass - else: - FS_NONASCII = character - break - -# TESTFN_UNICODE is a non-ascii filename -TESTFN_UNICODE = TESTFN + "-\xe0\xf2\u0258\u0141\u011f" -if sys.platform == 'darwin': - # In Mac OS X's VFS API file names are, by definition, canonically - # decomposed Unicode, encoded using UTF-8. See QA1173: - # http://developer.apple.com/mac/library/qa/qa2001/qa1173.html - import unicodedata - TESTFN_UNICODE = unicodedata.normalize('NFD', TESTFN_UNICODE) -TESTFN_ENCODING = sys.getfilesystemencoding() - -# TESTFN_UNENCODABLE is a filename (str type) that should *not* be able to be -# encoded by the filesystem encoding (in strict mode). It can be None if we -# cannot generate such filename. -TESTFN_UNENCODABLE = None -if os.name == 'nt': - # skip win32s (0) or Windows 9x/ME (1) - if sys.getwindowsversion().platform >= 2: - # Different kinds of characters from various languages to minimize the - # probability that the whole name is encodable to MBCS (issue #9819) - TESTFN_UNENCODABLE = TESTFN + "-\u5171\u0141\u2661\u0363\uDC80" - try: - TESTFN_UNENCODABLE.encode(TESTFN_ENCODING) - except UnicodeEncodeError: - pass - else: - print('WARNING: The filename %r CAN be encoded by the filesystem encoding (%s). ' - 'Unicode filename tests may not be effective' - % (TESTFN_UNENCODABLE, TESTFN_ENCODING)) - TESTFN_UNENCODABLE = None -# Mac OS X denies unencodable filenames (invalid utf-8) -elif sys.platform != 'darwin': - try: - # ascii and utf-8 cannot encode the byte 0xff - b'\xff'.decode(TESTFN_ENCODING) - except UnicodeDecodeError: - # 0xff will be encoded using the surrogate character u+DCFF - TESTFN_UNENCODABLE = TESTFN \ - + b'-\xff'.decode(TESTFN_ENCODING, 'surrogateescape') - else: - # File system encoding (eg. ISO-8859-* encodings) can encode - # the byte 0xff. Skip some unicode filename tests. - pass - -# TESTFN_UNDECODABLE is a filename (bytes type) that should *not* be able to be -# decoded from the filesystem encoding (in strict mode). It can be None if we -# cannot generate such filename (ex: the latin1 encoding can decode any byte -# sequence). On UNIX, TESTFN_UNDECODABLE can be decoded by os.fsdecode() thanks -# to the surrogateescape error handler (PEP 383), but not from the filesystem -# encoding in strict mode. -TESTFN_UNDECODABLE = None -for name in ( - # b'\xff' is not decodable by os.fsdecode() with code page 932. Windows - # accepts it to create a file or a directory, or don't accept to enter to - # such directory (when the bytes name is used). So test b'\xe7' first: it is - # not decodable from cp932. - b'\xe7w\xf0', - # undecodable from ASCII, UTF-8 - b'\xff', - # undecodable from iso8859-3, iso8859-6, iso8859-7, cp424, iso8859-8, cp856 - # and cp857 - b'\xae\xd5' - # undecodable from UTF-8 (UNIX and Mac OS X) - b'\xed\xb2\x80', b'\xed\xb4\x80', - # undecodable from shift_jis, cp869, cp874, cp932, cp1250, cp1251, cp1252, - # cp1253, cp1254, cp1255, cp1257, cp1258 - b'\x81\x98', -): - try: - name.decode(TESTFN_ENCODING) - except UnicodeDecodeError: - TESTFN_UNDECODABLE = os.fsencode(TESTFN) + name - break - -if FS_NONASCII: - TESTFN_NONASCII = TESTFN + '-' + FS_NONASCII -else: - TESTFN_NONASCII = None - # Save the initial cwd SAVEDCWD = os.getcwd() -# Set by libregrtest/main.py so we can skip tests that are not -# useful for PGO -PGO = False - -# Set by libregrtest/main.py if we are running the extended (time consuming) -# PGO task. If this is True, PGO is also True. -PGO_EXTENDED = False - @contextlib.contextmanager def temp_dir(path=None, quiet=False): """Return a context manager that creates a temporary directory. @@ -1074,18 +743,26 @@ def temp_dir(path=None, quiet=False): """ dir_created = False if path is None: + import tempfile path = tempfile.mkdtemp() dir_created = True path = os.path.realpath(path) else: + if (have_unicode and isinstance(path, unicode) and + not os.path.supports_unicode_filenames): + try: + path = path.encode(sys.getfilesystemencoding() or 'ascii') + except UnicodeEncodeError: + if not quiet: + raise unittest.SkipTest('unable to encode the cwd name with ' + 'the filesystem encoding.') try: os.mkdir(path) dir_created = True - except OSError as exc: + except OSError: if not quiet: raise - warnings.warn(f'tests may fail, unable to create ' - f'temporary directory {path!r}: {exc}', + warnings.warn('tests may fail, unable to create temp dir: ' + path, RuntimeWarning, stacklevel=3) if dir_created: pid = os.getpid() @@ -1093,7 +770,7 @@ def temp_dir(path=None, quiet=False): yield path finally: # In case the process forks, let only the parent remove the - # directory. The child has a different process id. (bpo-30028) + # directory. The child has a diffent process id. (bpo-30028) if dir_created and pid == os.getpid(): rmtree(path) @@ -1112,12 +789,11 @@ def change_cwd(path, quiet=False): """ saved_dir = os.getcwd() try: - os.chdir(os.path.realpath(path)) - except OSError as exc: + os.chdir(path) + except OSError: if not quiet: raise - warnings.warn(f'tests may fail, unable to change the current working ' - f'directory to {path!r}: {exc}', + warnings.warn('tests may fail, unable to change CWD to: ' + path, RuntimeWarning, stacklevel=3) try: yield os.getcwd() @@ -1144,16 +820,6 @@ def temp_cwd(name='tempcwd', quiet=False): with change_cwd(temp_path, quiet=quiet) as cwd_dir: yield cwd_dir -if hasattr(os, "umask"): - @contextlib.contextmanager - def temp_umask(umask): - """Context manager that temporarily sets the process umask.""" - oldmask = os.umask(umask) - try: - yield - finally: - os.umask(oldmask) - # TEST_HOME_DIR refers to the top level directory of the "test" package # that contains Python's regression test suite TEST_SUPPORT_DIR = os.path.dirname(os.path.abspath(__file__)) @@ -1162,32 +828,24 @@ TEST_HOME_DIR = os.path.dirname(TEST_SUPPORT_DIR) # TEST_DATA_DIR is used as a target download location for remote resources TEST_DATA_DIR = os.path.join(TEST_HOME_DIR, "data") -def findfile(filename, subdir=None): - """Try to find a file on sys.path or in the test directory. If it is not +def findfile(file, subdir=None): + """Try to find a file on sys.path and the working directory. If it is not found the argument passed to the function is returned (this does not - necessarily signal failure; could still be the legitimate path). - - Setting *subdir* indicates a relative path to use to find the file - rather than looking directly in the path directories. - """ - if os.path.isabs(filename): - return filename + necessarily signal failure; could still be the legitimate path).""" + if os.path.isabs(file): + return file if subdir is not None: - filename = os.path.join(subdir, filename) + file = os.path.join(subdir, file) path = [TEST_HOME_DIR] + sys.path for dn in path: - fn = os.path.join(dn, filename) + fn = os.path.join(dn, file) if os.path.exists(fn): return fn - return filename - -def create_empty_file(filename): - """Create an empty file. If the file already exists, truncate it.""" - fd = os.open(filename, os.O_WRONLY | os.O_CREAT | os.O_TRUNC) - os.close(fd) + return file def sortdict(dict): "Like repr(dict), but in sorted order." - items = sorted(dict.items()) + items = dict.items() + items.sort() reprpairs = ["%r: %r" % pair for pair in items] withcommas = ", ".join(reprpairs) return "{%s}" % withcommas @@ -1204,56 +862,24 @@ def make_bad_fd(): file.close() unlink(TESTFN) - -def check_syntax_error(testcase, statement, errtext='', *, lineno=None, offset=None): - with testcase.assertRaisesRegex(SyntaxError, errtext) as cm: +def check_syntax_error(testcase, statement, errtext='', lineno=None, offset=None): + with testcase.assertRaisesRegexp(SyntaxError, errtext) as cm: compile(statement, '<test string>', 'exec') err = cm.exception - testcase.assertIsNotNone(err.lineno) if lineno is not None: testcase.assertEqual(err.lineno, lineno) - testcase.assertIsNotNone(err.offset) if offset is not None: testcase.assertEqual(err.offset, offset) -def check_syntax_warning(testcase, statement, errtext='', *, lineno=1, offset=None): - # Test also that a warning is emitted only once. - with warnings.catch_warnings(record=True) as warns: - warnings.simplefilter('always', SyntaxWarning) - compile(statement, '<testcase>', 'exec') - testcase.assertEqual(len(warns), 1, warns) - - warn, = warns - testcase.assertTrue(issubclass(warn.category, SyntaxWarning), warn.category) - if errtext: - testcase.assertRegex(str(warn.message), errtext) - testcase.assertEqual(warn.filename, '<testcase>') - testcase.assertIsNotNone(warn.lineno) - if lineno is not None: - testcase.assertEqual(warn.lineno, lineno) - - # SyntaxWarning should be converted to SyntaxError when raised, - # since the latter contains more information and provides better - # error report. - with warnings.catch_warnings(record=True) as warns: - warnings.simplefilter('error', SyntaxWarning) - check_syntax_error(testcase, statement, errtext, - lineno=lineno, offset=offset) - # No warnings are leaked when a SyntaxError is raised. - testcase.assertEqual(warns, []) - - -def open_urlresource(url, *args, **kw): - import urllib.request, urllib.parse +def open_urlresource(url, check=None): + import urlparse, urllib2 - check = kw.pop('check', None) - - filename = urllib.parse.urlparse(url)[2].split('/')[-1] # '/': it's URL! + filename = urlparse.urlparse(url)[2].split('/')[-1] # '/': it's URL! fn = os.path.join(TEST_DATA_DIR, filename) def check_valid_file(fn): - f = open(fn, *args, **kw) + f = open(fn) if check is None: return f elif check(f): @@ -1270,14 +896,8 @@ def open_urlresource(url, *args, **kw): # Verify the requirement before downloading the file requires('urlfetch') - if verbose: - print('\tfetching %s ...' % url, file=get_original_stdout()) - opener = urllib.request.build_opener() - if gzip: - opener.addheaders.append(('Accept-Encoding', 'gzip')) - f = opener.open(url, timeout=INTERNET_TIMEOUT) - if gzip and f.headers.get('Content-Encoding') == 'gzip': - f = gzip.GzipFile(fileobj=f) + print >> get_original_stdout(), '\tfetching %s ...' % url + f = urllib2.urlopen(url, timeout=15) try: with open(fn, "wb") as out: s = f.read() @@ -1290,7 +910,7 @@ def open_urlresource(url, *args, **kw): f = check_valid_file(fn) if f is not None: return f - raise TestFailed('invalid resource %r' % fn) + raise TestFailed('invalid resource "%s"' % fn) class WarningsRecorder(object): @@ -1334,22 +954,22 @@ def _filterwarnings(filters, quiet=False): sys.modules['warnings'].simplefilter("always") yield WarningsRecorder(w) # Filter the recorded warnings - reraise = list(w) + reraise = [warning.message for warning in w] missing = [] for msg, cat in filters: seen = False - for w in reraise[:]: - warning = w.message + for exc in reraise[:]: + message = str(exc) # Filter out the matching messages - if (re.match(msg, str(warning), re.I) and - issubclass(warning.__class__, cat)): + if (re.match(msg, message, re.I) and + issubclass(exc.__class__, cat)): seen = True - reraise.remove(w) + reraise.remove(exc) if not seen and not quiet: # This filter caught nothing missing.append((msg, cat.__name__)) if reraise: - raise AssertionError("unhandled warning %s" % reraise[0]) + raise AssertionError("unhandled warning %r" % reraise[0]) if missing: raise AssertionError("filter (%r, %s) did not catch any warning" % missing[0]) @@ -1380,45 +1000,26 @@ def check_warnings(*filters, **kwargs): @contextlib.contextmanager -def check_no_warnings(testcase, message='', category=Warning, force_gc=False): - """Context manager to check that no warnings are emitted. - - This context manager enables a given warning within its scope - and checks that no warnings are emitted even with that warning - enabled. - - If force_gc is True, a garbage collection is attempted before checking - for warnings. This may help to catch warnings emitted when objects - are deleted, such as ResourceWarning. +def check_py3k_warnings(*filters, **kwargs): + """Context manager to silence py3k warnings. - Other keyword arguments are passed to warnings.filterwarnings(). - """ - with warnings.catch_warnings(record=True) as warns: - warnings.filterwarnings('always', - message=message, - category=category) - yield - if force_gc: - gc_collect() - testcase.assertEqual(warns, []) - - -@contextlib.contextmanager -def check_no_resource_warning(testcase): - """Context manager to check that no ResourceWarning is emitted. - - Usage: + Accept 2-tuples as positional arguments: + ("message regexp", WarningCategory) - with check_no_resource_warning(self): - f = open(...) - ... - del f + Optional argument: + - if 'quiet' is True, it does not fail if a filter catches nothing + (default False) - You must remove the object which may emit ResourceWarning before - the end of the context manager. + Without argument, it defaults to: + check_py3k_warnings(("", DeprecationWarning), quiet=False) """ - with check_no_warnings(testcase, category=ResourceWarning, force_gc=True): - yield + if sys.py3kwarning: + if not filters: + filters = (("", DeprecationWarning),) + else: + # It should not raise any py3k warning + filters = () + return _filterwarnings(filters, kwargs.get('quiet')) class CleanImport(object): @@ -1453,7 +1054,7 @@ class CleanImport(object): sys.modules.update(self.original_modules) -class EnvironmentVarGuard(collections.abc.MutableMapping): +class EnvironmentVarGuard(UserDict.DictMixin): """Class to help protect the environment variable properly. Can be used as a context manager.""" @@ -1481,12 +1082,6 @@ class EnvironmentVarGuard(collections.abc.MutableMapping): def keys(self): return self._environ.keys() - def __iter__(self): - return iter(self._environ) - - def __len__(self): - return len(self._environ) - def set(self, envvar, value): self[envvar] = value @@ -1548,7 +1143,7 @@ class TransientResource(object): self.attrs, raise ResourceDenied. Otherwise let the exception propagate (if any).""" if type_ is not None and issubclass(self.exc, type_): - for attr, attr_value in self.attrs.items(): + for attr, attr_value in self.attrs.iteritems(): if not hasattr(value, attr): break if getattr(value, attr) != attr_value: @@ -1556,42 +1151,11 @@ class TransientResource(object): else: raise ResourceDenied("an optional resource is not available") -# Context managers that raise ResourceDenied when various issues -# with the Internet connection manifest themselves as exceptions. -# XXX deprecate these and use transient_internet() instead -time_out = TransientResource(OSError, errno=errno.ETIMEDOUT) -socket_peer_reset = TransientResource(OSError, errno=errno.ECONNRESET) -ioerror_peer_reset = TransientResource(OSError, errno=errno.ECONNRESET) - - -def get_socket_conn_refused_errs(): - """ - Get the different socket error numbers ('errno') which can be received - when a connection is refused. - """ - errors = [errno.ECONNREFUSED] - if hasattr(errno, 'ENETUNREACH'): - # On Solaris, ENETUNREACH is returned sometimes instead of ECONNREFUSED - errors.append(errno.ENETUNREACH) - if hasattr(errno, 'EADDRNOTAVAIL'): - # bpo-31910: socket.create_connection() fails randomly - # with EADDRNOTAVAIL on Travis CI - errors.append(errno.EADDRNOTAVAIL) - if hasattr(errno, 'EHOSTUNREACH'): - # bpo-37583: The destination host cannot be reached - errors.append(errno.EHOSTUNREACH) - if not IPV6_ENABLED: - errors.append(errno.EAFNOSUPPORT) - return errors - @contextlib.contextmanager -def transient_internet(resource_name, *, timeout=_NOT_SET, errnos=()): +def transient_internet(resource_name, timeout=30.0, errnos=()): """Return a context manager that raises ResourceDenied when various issues with the Internet connection manifest themselves as exceptions.""" - if timeout is _NOT_SET: - timeout = INTERNET_TIMEOUT - default_errnos = [ ('ECONNREFUSED', 111), ('ECONNRESET', 104), @@ -1607,11 +1171,12 @@ def transient_internet(resource_name, *, timeout=_NOT_SET, errnos=()): ('EAI_FAIL', -4), ('EAI_NONAME', -2), ('EAI_NODATA', -5), - # Encountered when trying to resolve IPv6-only hostnames + # Windows defines EAI_NODATA as 11001 but idiotic getaddrinfo() + # implementation actually returns WSANO_DATA i.e. 11004. ('WSANO_DATA', 11004), ] - denied = ResourceDenied("Resource %r is not available" % resource_name) + denied = ResourceDenied("Resource '%s' is not available" % resource_name) captured_errnos = errnos gai_errnos = [] if not captured_errnos: @@ -1624,37 +1189,27 @@ def transient_internet(resource_name, *, timeout=_NOT_SET, errnos=()): n = getattr(err, 'errno', None) if (isinstance(err, socket.timeout) or (isinstance(err, socket.gaierror) and n in gai_errnos) or - (isinstance(err, urllib.error.HTTPError) and - 500 <= err.code <= 599) or - (isinstance(err, urllib.error.URLError) and - (("ConnectionRefusedError" in err.reason) or - ("TimeoutError" in err.reason) or - ("EOFError" in err.reason))) or n in captured_errnos): if not verbose: sys.stderr.write(denied.args[0] + "\n") - raise denied from err + raise denied old_timeout = socket.getdefaulttimeout() try: if timeout is not None: socket.setdefaulttimeout(timeout) yield - except nntplib.NNTPTemporaryError as err: - if verbose: - sys.stderr.write(denied.args[0] + "\n") - raise denied from err - except OSError as err: + except IOError as err: # urllib can wrap original socket errors multiple times (!), we must # unwrap to get at the original error. while True: a = err.args - if len(a) >= 1 and isinstance(a[0], OSError): + if len(a) >= 1 and isinstance(a[0], IOError): err = a[0] # The error can also be wrapped as args[1]: # except socket.error as msg: - # raise OSError('socket error', msg).with_traceback(sys.exc_info()[2]) - elif len(a) >= 2 and isinstance(a[1], OSError): + # raise IOError('socket error', msg).with_traceback(sys.exc_info()[2]) + elif len(a) >= 2 and isinstance(a[1], IOError): err = a[1] else: break @@ -1668,11 +1223,11 @@ def transient_internet(resource_name, *, timeout=_NOT_SET, errnos=()): @contextlib.contextmanager def captured_output(stream_name): - """Return a context manager used by captured_stdout/stdin/stderr + """Return a context manager used by captured_stdout and captured_stdin that temporarily replaces the sys stream *stream_name* with a StringIO.""" - import io + import StringIO orig_stdout = getattr(sys, stream_name) - setattr(sys, stream_name, io.StringIO()) + setattr(sys, stream_name, StringIO.StringIO()) try: yield getattr(sys, stream_name) finally: @@ -1681,34 +1236,18 @@ def captured_output(stream_name): def captured_stdout(): """Capture the output of sys.stdout: - with captured_stdout() as stdout: - print("hello") - self.assertEqual(stdout.getvalue(), "hello\\n") + with captured_stdout() as s: + print "hello" + self.assertEqual(s.getvalue(), "hello") """ return captured_output("stdout") def captured_stderr(): - """Capture the output of sys.stderr: - - with captured_stderr() as stderr: - print("hello", file=sys.stderr) - self.assertEqual(stderr.getvalue(), "hello\\n") - """ return captured_output("stderr") def captured_stdin(): - """Capture the input to sys.stdin: - - with captured_stdin() as stdin: - stdin.write('hello\\n') - stdin.seek(0) - # call test code that consumes from sys.stdin - captured = input() - self.assertEqual(captured, "hello") - """ return captured_output("stdin") - def gc_collect(): """Force as many objects as possible to be collected. @@ -1725,39 +1264,17 @@ def gc_collect(): gc.collect() gc.collect() -@contextlib.contextmanager -def disable_gc(): - have_gc = gc.isenabled() - gc.disable() - try: - yield - finally: - if have_gc: - gc.enable() - - -def python_is_optimized(): - """Find if Python was built with optimizations.""" - cflags = sysconfig.get_config_var('PY_CFLAGS') or '' - final_opt = "" - for opt in cflags.split(): - if opt.startswith('-O'): - final_opt = opt - return final_opt not in ('', '-O0', '-Og') - -_header = 'nP' -_align = '0n' -if hasattr(sys, "getobjects"): +_header = '2P' +if hasattr(sys, "gettotalrefcount"): _header = '2P' + _header - _align = '0P' -_vheader = _header + 'n' +_vheader = _header + 'P' def calcobjsize(fmt): - return struct.calcsize(_header + fmt + _align) + return struct.calcsize(_header + fmt + '0P') def calcvobjsize(fmt): - return struct.calcsize(_vheader + fmt + _align) + return struct.calcsize(_vheader + fmt + '0P') _TPFLAGS_HAVE_GC = 1<<14 @@ -1774,6 +1291,7 @@ def check_sizeof(test, o, size): % (type(o), result, size) test.assertEqual(result, size, msg) + #======================================================================= # Decorator for running a function in a different locale, correctly resetting # it afterwards. @@ -1805,7 +1323,7 @@ def run_with_locale(catstr, *locales): finally: if locale and orig_locale: locale.setlocale(category, orig_locale) - inner.__name__ = func.__name__ + inner.func_name = func.func_name inner.__doc__ = func.__doc__ return inner return decorator @@ -1844,8 +1362,7 @@ def run_with_tz(tz): return decorator #======================================================================= -# Big-memory-test support. Separate from 'resources' because memory use -# should be configurable. +# Big-memory-test support. Separate from 'resources' because memory use should be configurable. # Some handy shorthands. Note that these are used for byte-limits as well # as size-limits, in the various bigmem tests @@ -1877,54 +1394,48 @@ def set_memlimit(limit): raise ValueError('Memory limit %r too low to be useful' % (limit,)) max_memuse = memlimit -class _MemoryWatchdog: - """An object which periodically watches the process' memory consumption - and prints it out. - """ - - def __init__(self): - self.procfile = '/proc/{pid}/statm'.format(pid=os.getpid()) - self.started = False - - def start(self): - try: - f = open(self.procfile, 'r') - except OSError as e: - warnings.warn('/proc not available for stats: {}'.format(e), - RuntimeWarning) - sys.stderr.flush() - return - - with f: - watchdog_script = findfile("memory_watchdog.py") - self.mem_watchdog = subprocess.Popen([sys.executable, watchdog_script], - stdin=f, - stderr=subprocess.DEVNULL) - self.started = True - - def stop(self): - if self.started: - self.mem_watchdog.terminate() - self.mem_watchdog.wait() - - -def bigmemtest(size, memuse, dry_run=True): +def bigmemtest(minsize, memuse, overhead=5*_1M): """Decorator for bigmem tests. - 'size' is a requested size for the test (in arbitrary, test-interpreted - units.) 'memuse' is the number of bytes per unit for the test, or a good - estimate of it. For example, a test that needs two byte buffers, of 4 GiB - each, could be decorated with @bigmemtest(size=_4G, memuse=2). + 'minsize' is the minimum useful size for the test (in arbitrary, + test-interpreted units.) 'memuse' is the number of 'bytes per size' for + the test, or a good estimate of it. 'overhead' specifies fixed overhead, + independent of the testsize, and defaults to 5Mb. - The 'size' argument is normally passed to the decorated test method as an - extra argument. If 'dry_run' is true, the value passed to the test method - may be less than the requested value. If 'dry_run' is false, it means the - test doesn't support dummy runs when -M is not specified. + The decorator tries to guess a good value for 'size' and passes it to + the decorated test function. If minsize * memuse is more than the + allowed memory use (as defined by max_memuse), the test is skipped. + Otherwise, minsize is adjusted upward to use up to max_memuse. """ def decorator(f): def wrapper(self): - size = wrapper.size - memuse = wrapper.memuse + if not max_memuse: + # If max_memuse is 0 (the default), + # we still want to run the tests with size set to a few kb, + # to make sure they work. We still want to avoid using + # too much memory, though, but we do that noisily. + maxsize = 5147 + self.assertFalse(maxsize * memuse + overhead > 20 * _1M) + else: + maxsize = int((max_memuse - overhead) / memuse) + if maxsize < minsize: + # Really ought to print 'test skipped' or something + if verbose: + sys.stderr.write("Skipping %s because of memory " + "constraint\n" % (f.__name__,)) + return + # Try to keep some breathing room in memory use + maxsize = max(maxsize - 50 * _1M, minsize) + return f(self, maxsize) + wrapper.minsize = minsize + wrapper.memuse = memuse + wrapper.overhead = overhead + return wrapper + return decorator + +def precisionbigmemtest(size, memuse, overhead=5*_1M, dry_run=True): + def decorator(f): + def wrapper(self): if not real_max_memuse: maxsize = 5147 else: @@ -1932,27 +1443,15 @@ def bigmemtest(size, memuse, dry_run=True): if ((real_max_memuse or not dry_run) and real_max_memuse < maxsize * memuse): - raise unittest.SkipTest( - "not enough memory: %.1fG minimum needed" - % (size * memuse / (1024 ** 3))) - - if real_max_memuse and verbose: - print() - print(" ... expected peak memory use: {peak:.1f}G" - .format(peak=size * memuse / (1024 ** 3))) - watchdog = _MemoryWatchdog() - watchdog.start() - else: - watchdog = None - - try: - return f(self, maxsize) - finally: - if watchdog: - watchdog.stop() + if verbose: + sys.stderr.write("Skipping %s because of memory " + "constraint\n" % (f.__name__,)) + return + return f(self, maxsize) wrapper.size = size wrapper.memuse = memuse + wrapper.overhead = overhead return wrapper return decorator @@ -1960,13 +1459,9 @@ def bigaddrspacetest(f): """Decorator for tests that fill the address space.""" def wrapper(self): if max_memuse < MAX_Py_ssize_t: - if MAX_Py_ssize_t >= 2**63 - 1 and max_memuse >= 2**31: - raise unittest.SkipTest( - "not enough memory: try a 32-bit build instead") - else: - raise unittest.SkipTest( - "not enough memory: %.1fG minimum needed" - % (MAX_Py_ssize_t / (1024 ** 3))) + if verbose: + sys.stderr.write("Skipping %s because of memory " + "constraint\n" % (f.__name__,)) else: return f(self) return wrapper @@ -2014,8 +1509,8 @@ def _parse_guards(guards): # Returns a tuple ({platform_name: run_me}, default_value) if not guards: return ({'cpython': True}, False) - is_true = list(guards.values())[0] - assert list(guards.values()) == [is_true] * len(guards) # all True or all False + is_true = guards.values()[0] + assert guards.values() == [is_true] * len(guards) # all True or all False return (guards, not is_true) # Use the following check to guard CPython's implementation-specific tests -- @@ -2031,33 +1526,6 @@ def check_impl_detail(**guards): return guards.get(platform.python_implementation().lower(), default) -def no_tracing(func): - """Decorator to temporarily turn off tracing for the duration of a test.""" - if not hasattr(sys, 'gettrace'): - return func - else: - @functools.wraps(func) - def wrapper(*args, **kwargs): - original_trace = sys.gettrace() - try: - sys.settrace(None) - return func(*args, **kwargs) - finally: - sys.settrace(original_trace) - return wrapper - - -def refcount_test(test): - """Decorator for tests which involve reference counting. - - To start, the decorator does not run the test if is not run by CPython. - After that, any trace function is unset during the test to prevent - unexpected refcounts caused by the trace function. - - """ - return no_tracing(cpython_only(test)) - - def _filter_suite(suite, pred): """Recursively filter test cases in a suite based on a predicate.""" newtests = [] @@ -2072,15 +1540,13 @@ def _filter_suite(suite, pred): def _run_suite(suite): """Run tests from a unittest.TestSuite-derived class.""" - runner = get_test_runner(sys.stdout, - verbosity=verbose, - capture_output=(junit_xml_list is not None)) + if verbose: + runner = unittest.TextTestRunner(sys.stdout, verbosity=2, + failfast=failfast) + else: + runner = BasicTestRunner() result = runner.run(suite) - - if junit_xml_list is not None: - junit_xml_list.append(result.get_xml_element()) - if not result.testsRun and not result.skipped: raise TestDidNotRun if not result.wasSuccessful(): @@ -2090,15 +1556,14 @@ def _run_suite(suite): err = result.failures[0][1] else: err = "multiple errors occurred" - if not verbose: err += "; run in verbose mode for details" + if not verbose: + err += "; run in verbose mode for details" raise TestFailed(err) # By default, don't filter tests _match_test_func = None - -_accept_test_patterns = None -_ignore_test_patterns = None +_match_test_patterns = None def match_test(test): @@ -2114,52 +1579,25 @@ def _is_full_match_test(pattern): # as a full test identifier. # Example: 'test.test_os.FileTests.test_access'. # - # ignore patterns which contain fnmatch patterns: '*', '?', '[...]' - # or '[!...]'. For example, ignore 'test_access*'. + # Reject patterns which contain fnmatch patterns: '*', '?', '[...]' + # or '[!...]'. For example, reject 'test_access*'. return ('.' in pattern) and (not re.search(r'[?*\[\]]', pattern)) -def set_match_tests(accept_patterns=None, ignore_patterns=None): - global _match_test_func, _accept_test_patterns, _ignore_test_patterns - - - if accept_patterns is None: - accept_patterns = () - if ignore_patterns is None: - ignore_patterns = () - - accept_func = ignore_func = None - - if accept_patterns != _accept_test_patterns: - accept_patterns, accept_func = _compile_match_function(accept_patterns) - if ignore_patterns != _ignore_test_patterns: - ignore_patterns, ignore_func = _compile_match_function(ignore_patterns) - - # Create a copy since patterns can be mutable and so modified later - _accept_test_patterns = tuple(accept_patterns) - _ignore_test_patterns = tuple(ignore_patterns) - - if accept_func is not None or ignore_func is not None: - def match_function(test_id): - accept = True - ignore = False - if accept_func: - accept = accept_func(test_id) - if ignore_func: - ignore = ignore_func(test_id) - return accept and not ignore - - _match_test_func = match_function +def set_match_tests(patterns): + global _match_test_func, _match_test_patterns + if patterns == _match_test_patterns: + # No change: no need to recompile patterns. + return -def _compile_match_function(patterns): if not patterns: func = None # set_match_tests(None) behaves as set_match_tests(()) patterns = () elif all(map(_is_full_match_test, patterns)): # Simple case: all patterns are full test identifier. - # The test.bisect_cmd utility only uses such full test identifiers. + # The test.bisect utility only uses such full test identifiers. func = set(patterns).__contains__ else: regex = '|'.join(map(fnmatch.translate, patterns)) @@ -2169,8 +1607,8 @@ def _compile_match_function(patterns): def match_test_regex(test_id): if regex_match(test_id): - # The regex matches the whole identifier, for example - # 'test.test_os.FileTests.test_access'. + # The regex matchs the whole identifier like + # 'test.test_os.FileTests.test_access' return True else: # Try to match parts of the test identifier. @@ -2180,7 +1618,10 @@ def _compile_match_function(patterns): func = match_test_regex - return patterns, func + # Create a copy since patterns can be mutable and so modified later + _match_test_patterns = tuple(patterns) + _match_test_func = func + def run_unittest(*classes): @@ -2203,18 +1644,9 @@ def run_unittest(*classes): #======================================================================= # Check for the presence of docstrings. -# Rather than trying to enumerate all the cases where docstrings may be -# disabled, we just check for that directly - -def _check_docstrings(): - """Just used to check if docstrings are enabled""" - -MISSING_C_DOCSTRINGS = (check_impl_detail() and - sys.platform != 'win32' and - not sysconfig.get_config_var('WITH_DOC_STRINGS')) - -HAVE_DOCSTRINGS = (_check_docstrings.__doc__ is not None and - not MISSING_C_DOCSTRINGS) +HAVE_DOCSTRINGS = (check_impl_detail(cpython=False) or + sys.platform == 'win32' or + sysconfig.get_config_var('WITH_DOC_STRINGS')) requires_docstrings = unittest.skipUnless(HAVE_DOCSTRINGS, "test requires docstrings") @@ -2223,11 +1655,11 @@ requires_docstrings = unittest.skipUnless(HAVE_DOCSTRINGS, #======================================================================= # doctest driver. -def run_doctest(module, verbosity=None, optionflags=0): +def run_doctest(module, verbosity=None): """Run doctest on the given module. Return (#failures, #tests). If optional argument verbosity is not specified (or is None), pass - support's belief about verbosity on to doctest. Else doctest's + test.support's belief about verbosity on to doctest. Else doctest's usual behavior is used (it searches sys.argv for -v). """ @@ -2238,37 +1670,20 @@ def run_doctest(module, verbosity=None, optionflags=0): else: verbosity = None - f, t = doctest.testmod(module, verbose=verbosity, optionflags=optionflags) - if f: - raise TestFailed("%d of %d doctests failed" % (f, t)) + # Direct doctest output (normally just errors) to real stdout; doctest + # output shouldn't be compared by regrtest. + save_stdout = sys.stdout + sys.stdout = get_original_stdout() + try: + f, t = doctest.testmod(module, verbose=verbosity) + if f: + raise TestFailed("%d of %d doctests failed" % (f, t)) + finally: + sys.stdout = save_stdout if verbose: - print('doctest (%s) ... %d tests with zero failures' % - (module.__name__, t)) + print 'doctest (%s) ... %d tests with zero failures' % (module.__name__, t) return f, t - -#======================================================================= -# Support for saving and restoring the imported modules. - -def modules_setup(): - return sys.modules.copy(), - -def modules_cleanup(oldmodules): - # Encoders/decoders are registered permanently within the internal - # codec cache. If we destroy the corresponding modules their - # globals will be set to None which will trip up the cached functions. - encodings = [(k, v) for k, v in sys.modules.items() - if k.startswith('encodings.')] - sys.modules.clear() - sys.modules.update(encodings) - # XXX: This kind of problem can affect more than just encodings. In particular - # extension modules (such as _ssl) don't cope with reloading properly. - # Really, test modules should be cleaning out the test specific modules they - # know they added (ala test_runpy) rather than relying on this function (as - # test_importhooks and test_pkg do currently). - # Implicitly imported *real* modules should be left alone (see issue 10556). - sys.modules.update(oldmodules) - #======================================================================= # Threading support to prevent reporting refleaks when running regrtest.py -R @@ -2289,43 +1704,31 @@ environment_altered = False # at the end of a test run. def threading_setup(): - return _thread._count(), threading._dangling.copy() - -def threading_cleanup(*original_values): - global environment_altered + if thread: + return thread._count(), + else: + return 1, - _MAX_COUNT = 100 +def threading_cleanup(nb_threads): + if not thread: + return + _MAX_COUNT = 10 for count in range(_MAX_COUNT): - values = _thread._count(), threading._dangling - if values == original_values: + n = thread._count() + if n == nb_threads: break - - if not count: - # Display a warning at the first iteration - environment_altered = True - dangling_threads = values[1] - print("Warning -- threading_cleanup() failed to cleanup " - "%s threads (count: %s, dangling: %s)" - % (values[0] - original_values[0], - values[0], len(dangling_threads)), - file=sys.stderr) - for thread in dangling_threads: - print(f"Dangling thread: {thread!r}", file=sys.stderr) - sys.stderr.flush() - - # Don't hold references to threads - dangling_threads = None - values = None - - time.sleep(0.01) - gc_collect() - + time.sleep(0.1) + # XXX print a warning in case of failure? def reap_threads(func): """Use this function when threads are being used. This will ensure that the threads are cleaned up even when the test fails. + If threading is unavailable this function does nothing. """ + if not thread: + return func + @functools.wraps(func) def decorator(*args): key = threading_setup() @@ -2337,82 +1740,60 @@ def reap_threads(func): @contextlib.contextmanager -def wait_threads_exit(timeout=None): +def wait_threads_exit(timeout=60.0): """ bpo-31234: Context manager to wait until all threads created in the with statement exit. - Use _thread.count() to check if threads exited. Indirectly, wait until - threads exit the internal t_bootstrap() C function of the _thread module. + Use thread.count() to check if threads exited. Indirectly, wait until + threads exit the internal t_bootstrap() C function of the thread module. threading_setup() and threading_cleanup() are designed to emit a warning if a test leaves running threads in the background. This context manager - is designed to cleanup threads started by the _thread.start_new_thread() + is designed to cleanup threads started by the thread.start_new_thread() which doesn't allow to wait for thread exit, whereas thread.Thread has a join() method. """ - if timeout is None: - timeout = SHORT_TIMEOUT - old_count = _thread._count() + old_count = thread._count() try: yield finally: - start_time = time.monotonic() + start_time = time.time() deadline = start_time + timeout while True: - count = _thread._count() + count = thread._count() if count <= old_count: break - if time.monotonic() > deadline: - dt = time.monotonic() - start_time - msg = (f"wait_threads() failed to cleanup {count - old_count} " - f"threads after {dt:.1f} seconds " - f"(count: {count}, old count: {old_count})") + if time.time() > deadline: + dt = time.time() - start_time + msg = ("wait_threads() failed to cleanup %s " + "threads after %.1f seconds " + "(count: %s, old count: %s)" + % (count - old_count, dt, count, old_count)) raise AssertionError(msg) time.sleep(0.010) gc_collect() -def join_thread(thread, timeout=None): - """Join a thread. Raise an AssertionError if the thread is still alive - after timeout seconds. - """ - if timeout is None: - timeout = SHORT_TIMEOUT - thread.join(timeout) - if thread.is_alive(): - msg = f"failed to join the thread in {timeout:.1f} seconds" - raise AssertionError(msg) - - def reap_children(): """Use this function at the end of test_main() whenever sub-processes are started. This will help ensure that no extra children (zombies) stick around to hog resources and create problems when looking for refleaks. """ - global environment_altered - - # Need os.waitpid(-1, os.WNOHANG): Windows is not supported - if not (hasattr(os, 'waitpid') and hasattr(os, 'WNOHANG')): - return # Reap all our dead child processes so we don't leave zombies around. # These hog resources and might be causing some of the buildbots to die. - while True: - try: - # Read the exit status of any child process which already completed - pid, status = os.waitpid(-1, os.WNOHANG) - except OSError: - break - - if pid == 0: - break - - print("Warning -- reap_children() reaped child process %s" - % pid, file=sys.stderr) - environment_altered = True - + if hasattr(os, 'waitpid'): + any_process = -1 + while True: + try: + # This will raise an exception on Windows. That's ok. + pid, status = os.waitpid(any_process, os.WNOHANG) + if pid == 0: + break + except: + break @contextlib.contextmanager def start_threads(threads, unlock=None): @@ -2430,25 +1811,22 @@ def start_threads(threads, unlock=None): raise yield finally: - try: - if unlock: - unlock() - endtime = starttime = time.monotonic() - for timeout in range(1, 16): - endtime += 60 - for t in started: - t.join(max(endtime - time.monotonic(), 0.01)) - started = [t for t in started if t.is_alive()] - if not started: - break - if verbose: - print('Unable to join %d threads during a period of ' - '%d minutes' % (len(started), timeout)) - finally: - started = [t for t in started if t.is_alive()] - if started: - faulthandler.dump_traceback(sys.stdout) - raise AssertionError('Unable to join %d threads' % len(started)) + if unlock: + unlock() + endtime = starttime = time.time() + for timeout in range(1, 16): + endtime += 60 + for t in started: + t.join(max(endtime - time.time(), 0.01)) + started = [t for t in started if t.isAlive()] + if not started: + break + if verbose: + print('Unable to join %d threads during a period of ' + '%d minutes' % (len(started), timeout)) + started = [t for t in started if t.isAlive()] + if started: + raise AssertionError('Unable to join %d threads' % len(started)) @contextlib.contextmanager def swap_attr(obj, attr, new_val): @@ -2512,357 +1890,77 @@ def swap_item(obj, item, new_val): if item in obj: del obj[item] +def py3k_bytes(b): + """Emulate the py3k bytes() constructor. + + NOTE: This is only a best effort function. + """ + try: + # memoryview? + return b.tobytes() + except AttributeError: + try: + # iterable of ints? + return b"".join(chr(x) for x in b) + except TypeError: + return bytes(b) + requires_type_collecting = unittest.skipIf(hasattr(sys, 'getcounts'), 'types are immortal if COUNT_ALLOCS is defined') def args_from_interpreter_flags(): """Return a list of command-line arguments reproducing the current - settings in sys.flags and sys.warnoptions.""" + settings in sys.flags.""" + import subprocess return subprocess._args_from_interpreter_flags() -def optim_args_from_interpreter_flags(): - """Return a list of command-line arguments reproducing the current - optimization settings in sys.flags.""" - return subprocess._optim_args_from_interpreter_flags() - -#============================================================ -# Support for assertions about logging. -#============================================================ - -class TestHandler(logging.handlers.BufferingHandler): - def __init__(self, matcher): - # BufferingHandler takes a "capacity" argument - # so as to know when to flush. As we're overriding - # shouldFlush anyway, we can set a capacity of zero. - # You can call flush() manually to clear out the - # buffer. - logging.handlers.BufferingHandler.__init__(self, 0) - self.matcher = matcher - - def shouldFlush(self): - return False - - def emit(self, record): - self.format(record) - self.buffer.append(record.__dict__) - - def matches(self, **kwargs): - """ - Look for a saved dict whose keys/values match the supplied arguments. - """ - result = False - for d in self.buffer: - if self.matcher.matches(d, **kwargs): - result = True - break - return result - -class Matcher(object): - - _partial_matches = ('msg', 'message') +def strip_python_stderr(stderr): + """Strip the stderr of a Python process from potential debug output + emitted by the interpreter. - def matches(self, d, **kwargs): - """ - Try to match a single dict with the supplied arguments. - - Keys whose values are strings and which are in self._partial_matches - will be checked for partial (i.e. substring) matches. You can extend - this scheme to (for example) do regular expression matching, etc. - """ - result = True - for k in kwargs: - v = kwargs[k] - dv = d.get(k) - if not self.match_value(k, dv, v): - result = False - break - return result - - def match_value(self, k, dv, v): - """ - Try to match a single stored value (dv) with a supplied value (v). - """ - if type(v) != type(dv): - result = False - elif type(dv) is not str or k not in self._partial_matches: - result = (v == dv) - else: - result = dv.find(v) >= 0 - return result - - -_can_symlink = None -def can_symlink(): - global _can_symlink - if _can_symlink is not None: - return _can_symlink - symlink_path = TESTFN + "can_symlink" - try: - os.symlink(TESTFN, symlink_path) - can = True - except (OSError, NotImplementedError, AttributeError): - can = False - else: - os.remove(symlink_path) - _can_symlink = can - return can - -def skip_unless_symlink(test): - """Skip decorator for tests that require functional symlink""" - ok = can_symlink() - msg = "Requires functional symlink implementation" - return test if ok else unittest.skip(msg)(test) - -_buggy_ucrt = None -def skip_if_buggy_ucrt_strfptime(test): + This will typically be run on the result of the communicate() method + of a subprocess.Popen object. """ - Skip decorator for tests that use buggy strptime/strftime - - If the UCRT bugs are present time.localtime().tm_zone will be - an empty string, otherwise we assume the UCRT bugs are fixed - - See bpo-37552 [Windows] strptime/strftime return invalid - results with UCRT version 17763.615 - """ - global _buggy_ucrt - if _buggy_ucrt is None: - if(sys.platform == 'win32' and - locale.getdefaultlocale()[1] == 'cp65001' and - time.localtime().tm_zone == ''): - _buggy_ucrt = True - else: - _buggy_ucrt = False - return unittest.skip("buggy MSVC UCRT strptime/strftime")(test) if _buggy_ucrt else test - -class PythonSymlink: - """Creates a symlink for the current Python executable""" - def __init__(self, link=None): - self.link = link or os.path.abspath(TESTFN) - self._linked = [] - self.real = os.path.realpath(sys.executable) - self._also_link = [] - - self._env = None - - self._platform_specific() - - def _platform_specific(self): - pass - - if sys.platform == "win32": - def _platform_specific(self): - import _winapi - - if os.path.lexists(self.real) and not os.path.exists(self.real): - # App symlink appears to not exist, but we want the - # real executable here anyway - self.real = _winapi.GetModuleFileName(0) - - dll = _winapi.GetModuleFileName(sys.dllhandle) - src_dir = os.path.dirname(dll) - dest_dir = os.path.dirname(self.link) - self._also_link.append(( - dll, - os.path.join(dest_dir, os.path.basename(dll)) - )) - for runtime in glob.glob(os.path.join(src_dir, "vcruntime*.dll")): - self._also_link.append(( - runtime, - os.path.join(dest_dir, os.path.basename(runtime)) - )) - - self._env = {k.upper(): os.getenv(k) for k in os.environ} - self._env["PYTHONHOME"] = os.path.dirname(self.real) - if sysconfig.is_python_build(True): - self._env["PYTHONPATH"] = os.path.dirname(os.__file__) + stderr = re.sub(br"\[\d+ refs\]\r?\n?$", b"", stderr).strip() + return stderr - def __enter__(self): - os.symlink(self.real, self.link) - self._linked.append(self.link) - for real, link in self._also_link: - os.symlink(real, link) - self._linked.append(link) - return self - def __exit__(self, exc_type, exc_value, exc_tb): - for link in self._linked: - try: - os.remove(link) - except IOError as ex: - if verbose: - print("failed to clean up {}: {}".format(link, ex)) - - def _call(self, python, args, env, returncode): - cmd = [python, *args] - p = subprocess.Popen(cmd, stdout=subprocess.PIPE, - stderr=subprocess.PIPE, env=env) - r = p.communicate() - if p.returncode != returncode: - if verbose: - print(repr(r[0])) - print(repr(r[1]), file=sys.stderr) - raise RuntimeError( - 'unexpected return code: {0} (0x{0:08X})'.format(p.returncode)) - return r - - def call_real(self, *args, returncode=0): - return self._call(self.real, args, None, returncode) - - def call_link(self, *args, returncode=0): - return self._call(self.link, args, self._env, returncode) - - -_can_xattr = None -def can_xattr(): - global _can_xattr - if _can_xattr is not None: - return _can_xattr - if not hasattr(os, "setxattr"): - can = False - else: - tmp_dir = tempfile.mkdtemp() - tmp_fp, tmp_name = tempfile.mkstemp(dir=tmp_dir) - try: - with open(TESTFN, "wb") as fp: - try: - # TESTFN & tempfile may use different file systems with - # different capabilities - os.setxattr(tmp_fp, b"user.test", b"") - os.setxattr(tmp_name, b"trusted.foo", b"42") - os.setxattr(fp.fileno(), b"user.test", b"") - # Kernels < 2.6.39 don't respect setxattr flags. - kernel_version = platform.release() - m = re.match(r"2.6.(\d{1,2})", kernel_version) - can = m is None or int(m.group(1)) >= 39 - except OSError: - can = False - finally: - unlink(TESTFN) - unlink(tmp_name) - rmdir(tmp_dir) - _can_xattr = can - return can - -def skip_unless_xattr(test): - """Skip decorator for tests that require functional extended attributes""" - ok = can_xattr() - msg = "no non-broken extended attribute support" - return test if ok else unittest.skip(msg)(test) - -def skip_if_pgo_task(test): - """Skip decorator for tests not run in (non-extended) PGO task""" - ok = not PGO or PGO_EXTENDED - msg = "Not run for (non-extended) PGO task" - return test if ok else unittest.skip(msg)(test) - -_bind_nix_socket_error = None -def skip_unless_bind_unix_socket(test): - """Decorator for tests requiring a functional bind() for unix sockets.""" - if not hasattr(socket, 'AF_UNIX'): - return unittest.skip('No UNIX Sockets')(test) - global _bind_nix_socket_error - if _bind_nix_socket_error is None: - path = TESTFN + "can_bind_unix_socket" - with socket.socket(socket.AF_UNIX) as sock: +def check_free_after_iterating(test, iter, cls, args=()): + class A(cls): + def __del__(self): + done[0] = True try: - sock.bind(path) - _bind_nix_socket_error = False - except OSError as e: - _bind_nix_socket_error = e - finally: - unlink(path) - if _bind_nix_socket_error: - msg = 'Requires a functional unix bind(): %s' % _bind_nix_socket_error - return unittest.skip(msg)(test) - else: - return test - - -def fs_is_case_insensitive(directory): - """Detects if the file system for the specified directory is case-insensitive.""" - with tempfile.NamedTemporaryFile(dir=directory) as base: - base_path = base.name - case_path = base_path.upper() - if case_path == base_path: - case_path = base_path.lower() - try: - return os.path.samefile(base_path, case_path) - except FileNotFoundError: - return False - - -def detect_api_mismatch(ref_api, other_api, *, ignore=()): - """Returns the set of items in ref_api not in other_api, except for a - defined list of items to be ignored in this check. - - By default this skips private attributes beginning with '_' but - includes all magic methods, i.e. those starting and ending in '__'. - """ - missing_items = set(dir(ref_api)) - set(dir(other_api)) - if ignore: - missing_items -= set(ignore) - missing_items = set(m for m in missing_items - if not m.startswith('_') or m.endswith('__')) - return missing_items - - -def check__all__(test_case, module, name_of_module=None, extra=(), - blacklist=()): - """Assert that the __all__ variable of 'module' contains all public names. - - The module's public names (its API) are detected automatically based on - whether they match the public name convention and were defined in - 'module'. - - The 'name_of_module' argument can specify (as a string or tuple thereof) - what module(s) an API could be defined in in order to be detected as a - public API. One case for this is when 'module' imports part of its public - API from other modules, possibly a C backend (like 'csv' and its '_csv'). - - The 'extra' argument can be a set of names that wouldn't otherwise be - automatically detected as "public", like objects without a proper - '__module__' attribute. If provided, it will be added to the - automatically detected ones. - - The 'blacklist' argument can be a set of names that must not be treated - as part of the public API even though their names indicate otherwise. - - Usage: - import bar - import foo - import unittest - from test import support - - class MiscTestCase(unittest.TestCase): - def test__all__(self): - support.check__all__(self, foo) - - class OtherTestCase(unittest.TestCase): - def test__all__(self): - extra = {'BAR_CONST', 'FOO_CONST'} - blacklist = {'baz'} # Undocumented name. - # bar imports part of its API from _bar. - support.check__all__(self, bar, ('bar', '_bar'), - extra=extra, blacklist=blacklist) + next(it) + except StopIteration: + pass - """ + done = [False] + it = iter(A(*args)) + # Issue 26494: Shouldn't crash + test.assertRaises(StopIteration, next, it) + # The sequence should be deallocated just after the end of iterating + gc_collect() + test.assertTrue(done[0]) - if name_of_module is None: - name_of_module = (module.__name__, ) - elif isinstance(name_of_module, str): - name_of_module = (name_of_module, ) +@contextlib.contextmanager +def disable_gc(): + have_gc = gc.isenabled() + gc.disable() + try: + yield + finally: + if have_gc: + gc.enable() - expected = set(extra) - for name in dir(module): - if name.startswith('_') or name in blacklist: - continue - obj = getattr(module, name) - if (getattr(obj, '__module__', None) in name_of_module or - (not hasattr(obj, '__module__') and - not isinstance(obj, types.ModuleType))): - expected.add(name) - test_case.assertCountEqual(module.__all__, expected) +def python_is_optimized(): + """Find if Python was built with optimizations.""" + cflags = sysconfig.get_config_var('PY_CFLAGS') or '' + final_opt = "" + for opt in cflags.split(): + if opt.startswith('-O'): + final_opt = opt + return final_opt not in ('', '-O0', '-Og') class SuppressCrashReport: @@ -2894,23 +1992,28 @@ class SuppressCrashReport: # Suppress assert dialogs in debug builds # (see http://bugs.python.org/issue23314) try: - import msvcrt - msvcrt.CrtSetReportMode + import _testcapi + _testcapi.CrtSetReportMode except (AttributeError, ImportError): - # no msvcrt or a release build + # no _testcapi or a release build pass else: self.old_modes = {} - for report_type in [msvcrt.CRT_WARN, - msvcrt.CRT_ERROR, - msvcrt.CRT_ASSERT]: - old_mode = msvcrt.CrtSetReportMode(report_type, - msvcrt.CRTDBG_MODE_FILE) - old_file = msvcrt.CrtSetReportFile(report_type, - msvcrt.CRTDBG_FILE_STDERR) + for report_type in [_testcapi.CRT_WARN, + _testcapi.CRT_ERROR, + _testcapi.CRT_ASSERT]: + old_mode = _testcapi.CrtSetReportMode(report_type, + _testcapi.CRTDBG_MODE_FILE) + old_file = _testcapi.CrtSetReportFile(report_type, + _testcapi.CRTDBG_FILE_STDERR) self.old_modes[report_type] = old_mode, old_file else: + try: + import resource + except ImportError: + resource = None + if resource is not None: try: self.old_value = resource.getrlimit(resource.RLIMIT_CORE) @@ -2926,16 +2029,17 @@ class SuppressCrashReport: # # This assumes that this context manager is used in tests # that might trigger the next manager. + import subprocess cmd = ['/usr/bin/defaults', 'read', 'com.apple.CrashReporter', 'DialogType'] proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE) - with proc: - stdout = proc.communicate()[0] + stdout = proc.communicate()[0] if stdout.strip() == b'developer': - print("this test triggers the Crash Reporter, " - "that is intentional", end='', flush=True) + sys.stdout.write("this test triggers the Crash Reporter, " + "that is intentional") + sys.stdout.flush() return self @@ -2948,145 +2052,30 @@ class SuppressCrashReport: self._k32.SetErrorMode(self.old_value) if self.old_modes: - import msvcrt + import _testcapi for report_type, (old_mode, old_file) in self.old_modes.items(): - msvcrt.CrtSetReportMode(report_type, old_mode) - msvcrt.CrtSetReportFile(report_type, old_file) - else: - if resource is not None: - try: - resource.setrlimit(resource.RLIMIT_CORE, self.old_value) - except (ValueError, OSError): - pass - - -def patch(test_instance, object_to_patch, attr_name, new_value): - """Override 'object_to_patch'.'attr_name' with 'new_value'. - - Also, add a cleanup procedure to 'test_instance' to restore - 'object_to_patch' value for 'attr_name'. - The 'attr_name' should be a valid attribute for 'object_to_patch'. - - """ - # check that 'attr_name' is a real attribute for 'object_to_patch' - # will raise AttributeError if it does not exist - getattr(object_to_patch, attr_name) - - # keep a copy of the old value - attr_is_local = False - try: - old_value = object_to_patch.__dict__[attr_name] - except (AttributeError, KeyError): - old_value = getattr(object_to_patch, attr_name, None) - else: - attr_is_local = True - - # restore the value when the test is done - def cleanup(): - if attr_is_local: - setattr(object_to_patch, attr_name, old_value) + _testcapi.CrtSetReportMode(report_type, old_mode) + _testcapi.CrtSetReportFile(report_type, old_file) else: - delattr(object_to_patch, attr_name) - - test_instance.addCleanup(cleanup) - - # actually override the attribute - setattr(object_to_patch, attr_name, new_value) - - -def run_in_subinterp(code): - """ - Run code in a subinterpreter. Raise unittest.SkipTest if the tracemalloc - module is enabled. - """ - # Issue #10915, #15751: PyGILState_*() functions don't work with - # sub-interpreters, the tracemalloc module uses these functions internally - try: - import tracemalloc - except ImportError: - pass - else: - if tracemalloc.is_tracing(): - raise unittest.SkipTest("run_in_subinterp() cannot be used " - "if tracemalloc module is tracing " - "memory allocations") - import _testcapi - return _testcapi.run_in_subinterp(code) - - -def check_free_after_iterating(test, iter, cls, args=()): - class A(cls): - def __del__(self): - nonlocal done - done = True + import resource try: - next(it) - except StopIteration: + resource.setrlimit(resource.RLIMIT_CORE, self.old_value) + except (ValueError, OSError): pass - done = False - it = iter(A(*args)) - # Issue 26494: Shouldn't crash - test.assertRaises(StopIteration, next, it) - # The sequence should be deallocated just after the end of iterating - gc_collect() - test.assertTrue(done) +def _crash_python(): + """Deliberate crash of Python. -def missing_compiler_executable(cmd_names=[]): - """Check if the compiler components used to build the interpreter exist. - - Check for the existence of the compiler executables whose names are listed - in 'cmd_names' or all the compiler executables when 'cmd_names' is empty - and return the first missing executable or None when none is found - missing. + Python can be killed by a segmentation fault (SIGSEGV), a bus error + (SIGBUS), or a different error depending on the platform. + Use SuppressCrashReport() to prevent a crash report from popping up. """ - from distutils import ccompiler, sysconfig, spawn - compiler = ccompiler.new_compiler() - sysconfig.customize_compiler(compiler) - for name in compiler.executables: - if cmd_names and name not in cmd_names: - continue - cmd = getattr(compiler, name) - if cmd_names: - assert cmd is not None, \ - "the '%s' executable is not configured" % name - elif not cmd: - continue - if spawn.find_executable(cmd[0]) is None: - return cmd[0] - - -_is_android_emulator = None -def setswitchinterval(interval): - # Setting a very low gil interval on the Android emulator causes python - # to hang (issue #26939). - minimum_interval = 1e-5 - if is_android and interval < minimum_interval: - global _is_android_emulator - if _is_android_emulator is None: - _is_android_emulator = (subprocess.check_output( - ['getprop', 'ro.kernel.qemu']).strip() == b'1') - if _is_android_emulator: - interval = minimum_interval - return sys.setswitchinterval(interval) - -@contextlib.contextmanager -def disable_faulthandler(): - # use sys.__stderr__ instead of sys.stderr, since regrtest replaces - # sys.stderr with a StringIO which has no file descriptor when a test - # is run with -W/--verbose3. - fd = sys.__stderr__.fileno() - - is_enabled = faulthandler.is_enabled() - try: - faulthandler.disable() - yield - finally: - if is_enabled: - faulthandler.enable(file=fd, all_threads=True) + import _testcapi + with SuppressCrashReport(): + _testcapi._read_null() def fd_count(): @@ -3095,11 +2084,12 @@ def fd_count(): if sys.platform.startswith(('linux', 'freebsd')): try: names = os.listdir("/proc/self/fd") - # Subtract one because listdir() internally opens a file + # Substract one because listdir() opens internally a file # descriptor to list the content of the /proc/self/fd/ directory. return len(names) - 1 - except FileNotFoundError: - pass + except OSError as exc: + if exc.errno != errno.ENOENT: + raise MAXFD = 256 if hasattr(os, 'sysconf'): @@ -3150,7 +2140,7 @@ def fd_count(): class SaveSignals: """ - Save and restore signal handlers. + Save an restore signal handlers. This class is only able to save/restore signal handlers registered by the Python signal module: see bpo-13285 for "external" signal @@ -3160,8 +2150,8 @@ class SaveSignals: def __init__(self): import signal self.signal = signal - self.signals = signal.valid_signals() - # SIGKILL and SIGSTOP signals cannot be ignored nor caught + self.signals = list(range(1, signal.NSIG)) + # SIGKILL and SIGSTOP signals cannot be ignored nor catched for signame in ('SIGKILL', 'SIGSTOP'): try: signum = getattr(signal, signame) @@ -3185,221 +2175,3 @@ class SaveSignals: def restore(self): for signum, handler in self.handlers.items(): self.signal.signal(signum, handler) - - -def with_pymalloc(): - import _testcapi - return _testcapi.WITH_PYMALLOC - - -class FakePath: - """Simple implementing of the path protocol. - """ - def __init__(self, path): - self.path = path - - def __repr__(self): - return f'<FakePath {self.path!r}>' - - def __fspath__(self): - if (isinstance(self.path, BaseException) or - isinstance(self.path, type) and - issubclass(self.path, BaseException)): - raise self.path - else: - return self.path - - -class _ALWAYS_EQ: - """ - Object that is equal to anything. - """ - def __eq__(self, other): - return True - def __ne__(self, other): - return False - -ALWAYS_EQ = _ALWAYS_EQ() - -class _NEVER_EQ: - """ - Object that is not equal to anything. - """ - def __eq__(self, other): - return False - def __ne__(self, other): - return True - def __hash__(self): - return 1 - -NEVER_EQ = _NEVER_EQ() - -@functools.total_ordering -class _LARGEST: - """ - Object that is greater than anything (except itself). - """ - def __eq__(self, other): - return isinstance(other, _LARGEST) - def __lt__(self, other): - return False - -LARGEST = _LARGEST() - -@functools.total_ordering -class _SMALLEST: - """ - Object that is less than anything (except itself). - """ - def __eq__(self, other): - return isinstance(other, _SMALLEST) - def __gt__(self, other): - return False - -SMALLEST = _SMALLEST() - -def maybe_get_event_loop_policy(): - """Return the global event loop policy if one is set, else return None.""" - return asyncio.events._event_loop_policy - -# Helpers for testing hashing. -NHASHBITS = sys.hash_info.width # number of bits in hash() result -assert NHASHBITS in (32, 64) - -# Return mean and sdev of number of collisions when tossing nballs balls -# uniformly at random into nbins bins. By definition, the number of -# collisions is the number of balls minus the number of occupied bins at -# the end. -def collision_stats(nbins, nballs): - n, k = nbins, nballs - # prob a bin empty after k trials = (1 - 1/n)**k - # mean # empty is then n * (1 - 1/n)**k - # so mean # occupied is n - n * (1 - 1/n)**k - # so collisions = k - (n - n*(1 - 1/n)**k) - # - # For the variance: - # n*(n-1)*(1-2/n)**k + meanempty - meanempty**2 = - # n*(n-1)*(1-2/n)**k + meanempty * (1 - meanempty) - # - # Massive cancellation occurs, and, e.g., for a 64-bit hash code - # 1-1/2**64 rounds uselessly to 1.0. Rather than make heroic (and - # error-prone) efforts to rework the naive formulas to avoid those, - # we use the `decimal` module to get plenty of extra precision. - # - # Note: the exact values are straightforward to compute with - # rationals, but in context that's unbearably slow, requiring - # multi-million bit arithmetic. - import decimal - with decimal.localcontext() as ctx: - bits = n.bit_length() * 2 # bits in n**2 - # At least that many bits will likely cancel out. - # Use that many decimal digits instead. - ctx.prec = max(bits, 30) - dn = decimal.Decimal(n) - p1empty = ((dn - 1) / dn) ** k - meanempty = n * p1empty - occupied = n - meanempty - collisions = k - occupied - var = dn*(dn-1)*((dn-2)/dn)**k + meanempty * (1 - meanempty) - return float(collisions), float(var.sqrt()) - - -class catch_unraisable_exception: - """ - Context manager catching unraisable exception using sys.unraisablehook. - - Storing the exception value (cm.unraisable.exc_value) creates a reference - cycle. The reference cycle is broken explicitly when the context manager - exits. - - Storing the object (cm.unraisable.object) can resurrect it if it is set to - an object which is being finalized. Exiting the context manager clears the - stored object. - - Usage: - - with support.catch_unraisable_exception() as cm: - # code creating an "unraisable exception" - ... - - # check the unraisable exception: use cm.unraisable - ... - - # cm.unraisable attribute no longer exists at this point - # (to break a reference cycle) - """ - - def __init__(self): - self.unraisable = None - self._old_hook = None - - def _hook(self, unraisable): - # Storing unraisable.object can resurrect an object which is being - # finalized. Storing unraisable.exc_value creates a reference cycle. - self.unraisable = unraisable - - def __enter__(self): - self._old_hook = sys.unraisablehook - sys.unraisablehook = self._hook - return self - - def __exit__(self, *exc_info): - sys.unraisablehook = self._old_hook - del self.unraisable - - -class catch_threading_exception: - """ - Context manager catching threading.Thread exception using - threading.excepthook. - - Attributes set when an exception is catched: - - * exc_type - * exc_value - * exc_traceback - * thread - - See threading.excepthook() documentation for these attributes. - - These attributes are deleted at the context manager exit. - - Usage: - - with support.catch_threading_exception() as cm: - # code spawning a thread which raises an exception - ... - - # check the thread exception, use cm attributes: - # exc_type, exc_value, exc_traceback, thread - ... - - # exc_type, exc_value, exc_traceback, thread attributes of cm no longer - # exists at this point - # (to avoid reference cycles) - """ - - def __init__(self): - self.exc_type = None - self.exc_value = None - self.exc_traceback = None - self.thread = None - self._old_hook = None - - def _hook(self, args): - self.exc_type = args.exc_type - self.exc_value = args.exc_value - self.exc_traceback = args.exc_traceback - self.thread = args.thread - - def __enter__(self): - self._old_hook = threading.excepthook - threading.excepthook = self._hook - return self - - def __exit__(self, *exc_info): - threading.excepthook = self._old_hook - del self.exc_type - del self.exc_value - del self.exc_traceback - del self.thread |