mirror of https://github.com/python/cpython
719 lines
22 KiB
Python
719 lines
22 KiB
Python
import contextlib
|
|
import faulthandler
|
|
import locale
|
|
import math
|
|
import os.path
|
|
import platform
|
|
import random
|
|
import re
|
|
import shlex
|
|
import signal
|
|
import subprocess
|
|
import sys
|
|
import sysconfig
|
|
import tempfile
|
|
import textwrap
|
|
from collections.abc import Callable, Iterable
|
|
|
|
from test import support
|
|
from test.support import os_helper
|
|
from test.support import threading_helper
|
|
|
|
|
|
# All temporary files and temporary directories created by libregrtest should
|
|
# use TMP_PREFIX so cleanup_temp_dir() can remove them all.
|
|
TMP_PREFIX = 'test_python_'
|
|
WORK_DIR_PREFIX = TMP_PREFIX
|
|
WORKER_WORK_DIR_PREFIX = WORK_DIR_PREFIX + 'worker_'
|
|
|
|
# bpo-38203: Maximum delay in seconds to exit Python (call Py_Finalize()).
|
|
# Used to protect against threading._shutdown() hang.
|
|
# Must be smaller than buildbot "1200 seconds without output" limit.
|
|
EXIT_TIMEOUT = 120.0
|
|
|
|
|
|
ALL_RESOURCES = ('audio', 'curses', 'largefile', 'network',
|
|
'decimal', 'cpu', 'subprocess', 'urlfetch', 'gui', 'walltime')
|
|
|
|
# Other resources excluded from --use=all:
|
|
#
|
|
# - extralagefile (ex: test_zipfile64): really too slow to be enabled
|
|
# "by default"
|
|
# - tzdata: while needed to validate fully test_datetime, it makes
|
|
# test_datetime too slow (15-20 min on some buildbots) and so is disabled by
|
|
# default (see bpo-30822).
|
|
RESOURCE_NAMES = ALL_RESOURCES + ('extralargefile', 'tzdata')
|
|
|
|
|
|
# Types for types hints
|
|
StrPath = str
|
|
TestName = str
|
|
StrJSON = str
|
|
TestTuple = tuple[TestName, ...]
|
|
TestList = list[TestName]
|
|
# --match and --ignore options: list of patterns
|
|
# ('*' joker character can be used)
|
|
TestFilter = list[tuple[TestName, bool]]
|
|
FilterTuple = tuple[TestName, ...]
|
|
FilterDict = dict[TestName, FilterTuple]
|
|
|
|
|
|
def format_duration(seconds):
|
|
ms = math.ceil(seconds * 1e3)
|
|
seconds, ms = divmod(ms, 1000)
|
|
minutes, seconds = divmod(seconds, 60)
|
|
hours, minutes = divmod(minutes, 60)
|
|
|
|
parts = []
|
|
if hours:
|
|
parts.append('%s hour' % hours)
|
|
if minutes:
|
|
parts.append('%s min' % minutes)
|
|
if seconds:
|
|
if parts:
|
|
# 2 min 1 sec
|
|
parts.append('%s sec' % seconds)
|
|
else:
|
|
# 1.0 sec
|
|
parts.append('%.1f sec' % (seconds + ms / 1000))
|
|
if not parts:
|
|
return '%s ms' % ms
|
|
|
|
parts = parts[:2]
|
|
return ' '.join(parts)
|
|
|
|
|
|
def strip_py_suffix(names: list[str] | None) -> None:
|
|
if not names:
|
|
return
|
|
for idx, name in enumerate(names):
|
|
basename, ext = os.path.splitext(name)
|
|
if ext == '.py':
|
|
names[idx] = basename
|
|
|
|
|
|
def plural(n, singular, plural=None):
|
|
if n == 1:
|
|
return singular
|
|
elif plural is not None:
|
|
return plural
|
|
else:
|
|
return singular + 's'
|
|
|
|
|
|
def count(n, word):
|
|
if n == 1:
|
|
return f"{n} {word}"
|
|
else:
|
|
return f"{n} {word}s"
|
|
|
|
|
|
def printlist(x, width=70, indent=4, file=None):
|
|
"""Print the elements of iterable x to stdout.
|
|
|
|
Optional arg width (default 70) is the maximum line length.
|
|
Optional arg indent (default 4) is the number of blanks with which to
|
|
begin each line.
|
|
"""
|
|
|
|
blanks = ' ' * indent
|
|
# Print the sorted list: 'x' may be a '--random' list or a set()
|
|
print(textwrap.fill(' '.join(str(elt) for elt in sorted(x)), width,
|
|
initial_indent=blanks, subsequent_indent=blanks),
|
|
file=file)
|
|
|
|
|
|
def print_warning(msg):
|
|
support.print_warning(msg)
|
|
|
|
|
|
orig_unraisablehook = None
|
|
|
|
|
|
def regrtest_unraisable_hook(unraisable):
|
|
global orig_unraisablehook
|
|
support.environment_altered = True
|
|
support.print_warning("Unraisable exception")
|
|
old_stderr = sys.stderr
|
|
try:
|
|
support.flush_std_streams()
|
|
sys.stderr = support.print_warning.orig_stderr
|
|
orig_unraisablehook(unraisable)
|
|
sys.stderr.flush()
|
|
finally:
|
|
sys.stderr = old_stderr
|
|
|
|
|
|
def setup_unraisable_hook():
|
|
global orig_unraisablehook
|
|
orig_unraisablehook = sys.unraisablehook
|
|
sys.unraisablehook = regrtest_unraisable_hook
|
|
|
|
|
|
orig_threading_excepthook = None
|
|
|
|
|
|
def regrtest_threading_excepthook(args):
|
|
global orig_threading_excepthook
|
|
support.environment_altered = True
|
|
support.print_warning(f"Uncaught thread exception: {args.exc_type.__name__}")
|
|
old_stderr = sys.stderr
|
|
try:
|
|
support.flush_std_streams()
|
|
sys.stderr = support.print_warning.orig_stderr
|
|
orig_threading_excepthook(args)
|
|
sys.stderr.flush()
|
|
finally:
|
|
sys.stderr = old_stderr
|
|
|
|
|
|
def setup_threading_excepthook():
|
|
global orig_threading_excepthook
|
|
import threading
|
|
orig_threading_excepthook = threading.excepthook
|
|
threading.excepthook = regrtest_threading_excepthook
|
|
|
|
|
|
def clear_caches():
|
|
# Clear the warnings registry, so they can be displayed again
|
|
for mod in sys.modules.values():
|
|
if hasattr(mod, '__warningregistry__'):
|
|
del mod.__warningregistry__
|
|
|
|
# Flush standard output, so that buffered data is sent to the OS and
|
|
# associated Python objects are reclaimed.
|
|
for stream in (sys.stdout, sys.stderr, sys.__stdout__, sys.__stderr__):
|
|
if stream is not None:
|
|
stream.flush()
|
|
|
|
try:
|
|
re = sys.modules['re']
|
|
except KeyError:
|
|
pass
|
|
else:
|
|
re.purge()
|
|
|
|
try:
|
|
_strptime = sys.modules['_strptime']
|
|
except KeyError:
|
|
pass
|
|
else:
|
|
_strptime._regex_cache.clear()
|
|
|
|
try:
|
|
urllib_parse = sys.modules['urllib.parse']
|
|
except KeyError:
|
|
pass
|
|
else:
|
|
urllib_parse.clear_cache()
|
|
|
|
try:
|
|
urllib_request = sys.modules['urllib.request']
|
|
except KeyError:
|
|
pass
|
|
else:
|
|
urllib_request.urlcleanup()
|
|
|
|
try:
|
|
linecache = sys.modules['linecache']
|
|
except KeyError:
|
|
pass
|
|
else:
|
|
linecache.clearcache()
|
|
|
|
try:
|
|
mimetypes = sys.modules['mimetypes']
|
|
except KeyError:
|
|
pass
|
|
else:
|
|
mimetypes._default_mime_types()
|
|
|
|
try:
|
|
filecmp = sys.modules['filecmp']
|
|
except KeyError:
|
|
pass
|
|
else:
|
|
filecmp._cache.clear()
|
|
|
|
try:
|
|
struct = sys.modules['struct']
|
|
except KeyError:
|
|
pass
|
|
else:
|
|
struct._clearcache()
|
|
|
|
try:
|
|
doctest = sys.modules['doctest']
|
|
except KeyError:
|
|
pass
|
|
else:
|
|
doctest.master = None
|
|
|
|
try:
|
|
ctypes = sys.modules['ctypes']
|
|
except KeyError:
|
|
pass
|
|
else:
|
|
ctypes._reset_cache()
|
|
|
|
try:
|
|
typing = sys.modules['typing']
|
|
except KeyError:
|
|
pass
|
|
else:
|
|
for f in typing._cleanups:
|
|
f()
|
|
|
|
import inspect
|
|
abs_classes = filter(inspect.isabstract, typing.__dict__.values())
|
|
for abc in abs_classes:
|
|
for obj in abc.__subclasses__() + [abc]:
|
|
obj._abc_caches_clear()
|
|
|
|
try:
|
|
fractions = sys.modules['fractions']
|
|
except KeyError:
|
|
pass
|
|
else:
|
|
fractions._hash_algorithm.cache_clear()
|
|
|
|
try:
|
|
inspect = sys.modules['inspect']
|
|
except KeyError:
|
|
pass
|
|
else:
|
|
inspect._shadowed_dict_from_weakref_mro_tuple.cache_clear()
|
|
inspect._filesbymodname.clear()
|
|
inspect.modulesbyfile.clear()
|
|
|
|
try:
|
|
importlib_metadata = sys.modules['importlib.metadata']
|
|
except KeyError:
|
|
pass
|
|
else:
|
|
importlib_metadata.FastPath.__new__.cache_clear()
|
|
|
|
|
|
def get_build_info():
|
|
# Get most important configure and build options as a list of strings.
|
|
# Example: ['debug', 'ASAN+MSAN'] or ['release', 'LTO+PGO'].
|
|
|
|
config_args = sysconfig.get_config_var('CONFIG_ARGS') or ''
|
|
cflags = sysconfig.get_config_var('PY_CFLAGS') or ''
|
|
cflags_nodist = sysconfig.get_config_var('PY_CFLAGS_NODIST') or ''
|
|
ldflags_nodist = sysconfig.get_config_var('PY_LDFLAGS_NODIST') or ''
|
|
|
|
build = []
|
|
|
|
# --disable-gil
|
|
if sysconfig.get_config_var('Py_GIL_DISABLED'):
|
|
build.append("free_threading")
|
|
|
|
if hasattr(sys, 'gettotalrefcount'):
|
|
# --with-pydebug
|
|
build.append('debug')
|
|
|
|
if '-DNDEBUG' in (cflags + cflags_nodist):
|
|
build.append('without_assert')
|
|
else:
|
|
build.append('release')
|
|
|
|
if '--with-assertions' in config_args:
|
|
build.append('with_assert')
|
|
elif '-DNDEBUG' not in (cflags + cflags_nodist):
|
|
build.append('with_assert')
|
|
|
|
# --enable-framework=name
|
|
framework = sysconfig.get_config_var('PYTHONFRAMEWORK')
|
|
if framework:
|
|
build.append(f'framework={framework}')
|
|
|
|
# --enable-shared
|
|
shared = int(sysconfig.get_config_var('PY_ENABLE_SHARED') or '0')
|
|
if shared:
|
|
build.append('shared')
|
|
|
|
# --with-lto
|
|
optimizations = []
|
|
if '-flto=thin' in ldflags_nodist:
|
|
optimizations.append('ThinLTO')
|
|
elif '-flto' in ldflags_nodist:
|
|
optimizations.append('LTO')
|
|
|
|
if support.check_cflags_pgo():
|
|
# PGO (--enable-optimizations)
|
|
optimizations.append('PGO')
|
|
|
|
if support.check_bolt_optimized():
|
|
# BOLT (--enable-bolt)
|
|
optimizations.append('BOLT')
|
|
|
|
if optimizations:
|
|
build.append('+'.join(optimizations))
|
|
|
|
# --with-address-sanitizer
|
|
sanitizers = []
|
|
if support.check_sanitizer(address=True):
|
|
sanitizers.append("ASAN")
|
|
# --with-memory-sanitizer
|
|
if support.check_sanitizer(memory=True):
|
|
sanitizers.append("MSAN")
|
|
# --with-undefined-behavior-sanitizer
|
|
if support.check_sanitizer(ub=True):
|
|
sanitizers.append("UBSAN")
|
|
# --with-thread-sanitizer
|
|
if support.check_sanitizer(thread=True):
|
|
sanitizers.append("TSAN")
|
|
if sanitizers:
|
|
build.append('+'.join(sanitizers))
|
|
|
|
# --with-trace-refs
|
|
if hasattr(sys, 'getobjects'):
|
|
build.append("TraceRefs")
|
|
# --enable-pystats
|
|
if hasattr(sys, '_stats_on'):
|
|
build.append("pystats")
|
|
# --with-valgrind
|
|
if sysconfig.get_config_var('WITH_VALGRIND'):
|
|
build.append("valgrind")
|
|
# --with-dtrace
|
|
if sysconfig.get_config_var('WITH_DTRACE'):
|
|
build.append("dtrace")
|
|
|
|
return build
|
|
|
|
|
|
def get_temp_dir(tmp_dir: StrPath | None = None) -> StrPath:
|
|
if tmp_dir:
|
|
tmp_dir = os.path.expanduser(tmp_dir)
|
|
else:
|
|
# When tests are run from the Python build directory, it is best practice
|
|
# to keep the test files in a subfolder. This eases the cleanup of leftover
|
|
# files using the "make distclean" command.
|
|
if sysconfig.is_python_build():
|
|
if not support.is_wasi:
|
|
tmp_dir = sysconfig.get_config_var('abs_builddir')
|
|
if tmp_dir is None:
|
|
tmp_dir = sysconfig.get_config_var('abs_srcdir')
|
|
if not tmp_dir:
|
|
# gh-74470: On Windows, only srcdir is available. Using
|
|
# abs_builddir mostly matters on UNIX when building
|
|
# Python out of the source tree, especially when the
|
|
# source tree is read only.
|
|
tmp_dir = sysconfig.get_config_var('srcdir')
|
|
if not tmp_dir:
|
|
raise RuntimeError(
|
|
"Could not determine the correct value for tmp_dir"
|
|
)
|
|
tmp_dir = os.path.join(tmp_dir, 'build')
|
|
else:
|
|
# WASI platform
|
|
tmp_dir = sysconfig.get_config_var('projectbase')
|
|
if not tmp_dir:
|
|
raise RuntimeError(
|
|
"sysconfig.get_config_var('projectbase') "
|
|
f"unexpectedly returned {tmp_dir!r} on WASI"
|
|
)
|
|
tmp_dir = os.path.join(tmp_dir, 'build')
|
|
|
|
# When get_temp_dir() is called in a worker process,
|
|
# get_temp_dir() path is different than in the parent process
|
|
# which is not a WASI process. So the parent does not create
|
|
# the same "tmp_dir" than the test worker process.
|
|
os.makedirs(tmp_dir, exist_ok=True)
|
|
else:
|
|
tmp_dir = tempfile.gettempdir()
|
|
|
|
return os.path.abspath(tmp_dir)
|
|
|
|
|
|
def fix_umask():
|
|
if support.is_emscripten:
|
|
# Emscripten has default umask 0o777, which breaks some tests.
|
|
# see https://github.com/emscripten-core/emscripten/issues/17269
|
|
old_mask = os.umask(0)
|
|
if old_mask == 0o777:
|
|
os.umask(0o027)
|
|
else:
|
|
os.umask(old_mask)
|
|
|
|
|
|
def get_work_dir(parent_dir: StrPath, worker: bool = False) -> StrPath:
|
|
# Define a writable temp dir that will be used as cwd while running
|
|
# the tests. The name of the dir includes the pid to allow parallel
|
|
# testing (see the -j option).
|
|
# Emscripten and WASI have stubbed getpid(), Emscripten has only
|
|
# millisecond clock resolution. Use randint() instead.
|
|
if support.is_emscripten or support.is_wasi:
|
|
nounce = random.randint(0, 1_000_000)
|
|
else:
|
|
nounce = os.getpid()
|
|
|
|
if worker:
|
|
work_dir = WORK_DIR_PREFIX + str(nounce)
|
|
else:
|
|
work_dir = WORKER_WORK_DIR_PREFIX + str(nounce)
|
|
work_dir += os_helper.FS_NONASCII
|
|
work_dir = os.path.join(parent_dir, work_dir)
|
|
return work_dir
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def exit_timeout():
|
|
try:
|
|
yield
|
|
except SystemExit as exc:
|
|
# bpo-38203: Python can hang at exit in Py_Finalize(), especially
|
|
# on threading._shutdown() call: put a timeout
|
|
if threading_helper.can_start_thread:
|
|
faulthandler.dump_traceback_later(EXIT_TIMEOUT, exit=True)
|
|
sys.exit(exc.code)
|
|
|
|
|
|
def remove_testfn(test_name: TestName, verbose: int) -> None:
|
|
# Try to clean up os_helper.TESTFN if left behind.
|
|
#
|
|
# While tests shouldn't leave any files or directories behind, when a test
|
|
# fails that can be tedious for it to arrange. The consequences can be
|
|
# especially nasty on Windows, since if a test leaves a file open, it
|
|
# cannot be deleted by name (while there's nothing we can do about that
|
|
# here either, we can display the name of the offending test, which is a
|
|
# real help).
|
|
name = os_helper.TESTFN
|
|
if not os.path.exists(name):
|
|
return
|
|
|
|
nuker: Callable[[str], None]
|
|
if os.path.isdir(name):
|
|
import shutil
|
|
kind, nuker = "directory", shutil.rmtree
|
|
elif os.path.isfile(name):
|
|
kind, nuker = "file", os.unlink
|
|
else:
|
|
raise RuntimeError(f"os.path says {name!r} exists but is neither "
|
|
f"directory nor file")
|
|
|
|
if verbose:
|
|
print_warning(f"{test_name} left behind {kind} {name!r}")
|
|
support.environment_altered = True
|
|
|
|
try:
|
|
import stat
|
|
# fix possible permissions problems that might prevent cleanup
|
|
os.chmod(name, stat.S_IRWXU | stat.S_IRWXG | stat.S_IRWXO)
|
|
nuker(name)
|
|
except Exception as exc:
|
|
print_warning(f"{test_name} left behind {kind} {name!r} "
|
|
f"and it couldn't be removed: {exc}")
|
|
|
|
|
|
def abs_module_name(test_name: TestName, test_dir: StrPath | None) -> TestName:
|
|
if test_name.startswith('test.') or test_dir:
|
|
return test_name
|
|
else:
|
|
# Import it from the test package
|
|
return 'test.' + test_name
|
|
|
|
|
|
# gh-90681: When rerunning tests, we might need to rerun the whole
|
|
# class or module suite if some its life-cycle hooks fail.
|
|
# Test level hooks are not affected.
|
|
_TEST_LIFECYCLE_HOOKS = frozenset((
|
|
'setUpClass', 'tearDownClass',
|
|
'setUpModule', 'tearDownModule',
|
|
))
|
|
|
|
def normalize_test_name(test_full_name, *, is_error=False):
|
|
short_name = test_full_name.split(" ")[0]
|
|
if is_error and short_name in _TEST_LIFECYCLE_HOOKS:
|
|
if test_full_name.startswith(('setUpModule (', 'tearDownModule (')):
|
|
# if setUpModule() or tearDownModule() failed, don't filter
|
|
# tests with the test file name, don't use use filters.
|
|
return None
|
|
|
|
# This means that we have a failure in a life-cycle hook,
|
|
# we need to rerun the whole module or class suite.
|
|
# Basically the error looks like this:
|
|
# ERROR: setUpClass (test.test_reg_ex.RegTest)
|
|
# or
|
|
# ERROR: setUpModule (test.test_reg_ex)
|
|
# So, we need to parse the class / module name.
|
|
lpar = test_full_name.index('(')
|
|
rpar = test_full_name.index(')')
|
|
return test_full_name[lpar + 1: rpar].split('.')[-1]
|
|
return short_name
|
|
|
|
|
|
def adjust_rlimit_nofile():
|
|
"""
|
|
On macOS the default fd limit (RLIMIT_NOFILE) is sometimes too low (256)
|
|
for our test suite to succeed. Raise it to something more reasonable. 1024
|
|
is a common Linux default.
|
|
"""
|
|
try:
|
|
import resource
|
|
except ImportError:
|
|
return
|
|
|
|
fd_limit, max_fds = resource.getrlimit(resource.RLIMIT_NOFILE)
|
|
|
|
desired_fds = 1024
|
|
|
|
if fd_limit < desired_fds and fd_limit < max_fds:
|
|
new_fd_limit = min(desired_fds, max_fds)
|
|
try:
|
|
resource.setrlimit(resource.RLIMIT_NOFILE,
|
|
(new_fd_limit, max_fds))
|
|
print(f"Raised RLIMIT_NOFILE: {fd_limit} -> {new_fd_limit}")
|
|
except (ValueError, OSError) as err:
|
|
print_warning(f"Unable to raise RLIMIT_NOFILE from {fd_limit} to "
|
|
f"{new_fd_limit}: {err}.")
|
|
|
|
|
|
def get_host_runner():
|
|
if (hostrunner := os.environ.get("_PYTHON_HOSTRUNNER")) is None:
|
|
hostrunner = sysconfig.get_config_var("HOSTRUNNER")
|
|
return hostrunner
|
|
|
|
|
|
def is_cross_compiled():
|
|
return ('_PYTHON_HOST_PLATFORM' in os.environ)
|
|
|
|
|
|
def format_resources(use_resources: Iterable[str]):
|
|
use_resources = set(use_resources)
|
|
all_resources = set(ALL_RESOURCES)
|
|
|
|
# Express resources relative to "all"
|
|
relative_all = ['all']
|
|
for name in sorted(all_resources - use_resources):
|
|
relative_all.append(f'-{name}')
|
|
for name in sorted(use_resources - all_resources):
|
|
relative_all.append(f'{name}')
|
|
all_text = ','.join(relative_all)
|
|
all_text = f"resources: {all_text}"
|
|
|
|
# List of enabled resources
|
|
text = ','.join(sorted(use_resources))
|
|
text = f"resources ({len(use_resources)}): {text}"
|
|
|
|
# Pick the shortest string (prefer relative to all if lengths are equal)
|
|
if len(all_text) <= len(text):
|
|
return all_text
|
|
else:
|
|
return text
|
|
|
|
|
|
def display_header(use_resources: tuple[str, ...],
|
|
python_cmd: tuple[str, ...] | None):
|
|
# Print basic platform information
|
|
print("==", platform.python_implementation(), *sys.version.split())
|
|
print("==", platform.platform(aliased=True),
|
|
"%s-endian" % sys.byteorder)
|
|
print("== Python build:", ' '.join(get_build_info()))
|
|
print("== cwd:", os.getcwd())
|
|
|
|
cpu_count: object = os.cpu_count()
|
|
if cpu_count:
|
|
# The function is new in Python 3.13; mypy doesn't know about it yet:
|
|
process_cpu_count = os.process_cpu_count() # type: ignore[attr-defined]
|
|
if process_cpu_count and process_cpu_count != cpu_count:
|
|
cpu_count = f"{process_cpu_count} (process) / {cpu_count} (system)"
|
|
print("== CPU count:", cpu_count)
|
|
print("== encodings: locale=%s FS=%s"
|
|
% (locale.getencoding(), sys.getfilesystemencoding()))
|
|
|
|
if use_resources:
|
|
text = format_resources(use_resources)
|
|
print(f"== {text}")
|
|
else:
|
|
print("== resources: all test resources are disabled, "
|
|
"use -u option to unskip tests")
|
|
|
|
cross_compile = is_cross_compiled()
|
|
if cross_compile:
|
|
print("== cross compiled: Yes")
|
|
if python_cmd:
|
|
cmd = shlex.join(python_cmd)
|
|
print(f"== host python: {cmd}")
|
|
|
|
get_cmd = [*python_cmd, '-m', 'platform']
|
|
proc = subprocess.run(
|
|
get_cmd,
|
|
stdout=subprocess.PIPE,
|
|
text=True,
|
|
cwd=os_helper.SAVEDCWD)
|
|
stdout = proc.stdout.replace('\n', ' ').strip()
|
|
if stdout:
|
|
print(f"== host platform: {stdout}")
|
|
elif proc.returncode:
|
|
print(f"== host platform: <command failed with exit code {proc.returncode}>")
|
|
else:
|
|
hostrunner = get_host_runner()
|
|
if hostrunner:
|
|
print(f"== host runner: {hostrunner}")
|
|
|
|
# This makes it easier to remember what to set in your local
|
|
# environment when trying to reproduce a sanitizer failure.
|
|
asan = support.check_sanitizer(address=True)
|
|
msan = support.check_sanitizer(memory=True)
|
|
ubsan = support.check_sanitizer(ub=True)
|
|
tsan = support.check_sanitizer(thread=True)
|
|
sanitizers = []
|
|
if asan:
|
|
sanitizers.append("address")
|
|
if msan:
|
|
sanitizers.append("memory")
|
|
if ubsan:
|
|
sanitizers.append("undefined behavior")
|
|
if tsan:
|
|
sanitizers.append("thread")
|
|
if sanitizers:
|
|
print(f"== sanitizers: {', '.join(sanitizers)}")
|
|
for sanitizer, env_var in (
|
|
(asan, "ASAN_OPTIONS"),
|
|
(msan, "MSAN_OPTIONS"),
|
|
(ubsan, "UBSAN_OPTIONS"),
|
|
(tsan, "TSAN_OPTIONS"),
|
|
):
|
|
options= os.environ.get(env_var)
|
|
if sanitizer and options is not None:
|
|
print(f"== {env_var}={options!r}")
|
|
|
|
print(flush=True)
|
|
|
|
|
|
def cleanup_temp_dir(tmp_dir: StrPath):
|
|
import glob
|
|
|
|
path = os.path.join(glob.escape(tmp_dir), TMP_PREFIX + '*')
|
|
print("Cleanup %s directory" % tmp_dir)
|
|
for name in glob.glob(path):
|
|
if os.path.isdir(name):
|
|
print("Remove directory: %s" % name)
|
|
os_helper.rmtree(name)
|
|
else:
|
|
print("Remove file: %s" % name)
|
|
os_helper.unlink(name)
|
|
|
|
|
|
ILLEGAL_XML_CHARS_RE = re.compile(
|
|
'['
|
|
# Control characters; newline (\x0A and \x0D) and TAB (\x09) are legal
|
|
'\x00-\x08\x0B\x0C\x0E-\x1F'
|
|
# Surrogate characters
|
|
'\uD800-\uDFFF'
|
|
# Special Unicode characters
|
|
'\uFFFE'
|
|
'\uFFFF'
|
|
# Match multiple sequential invalid characters for better efficiency
|
|
']+')
|
|
|
|
def _sanitize_xml_replace(regs):
|
|
text = regs[0]
|
|
return ''.join(f'\\x{ord(ch):02x}' if ch <= '\xff' else ascii(ch)[1:-1]
|
|
for ch in text)
|
|
|
|
def sanitize_xml(text):
|
|
return ILLEGAL_XML_CHARS_RE.sub(_sanitize_xml_replace, text)
|