group.add_argument('--wait', action='store_true',
help='wait for user input, e.g., allow a debugger '
'to be attached')
- group.add_argument('--worker-json', metavar='ARGS')
group.add_argument('-S', '--start', metavar='START',
help='the name of the test at which to start.' +
more_details)
-import faulthandler
import locale
import os
import platform
import random
import re
import sys
-import sysconfig
-import tempfile
import time
import unittest
+
+from test import support
+from test.support import os_helper
+
from test.libregrtest.cmdline import _parse_args, Namespace
from test.libregrtest.logger import Logger
from test.libregrtest.runtest import (
findtests, split_test_packages, run_single_test, abs_module_name,
PROGRESS_MIN_TIME, State, RunTests, HuntRefleak,
- FilterTuple, TestList, StrPath, StrJSON, TestName)
+ FilterTuple, TestList, StrJSON, TestName)
from test.libregrtest.setup import setup_tests, setup_test_dir
from test.libregrtest.pgo import setup_pgo_tests
from test.libregrtest.results import TestResults
-from test.libregrtest.utils import (strip_py_suffix, count, format_duration,
- printlist, get_build_info)
-from test import support
-from test.support import os_helper
-from test.support import threading_helper
-
-
-# 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
+from test.libregrtest.utils import (
+ strip_py_suffix, count, format_duration, StrPath,
+ printlist, get_build_info, get_temp_dir, get_work_dir, exit_timeout)
class Regrtest:
self.verbose: bool = ns.verbose
self.quiet: bool = ns.quiet
if ns.huntrleaks:
- self.hunt_refleak: HuntRefleak = HuntRefleak(*ns.huntrleaks)
+ warmups, runs, filename = ns.huntrleaks
+ filename = os.path.abspath(filename)
+ self.hunt_refleak: HuntRefleak = HuntRefleak(warmups, runs, filename)
else:
self.hunt_refleak = None
self.test_dir: StrPath | None = ns.testdir
state = self.get_state()
print(f"Result: {state}")
- @staticmethod
- 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)
-
- @staticmethod
- def select_temp_dir(tmp_dir):
- 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():
- tmp_dir = sysconfig.get_config_var('abs_builddir')
- if tmp_dir is None:
- # bpo-30284: 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')
- tmp_dir = os.path.join(tmp_dir, 'build')
- else:
- tmp_dir = tempfile.gettempdir()
-
- return os.path.abspath(tmp_dir)
-
- def is_worker(self):
- return (self.worker_json is not None)
-
- @staticmethod
- def make_temp_dir(tmp_dir: StrPath, is_worker: bool):
- os.makedirs(tmp_dir, exist_ok=True)
-
- # 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
- # milisecond clock resolution. Use randint() instead.
- if sys.platform in {"emscripten", "wasi"}:
- nounce = random.randint(0, 1_000_000)
- else:
- nounce = os.getpid()
-
- if is_worker:
- work_dir = 'test_python_worker_{}'.format(nounce)
- else:
- work_dir = 'test_python_{}'.format(nounce)
- work_dir += os_helper.FS_NONASCII
- work_dir = os.path.join(tmp_dir, work_dir)
- return work_dir
-
@staticmethod
def cleanup_temp_dir(tmp_dir: StrPath):
import glob
strip_py_suffix(self.cmdline_args)
- self.tmp_dir = self.select_temp_dir(self.tmp_dir)
-
- self.fix_umask()
+ self.tmp_dir = get_temp_dir(self.tmp_dir)
if self.want_cleanup:
self.cleanup_temp_dir(self.tmp_dir)
sys.exit(0)
- work_dir = self.make_temp_dir(self.tmp_dir, self.is_worker())
+ os.makedirs(self.tmp_dir, exist_ok=True)
+ work_dir = get_work_dir(parent_dir=self.tmp_dir)
- try:
+ 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.
# processes.
self._main()
- 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 create_run_tests(self):
return RunTests(
quiet=self.quiet,
hunt_refleak=self.hunt_refleak,
test_dir=self.test_dir,
- junit_filename=self.junit_filename,
+ use_junit=(self.junit_filename is not None),
memory_limit=self.memory_limit,
gc_threshold=self.gc_threshold,
use_resources=self.use_resources,
self.fail_rerun)
def _main(self):
- if self.is_worker():
- from test.libregrtest.runtest_mp import worker_process
- worker_process(self.worker_json)
- return
-
if self.want_wait:
input("Press any key to continue...")
warmups = hunt_refleak.warmups
runs = hunt_refleak.runs
filename = hunt_refleak.filename
- filename = os.path.join(os_helper.SAVEDCWD, filename)
repcount = warmups + runs
# Pre-allocate to ensure that the loop doesn't allocate anything new
from test.support import TestStats
from test.libregrtest.runtest import (
- TestName, TestTuple, TestList, FilterDict, StrPath, State,
+ TestName, TestTuple, TestList, FilterDict, State,
TestResult, RunTests)
-from test.libregrtest.utils import printlist, count, format_duration
+from test.libregrtest.utils import (
+ printlist, count, format_duration, StrPath)
EXITCODE_BAD_TEST = 2
from test.support import os_helper
from test.support import threading_helper
from test.libregrtest.save_env import saved_test_environment
-from test.libregrtest.utils import clear_caches, format_duration, print_warning
+from test.libregrtest.utils import (
+ clear_caches, format_duration, print_warning, StrPath)
StrJSON = str
-StrPath = str
TestName = str
TestTuple = tuple[TestName, ...]
TestList = list[TestName]
return None
return tuple(match_tests)
+ def write_json(self, file) -> None:
+ json.dump(self, file, cls=_EncodeTestResult)
+
+ @staticmethod
+ def from_json(worker_json) -> 'TestResult':
+ return json.loads(worker_json, object_hook=_decode_test_result)
+
+
+class _EncodeTestResult(json.JSONEncoder):
+ def default(self, o: Any) -> dict[str, Any]:
+ if isinstance(o, TestResult):
+ result = dataclasses.asdict(o)
+ result["__test_result__"] = o.__class__.__name__
+ return result
+ else:
+ return super().default(o)
+
+
+def _decode_test_result(data: dict[str, Any]) -> TestResult | dict[str, Any]:
+ if "__test_result__" in data:
+ data.pop('__test_result__')
+ if data['stats'] is not None:
+ data['stats'] = TestStats(**data['stats'])
+ return TestResult(**data)
+ else:
+ return data
+
@dataclasses.dataclass(slots=True, frozen=True)
class RunTests:
quiet: bool = False
hunt_refleak: HuntRefleak | None = None
test_dir: StrPath | None = None
- junit_filename: StrPath | None = None
+ use_junit: bool = False
memory_limit: str | None = None
gc_threshold: int | None = None
use_resources: list[str] = None
support.set_match_tests(runtests.match_tests, runtests.ignore_tests)
support.failfast = runtests.fail_fast
support.verbose = runtests.verbose
- if runtests.junit_filename:
+ if runtests.use_junit:
support.junit_xml_list = []
else:
support.junit_xml_list = None
Returns a TestResult.
- If runtests.junit_filename is not None, xml_data is a list containing each
- generated testsuite element.
+ If runtests.use_junit, xml_data is a list containing each generated
+ testsuite element.
"""
start_time = time.perf_counter()
result = TestResult(test_name)
import dataclasses
import faulthandler
-import json
import os.path
import queue
import signal
import threading
import time
import traceback
-from typing import NoReturn, Literal, Any, TextIO
+from typing import Literal, TextIO
from test import support
from test.support import os_helper
-from test.support import TestStats
from test.libregrtest.main import Regrtest
from test.libregrtest.runtest import (
- run_single_test, TestResult, State, PROGRESS_MIN_TIME,
- FilterTuple, RunTests, StrPath, StrJSON, TestName)
-from test.libregrtest.setup import setup_tests, setup_test_dir
+ TestResult, State, PROGRESS_MIN_TIME,
+ RunTests, TestName)
from test.libregrtest.results import TestResults
-from test.libregrtest.utils import format_duration, print_warning
+from test.libregrtest.utils import (
+ format_duration, print_warning, StrPath)
+from test.libregrtest.worker import create_worker_process, USE_PROCESS_GROUP
if sys.platform == 'win32':
import locale
# Time to wait until a worker completes: should be immediate
JOIN_TIMEOUT = 30.0 # seconds
-USE_PROCESS_GROUP = (hasattr(os, "setsid") and hasattr(os, "killpg"))
-
-
-@dataclasses.dataclass(slots=True)
-class WorkerJob:
- runtests: RunTests
-
-
-def create_worker_process(runtests: RunTests,
- output_file: TextIO,
- tmp_dir: StrPath | None = None) -> subprocess.Popen:
- python_cmd = runtests.python_cmd
- worker_json = runtests.as_json()
-
- if python_cmd is not None:
- executable = python_cmd
- else:
- executable = [sys.executable]
- cmd = [*executable, *support.args_from_interpreter_flags(),
- '-u', # Unbuffered stdout and stderr
- '-m', 'test.regrtest',
- '--worker-json', worker_json]
-
- env = dict(os.environ)
- if tmp_dir is not None:
- env['TMPDIR'] = tmp_dir
- env['TEMP'] = tmp_dir
- env['TMP'] = tmp_dir
-
- # Running the child from the same working directory as regrtest's original
- # invocation ensures that TEMPDIR for the child is the same when
- # sysconfig.is_python_build() is true. See issue 15300.
- kw = dict(
- env=env,
- stdout=output_file,
- # bpo-45410: Write stderr into stdout to keep messages order
- stderr=output_file,
- text=True,
- close_fds=(os.name != 'nt'),
- cwd=os_helper.SAVEDCWD,
- )
- if USE_PROCESS_GROUP:
- kw['start_new_session'] = True
- return subprocess.Popen(cmd, **kw)
-
-
-def worker_process(worker_json: StrJSON) -> NoReturn:
- runtests = RunTests.from_json(worker_json)
- test_name = runtests.tests[0]
- match_tests: FilterTuple | None = runtests.match_tests
-
- setup_test_dir(runtests.test_dir)
- setup_tests(runtests)
-
- if runtests.rerun:
- if match_tests:
- matching = "matching: " + ", ".join(match_tests)
- print(f"Re-running {test_name} in verbose mode ({matching})", flush=True)
- else:
- print(f"Re-running {test_name} in verbose mode", flush=True)
-
- result = run_single_test(test_name, runtests)
- print() # Force a newline (just in case)
-
- # Serialize TestResult as dict in JSON
- json.dump(result, sys.stdout, cls=EncodeTestResult)
- sys.stdout.flush()
- sys.exit(0)
-
# We do not use a generator so multiple threads can call next().
class MultiprocessIterator:
err_msg = "Failed to parse worker stdout"
else:
try:
- # deserialize run_tests_worker() output
- result = json.loads(worker_json,
- object_hook=decode_test_result)
+ result = TestResult.from_json(worker_json)
except Exception as exc:
err_msg = "Failed to parse worker JSON: %s" % exc
# worker when we exit this function
self.pending.stop()
self.stop_workers()
-
-
-class EncodeTestResult(json.JSONEncoder):
- """Encode a TestResult (sub)class object into a JSON dict."""
-
- def default(self, o: Any) -> dict[str, Any]:
- if isinstance(o, TestResult):
- result = dataclasses.asdict(o)
- result["__test_result__"] = o.__class__.__name__
- return result
-
- return super().default(o)
-
-
-def decode_test_result(d: dict[str, Any]) -> TestResult | dict[str, Any]:
- """Decode a TestResult (sub)class object from a JSON dict."""
-
- if "__test_result__" not in d:
- return d
-
- d.pop('__test_result__')
- if d['stats'] is not None:
- d['stats'] = TestStats(**d['stats'])
- return TestResult(**d)
except ImportError:
gc = None
-from test.libregrtest.utils import (setup_unraisable_hook,
- setup_threading_excepthook)
+from test.libregrtest.utils import (
+ setup_unraisable_hook, setup_threading_excepthook, fix_umask)
UNICODE_GUARD_ENV = "PYTHONREGRTEST_UNICODE_GUARD"
def setup_tests(runtests):
+ fix_umask()
+
try:
stderr_fd = sys.__stderr__.fileno()
except (ValueError, AttributeError):
support.SHORT_TIMEOUT = min(support.SHORT_TIMEOUT, timeout)
support.LONG_TIMEOUT = min(support.LONG_TIMEOUT, timeout)
- if runtests.junit_filename:
+ if runtests.use_junit:
from test.support.testresult import RegressionTestResult
RegressionTestResult.USE_XML = True
+import contextlib
+import faulthandler
import math
import os.path
+import random
import sys
import sysconfig
+import tempfile
import textwrap
+
from test import support
+from test.support import os_helper
+from test.support import threading_helper
MS_WINDOWS = (sys.platform == 'win32')
+# 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
+
+
+StrPath = str
+
def format_duration(seconds):
ms = math.ceil(seconds * 1e3)
build.append("dtrace")
return build
+
+
+def get_temp_dir(tmp_dir):
+ 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():
+ tmp_dir = sysconfig.get_config_var('abs_builddir')
+ if tmp_dir is None:
+ # bpo-30284: 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')
+ tmp_dir = os.path.join(tmp_dir, 'build')
+ 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):
+ # 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
+ # milisecond clock resolution. Use randint() instead.
+ if sys.platform in {"emscripten", "wasi"}:
+ nounce = random.randint(0, 1_000_000)
+ else:
+ nounce = os.getpid()
+
+ if worker:
+ work_dir = 'test_python_worker_{}'.format(nounce)
+ else:
+ work_dir = 'test_python_{}'.format(nounce)
+ work_dir += os_helper.FS_NONASCII
+ if parent_dir:
+ 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)
--- /dev/null
+import subprocess
+import sys
+import os
+from typing import TextIO, NoReturn
+
+from test import support
+from test.support import os_helper
+
+from test.libregrtest.setup import setup_tests, setup_test_dir
+from test.libregrtest.runtest import (
+ run_single_test, StrJSON, FilterTuple, RunTests)
+from test.libregrtest.utils import get_work_dir, exit_timeout, StrPath
+
+
+USE_PROCESS_GROUP = (hasattr(os, "setsid") and hasattr(os, "killpg"))
+
+
+def create_worker_process(runtests: RunTests,
+ output_file: TextIO,
+ tmp_dir: StrPath | None = None) -> subprocess.Popen:
+ python_cmd = runtests.python_cmd
+ worker_json = runtests.as_json()
+
+ if python_cmd is not None:
+ executable = python_cmd
+ else:
+ executable = [sys.executable]
+ cmd = [*executable, *support.args_from_interpreter_flags(),
+ '-u', # Unbuffered stdout and stderr
+ '-m', 'test.libregrtest.worker',
+ worker_json]
+
+ env = dict(os.environ)
+ if tmp_dir is not None:
+ env['TMPDIR'] = tmp_dir
+ env['TEMP'] = tmp_dir
+ env['TMP'] = tmp_dir
+
+ # Running the child from the same working directory as regrtest's original
+ # invocation ensures that TEMPDIR for the child is the same when
+ # sysconfig.is_python_build() is true. See issue 15300.
+ kw = dict(
+ env=env,
+ stdout=output_file,
+ # bpo-45410: Write stderr into stdout to keep messages order
+ stderr=output_file,
+ text=True,
+ close_fds=(os.name != 'nt'),
+ )
+ if USE_PROCESS_GROUP:
+ kw['start_new_session'] = True
+ return subprocess.Popen(cmd, **kw)
+
+
+def worker_process(worker_json: StrJSON) -> NoReturn:
+ runtests = RunTests.from_json(worker_json)
+ test_name = runtests.tests[0]
+ match_tests: FilterTuple | None = runtests.match_tests
+
+ setup_test_dir(runtests.test_dir)
+ setup_tests(runtests)
+
+ if runtests.rerun:
+ if match_tests:
+ matching = "matching: " + ", ".join(match_tests)
+ print(f"Re-running {test_name} in verbose mode ({matching})", flush=True)
+ else:
+ print(f"Re-running {test_name} in verbose mode", flush=True)
+
+ result = run_single_test(test_name, runtests)
+ print() # Force a newline (just in case)
+
+ # Serialize TestResult as dict in JSON
+ result.write_json(sys.stdout)
+ sys.stdout.flush()
+ sys.exit(0)
+
+
+def main():
+ if len(sys.argv) != 2:
+ print("usage: python -m test.libregrtest.worker JSON")
+ sys.exit(1)
+ worker_json = sys.argv[1]
+
+ work_dir = get_work_dir(worker=True)
+
+ with exit_timeout():
+ with os_helper.temp_cwd(work_dir, quiet=True):
+ worker_process(worker_json)
+
+
+if __name__ == "__main__":
+ main()
ns = libregrtest._parse_args(['--wait'])
self.assertTrue(ns.wait)
- def test_worker_json(self):
- ns = libregrtest._parse_args(['--worker-json', '[[], {}]'])
- self.assertEqual(ns.worker_json, '[[], {}]')
- self.checkError(['--worker-json'], 'expected one argument')
-
def test_start(self):
for opt in '-S', '--start':
with self.subTest(opt=opt):
projects / thirdparty / Python / cpython.git / commitdiff
