from test.libregrtest.logger import Logger
from test.libregrtest.result import State
from test.libregrtest.runtests import RunTests, HuntRefleak
-from test.libregrtest.setup import setup_tests, setup_test_dir
+from test.libregrtest.setup import setup_process, setup_test_dir
from test.libregrtest.single import run_single_test, PROGRESS_MIN_TIME
from test.libregrtest.pgo import setup_pgo_tests
from test.libregrtest.results import TestResults
from test.libregrtest.utils import (
- StrPath, StrJSON, TestName, TestList, FilterTuple,
+ StrPath, StrJSON, TestName, TestList, TestTuple, FilterTuple,
strip_py_suffix, count, format_duration,
printlist, get_build_info, get_temp_dir, get_work_dir, exit_timeout,
abs_module_name)
"""
def __init__(self, ns: Namespace):
# Log verbosity
- self.verbose: bool = ns.verbose
+ self.verbose: int = int(ns.verbose)
self.quiet: bool = ns.quiet
self.pgo: bool = ns.pgo
self.pgo_extended: bool = ns.pgo_extended
self.tmp_dir: StrPath | None = ns.tempdir
# tests
- self.tests = []
- self.selected: TestList = []
self.first_runtests: RunTests | None = None
# used by --slowest
def log(self, line=''):
self.logger.log(line)
- def find_tests(self):
+ def find_tests(self, tests: TestList | None = None) -> tuple[TestTuple, TestList | None]:
if self.single_test_run:
self.next_single_filename = os.path.join(self.tmp_dir, 'pynexttest')
try:
with open(self.next_single_filename, 'r') as fp:
next_test = fp.read().strip()
- self.tests = [next_test]
+ tests = [next_test]
except OSError:
pass
if self.fromfile:
- self.tests = []
+ tests = []
# regex to match 'test_builtin' in line:
# '0:00:00 [ 4/400] test_builtin -- test_dict took 1 sec'
regex = re.compile(r'\btest_[a-zA-Z0-9_]+\b')
line = line.strip()
match = regex.search(line)
if match is not None:
- self.tests.append(match.group())
+ tests.append(match.group())
- strip_py_suffix(self.tests)
+ strip_py_suffix(tests)
if self.pgo:
# add default PGO tests if no tests are specified
exclude=exclude_tests)
if not self.fromfile:
- self.selected = self.tests or self.cmdline_args
- if self.selected:
- self.selected = split_test_packages(self.selected)
+ selected = tests or self.cmdline_args
+ if selected:
+ selected = split_test_packages(selected)
else:
- self.selected = alltests
+ selected = alltests
else:
- self.selected = self.tests
+ selected = tests
if self.single_test_run:
- self.selected = self.selected[:1]
+ selected = selected[:1]
try:
- pos = alltests.index(self.selected[0])
+ pos = alltests.index(selected[0])
self.next_single_test = alltests[pos + 1]
except IndexError:
pass
# Remove all the selected tests that precede start if it's set.
if self.starting_test:
try:
- del self.selected[:self.selected.index(self.starting_test)]
+ del selected[:selected.index(self.starting_test)]
except ValueError:
print(f"Cannot find starting test: {self.starting_test}")
sys.exit(1)
if self.random_seed is None:
self.random_seed = random.randrange(100_000_000)
random.seed(self.random_seed)
- random.shuffle(self.selected)
+ random.shuffle(selected)
+
+ return (tuple(selected), tests)
@staticmethod
- def list_tests(tests: TestList):
+ def list_tests(tests: TestTuple):
for name in tests:
print(name)
if support.match_test(test):
print(test.id())
- def list_cases(self):
+ def list_cases(self, tests: TestTuple):
support.verbose = False
support.set_match_tests(self.match_tests, self.ignore_tests)
skipped = []
- for test_name in self.selected:
+ for test_name in tests:
module_name = abs_module_name(test_name, self.test_dir)
try:
suite = unittest.defaultTestLoader.loadTestsFromName(module_name)
def _rerun_failed_tests(self, runtests: RunTests):
# Configure the runner to re-run tests
if self.num_workers == 0:
+ # Always run tests in fresh processes to have more deterministic
+ # initial state. Don't re-run tests in parallel but limit to a
+ # single worker process to have side effects (on the system load
+ # and timings) between tests.
self.num_workers = 1
tests, match_tests_dict = self.results.prepare_rerun()
print()
print(f"== Tests result: {state} ==")
- self.results.display_result(self.selected, self.quiet, self.print_slowest)
+ self.results.display_result(runtests.tests,
+ self.quiet, self.print_slowest)
def run_test(self, test_name: TestName, runtests: RunTests, tracer):
if tracer is not None:
return state
def _run_tests_mp(self, runtests: RunTests, num_workers: int) -> None:
- from test.libregrtest.runtest_mp import RunWorkers
+ from test.libregrtest.run_workers import RunWorkers
RunWorkers(num_workers, runtests, self.logger, self.results).run()
def finalize_tests(self, tracer):
print("Remove file: %s" % name)
os_helper.unlink(name)
- def main(self, tests: TestList | None = None):
- if self.junit_filename and not os.path.isabs(self.junit_filename):
- self.junit_filename = os.path.abspath(self.junit_filename)
-
- self.tests = tests
-
- strip_py_suffix(self.cmdline_args)
-
- self.tmp_dir = get_temp_dir(self.tmp_dir)
-
- if self.want_cleanup:
- self.cleanup_temp_dir(self.tmp_dir)
- sys.exit(0)
-
- os.makedirs(self.tmp_dir, exist_ok=True)
- work_dir = get_work_dir(parent_dir=self.tmp_dir)
-
- with exit_timeout():
- # Run the tests in a context manager that temporarily changes the
- # CWD to a temporary and writable directory. If it's not possible
- # to create or change the CWD, the original CWD will be used.
- # The original CWD is available from os_helper.SAVEDCWD.
- with os_helper.temp_cwd(work_dir, quiet=True):
- # When using multiprocessing, worker processes will use
- # work_dir as their parent temporary directory. So when the
- # main process exit, it removes also subdirectories of worker
- # processes.
-
- self._main()
-
- def create_run_tests(self):
+ def create_run_tests(self, tests: TestTuple):
return RunTests(
- tuple(self.selected),
+ tests,
fail_fast=self.fail_fast,
match_tests=self.match_tests,
ignore_tests=self.ignore_tests,
python_cmd=self.python_cmd,
)
- def run_tests(self) -> int:
+ def _run_tests(self, selected: TestTuple, tests: TestList | None) -> int:
if self.hunt_refleak and self.hunt_refleak.warmups < 3:
msg = ("WARNING: Running tests with --huntrleaks/-R and "
"less than 3 warmup repetitions can give false positives!")
# For a partial run, we do not need to clutter the output.
if (self.want_header
or not(self.pgo or self.quiet or self.single_test_run
- or self.tests or self.cmdline_args)):
+ or tests or self.cmdline_args)):
self.display_header()
if self.randomize:
print("Using random seed", self.random_seed)
- runtests = self.create_run_tests()
+ runtests = self.create_run_tests(selected)
self.first_runtests = runtests
self.logger.set_tests(runtests)
- setup_tests(runtests)
+ setup_process()
self.logger.start_load_tracker()
try:
return self.results.get_exitcode(self.fail_env_changed,
self.fail_rerun)
- def _main(self):
+ def run_tests(self, selected: TestTuple, tests: TestList | None) -> int:
+ os.makedirs(self.tmp_dir, exist_ok=True)
+ work_dir = get_work_dir(parent_dir=self.tmp_dir)
+
+ # Put a timeout on Python exit
+ with exit_timeout():
+ # Run the tests in a context manager that temporarily changes the
+ # CWD to a temporary and writable directory. If it's not possible
+ # to create or change the CWD, the original CWD will be used.
+ # The original CWD is available from os_helper.SAVEDCWD.
+ with os_helper.temp_cwd(work_dir, quiet=True):
+ # When using multiprocessing, worker processes will use
+ # work_dir as their parent temporary directory. So when the
+ # main process exit, it removes also subdirectories of worker
+ # processes.
+ return self._run_tests(selected, tests)
+
+ def main(self, tests: TestList | None = None):
+ if self.junit_filename and not os.path.isabs(self.junit_filename):
+ self.junit_filename = os.path.abspath(self.junit_filename)
+
+ strip_py_suffix(self.cmdline_args)
+
+ self.tmp_dir = get_temp_dir(self.tmp_dir)
+
+ if self.want_cleanup:
+ self.cleanup_temp_dir(self.tmp_dir)
+ sys.exit(0)
+
if self.want_wait:
input("Press any key to continue...")
setup_test_dir(self.test_dir)
- self.find_tests()
+ selected, tests = self.find_tests(tests)
exitcode = 0
if self.want_list_tests:
- self.list_tests(self.selected)
+ self.list_tests(selected)
elif self.want_list_cases:
- self.list_cases()
+ self.list_cases(selected)
else:
- exitcode = self.run_tests()
+ exitcode = self.run_tests(selected, tests)
sys.exit(exitcode)
from test.support import TestStats
from test.libregrtest.utils import (
- TestName, FilterTuple,
+ StrJSON, TestName, FilterTuple,
format_duration, normalize_test_name, print_warning)
json.dump(self, file, cls=_EncodeTestResult)
@staticmethod
- def from_json(worker_json) -> 'TestResult':
+ def from_json(worker_json: StrJSON) -> 'TestResult':
return json.loads(worker_json, object_hook=_decode_test_result)
xml_data = result.xml_data
if xml_data:
- self.add_junit(result.xml_data)
+ self.add_junit(xml_data)
def need_rerun(self):
return bool(self.bad_results)
for s in ET.tostringlist(root):
f.write(s)
- def display_result(self, tests: TestList, quiet: bool, print_slowest: bool):
+ def display_result(self, tests: TestTuple, quiet: bool, print_slowest: bool):
if self.interrupted:
print("Test suite interrupted by signal SIGINT.")
from test.support import os_helper
from test.libregrtest.logger import Logger
-from test.libregrtest.main import Regrtest
from test.libregrtest.result import TestResult, State
from test.libregrtest.results import TestResults
from test.libregrtest.runtests import RunTests
) -> MultiprocessResult:
return MultiprocessResult(test_result, stdout, err_msg)
- def _run_process(self, worker_job, output_file: TextIO,
+ def _run_process(self, runtests: RunTests, output_file: TextIO,
tmp_dir: StrPath | None = None) -> int:
try:
- popen = create_worker_process(worker_job, output_file, tmp_dir)
+ popen = create_worker_process(runtests, output_file, tmp_dir)
self._killed = False
self._popen = popen
pgo_extended: bool = False
output_on_failure: bool = False
timeout: float | None = None
- verbose: bool = False
+ verbose: int = 0
quiet: bool = False
hunt_refleak: HuntRefleak | None = None
test_dir: StrPath | None = None
use_junit: bool = False
memory_limit: str | None = None
gc_threshold: int | None = None
- use_resources: list[str] = None
+ use_resources: list[str] = dataclasses.field(default_factory=list)
python_cmd: list[str] | None = None
def copy(self, **override):
-import atexit
import faulthandler
import os
import signal
from test.libregrtest.runtests import RunTests
from test.libregrtest.utils import (
- setup_unraisable_hook, setup_threading_excepthook, fix_umask)
+ setup_unraisable_hook, setup_threading_excepthook, fix_umask,
+ replace_stdout, adjust_rlimit_nofile)
UNICODE_GUARD_ENV = "PYTHONREGRTEST_UNICODE_GUARD"
sys.path.insert(0, os.path.abspath(testdir))
-def setup_support(runtests: RunTests):
- support.PGO = runtests.pgo
- support.PGO_EXTENDED = runtests.pgo_extended
- support.set_match_tests(runtests.match_tests, runtests.ignore_tests)
- support.failfast = runtests.fail_fast
- support.verbose = runtests.verbose
- if runtests.use_junit:
- support.junit_xml_list = []
- else:
- support.junit_xml_list = None
-
-
-def setup_tests(runtests):
+def setup_process():
fix_umask()
try:
for signum in signals:
faulthandler.register(signum, chain=True, file=stderr_fd)
- _adjust_resource_limits()
+ adjust_rlimit_nofile()
replace_stdout()
support.record_original_stdout(sys.stdout)
if getattr(module, '__file__', None):
module.__file__ = os.path.abspath(module.__file__)
- if runtests.hunt_refleak:
- unittest.BaseTestSuite._cleanup = False
-
- if runtests.memory_limit is not None:
- support.set_memlimit(runtests.memory_limit)
-
- if runtests.gc_threshold is not None:
- gc.set_threshold(runtests.gc_threshold)
-
- support.suppress_msvcrt_asserts(runtests.verbose and runtests.verbose >= 2)
-
- support.use_resources = runtests.use_resources
-
if hasattr(sys, 'addaudithook'):
# Add an auditing hook for all tests to ensure PySys_Audit is tested
def _test_audit_hook(name, args):
setup_unraisable_hook()
setup_threading_excepthook()
+ # Ensure there's a non-ASCII character in env vars at all times to force
+ # tests consider this case. See BPO-44647 for details.
+ if TESTFN_UNDECODABLE and os.supports_bytes_environ:
+ os.environb.setdefault(UNICODE_GUARD_ENV.encode(), TESTFN_UNDECODABLE)
+ elif FS_NONASCII:
+ os.environ.setdefault(UNICODE_GUARD_ENV, FS_NONASCII)
+
+
+def setup_tests(runtests: RunTests):
+ support.verbose = runtests.verbose
+ support.failfast = runtests.fail_fast
+ support.PGO = runtests.pgo
+ support.PGO_EXTENDED = runtests.pgo_extended
+
+ support.set_match_tests(runtests.match_tests, runtests.ignore_tests)
+
+ if runtests.use_junit:
+ support.junit_xml_list = []
+ from test.support.testresult import RegressionTestResult
+ RegressionTestResult.USE_XML = True
+ else:
+ support.junit_xml_list = None
+
+ if runtests.memory_limit is not None:
+ support.set_memlimit(runtests.memory_limit)
+
+ support.suppress_msvcrt_asserts(runtests.verbose >= 2)
+
+ support.use_resources = runtests.use_resources
+
timeout = runtests.timeout
if timeout is not None:
# For a slow buildbot worker, increase SHORT_TIMEOUT and LONG_TIMEOUT
support.SHORT_TIMEOUT = min(support.SHORT_TIMEOUT, timeout)
support.LONG_TIMEOUT = min(support.LONG_TIMEOUT, timeout)
- if runtests.use_junit:
- from test.support.testresult import RegressionTestResult
- RegressionTestResult.USE_XML = True
-
- # Ensure there's a non-ASCII character in env vars at all times to force
- # tests consider this case. See BPO-44647 for details.
- if TESTFN_UNDECODABLE and os.supports_bytes_environ:
- os.environb.setdefault(UNICODE_GUARD_ENV.encode(), TESTFN_UNDECODABLE)
- elif FS_NONASCII:
- os.environ.setdefault(UNICODE_GUARD_ENV, FS_NONASCII)
-
-
-def replace_stdout():
- """Set stdout encoder error handler to backslashreplace (as stderr error
- handler) to avoid UnicodeEncodeError when printing a traceback"""
- stdout = sys.stdout
- try:
- fd = stdout.fileno()
- except ValueError:
- # On IDLE, sys.stdout has no file descriptor and is not a TextIOWrapper
- # object. Leaving sys.stdout unchanged.
- #
- # Catch ValueError to catch io.UnsupportedOperation on TextIOBase
- # and ValueError on a closed stream.
- return
-
- sys.stdout = open(fd, 'w',
- encoding=stdout.encoding,
- errors="backslashreplace",
- closefd=False,
- newline='\n')
-
- def restore_stdout():
- sys.stdout.close()
- sys.stdout = stdout
- atexit.register(restore_stdout)
-
+ if runtests.hunt_refleak:
+ unittest.BaseTestSuite._cleanup = False
-def _adjust_resource_limits():
- """Adjust the system resource limits (ulimit) if needed."""
- try:
- import resource
- from resource import RLIMIT_NOFILE
- except ImportError:
- return
- fd_limit, max_fds = resource.getrlimit(RLIMIT_NOFILE)
- # On macOS the default fd limit is sometimes too low (256) for our
- # test suite to succeed. Raise it to something more reasonable.
- # 1024 is a common Linux default.
- desired_fds = 1024
- if fd_limit < desired_fds and fd_limit < max_fds:
- new_fd_limit = min(desired_fds, max_fds)
- try:
- resource.setrlimit(RLIMIT_NOFILE, (new_fd_limit, max_fds))
- print(f"Raised RLIMIT_NOFILE: {fd_limit} -> {new_fd_limit}")
- except (ValueError, OSError) as err:
- print(f"Unable to raise RLIMIT_NOFILE from {fd_limit} to "
- f"{new_fd_limit}: {err}.")
+ if runtests.gc_threshold is not None:
+ gc.set_threshold(runtests.gc_threshold)
from test.libregrtest.result import State, TestResult
from test.libregrtest.runtests import RunTests
from test.libregrtest.save_env import saved_test_environment
-from test.libregrtest.setup import setup_support
+from test.libregrtest.setup import setup_tests
from test.libregrtest.utils import (
TestName,
clear_caches, remove_testfn, abs_module_name, print_warning)
faulthandler.dump_traceback_later(timeout, exit=True)
try:
- setup_support(runtests)
+ setup_tests(runtests)
if output_on_failure:
support.verbose = True
+import atexit
import contextlib
import faulthandler
import math
rpar = test_full_name.index(')')
return test_full_name[lpar + 1: rpar].split('.')[-1]
return short_name
+
+
+def replace_stdout():
+ """Set stdout encoder error handler to backslashreplace (as stderr error
+ handler) to avoid UnicodeEncodeError when printing a traceback"""
+ stdout = sys.stdout
+ try:
+ fd = stdout.fileno()
+ except ValueError:
+ # On IDLE, sys.stdout has no file descriptor and is not a TextIOWrapper
+ # object. Leaving sys.stdout unchanged.
+ #
+ # Catch ValueError to catch io.UnsupportedOperation on TextIOBase
+ # and ValueError on a closed stream.
+ return
+
+ sys.stdout = open(fd, 'w',
+ encoding=stdout.encoding,
+ errors="backslashreplace",
+ closefd=False,
+ newline='\n')
+
+ def restore_stdout():
+ sys.stdout.close()
+ sys.stdout = stdout
+ atexit.register(restore_stdout)
+
+
+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}.")
from test import support
from test.support import os_helper
-from test.libregrtest.setup import setup_tests, setup_test_dir
+from test.libregrtest.setup import setup_process, setup_test_dir
from test.libregrtest.runtests import RunTests
from test.libregrtest.single import run_single_test
from test.libregrtest.utils import (
match_tests: FilterTuple | None = runtests.match_tests
setup_test_dir(runtests.test_dir)
- setup_tests(runtests)
+ setup_process()
if runtests.rerun:
if match_tests:
projects / thirdparty / Python / cpython.git / commitdiff
