1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131
|
import subprocess
import sys
import os
from typing import Any, NoReturn
from test.support import os_helper, Py_DEBUG
from .setup import setup_process, setup_test_dir
from .runtests import WorkerRunTests, JsonFile, JsonFileType
from .single import run_single_test
from .utils import (
StrPath, StrJSON, TestFilter,
get_temp_dir, get_work_dir, exit_timeout)
USE_PROCESS_GROUP = (hasattr(os, "setsid") and hasattr(os, "killpg"))
NEED_TTY = {
'test_ioctl',
}
def create_worker_process(runtests: WorkerRunTests, output_fd: int,
tmp_dir: StrPath | None = None) -> subprocess.Popen[str]:
worker_json = runtests.as_json()
cmd = runtests.create_python_cmd()
cmd.extend(['-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.
#
# Emscripten and WASI Python must start in the Python source code directory
# to get 'python.js' or 'python.wasm' file. Then worker_process() changes
# to a temporary directory created to run tests.
work_dir = os_helper.SAVEDCWD
kwargs: dict[str, Any] = dict(
env=env,
stdout=output_fd,
# bpo-45410: Write stderr into stdout to keep messages order
stderr=output_fd,
text=True,
close_fds=True,
cwd=work_dir,
)
# Don't use setsid() in tests using TTY
test_name = runtests.tests[0]
if USE_PROCESS_GROUP and test_name not in NEED_TTY:
kwargs['start_new_session'] = True
# Include the test name in the TSAN log file name
if 'TSAN_OPTIONS' in env:
parts = env['TSAN_OPTIONS'].split(' ')
for i, part in enumerate(parts):
if part.startswith('log_path='):
parts[i] = f'{part}.{test_name}'
break
env['TSAN_OPTIONS'] = ' '.join(parts)
# Pass json_file to the worker process
json_file = runtests.json_file
json_file.configure_subprocess(kwargs)
with json_file.inherit_subprocess():
return subprocess.Popen(cmd, **kwargs)
def worker_process(worker_json: StrJSON) -> NoReturn:
runtests = WorkerRunTests.from_json(worker_json)
test_name = runtests.tests[0]
match_tests: TestFilter = runtests.match_tests
json_file: JsonFile = runtests.json_file
setup_test_dir(runtests.test_dir)
setup_process()
if runtests.rerun:
if match_tests:
matching = "matching: " + ", ".join(pattern for pattern, result in match_tests if result)
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)
if runtests.coverage:
if "test.cov" in sys.modules: # imported by -Xpresite=
result.covered_lines = list(sys.modules["test.cov"].coverage)
elif not Py_DEBUG:
print(
"Gathering coverage in worker processes requires --with-pydebug",
flush=True,
)
else:
raise LookupError(
"`test.cov` not found in sys.modules but coverage wanted"
)
if json_file.file_type == JsonFileType.STDOUT:
print()
result.write_json_into(sys.stdout)
else:
with json_file.open('w', encoding='utf-8') as json_fp:
result.write_json_into(json_fp)
sys.exit(0)
def main() -> NoReturn:
if len(sys.argv) != 2:
print("usage: python -m test.libregrtest.worker JSON")
sys.exit(1)
worker_json = sys.argv[1]
tmp_dir = get_temp_dir()
work_dir = get_work_dir(tmp_dir, worker=True)
with exit_timeout():
with os_helper.temp_cwd(work_dir, quiet=True):
worker_process(worker_json)
if __name__ == "__main__":
main()
|