from .runtests import RunTests, JsonFile, JsonFileType
from .single import PROGRESS_MIN_TIME
from .utils import (
- StrPath, StrJSON, TestName, MS_WINDOWS, TMP_PREFIX,
+ StrPath, StrJSON, TestName, MS_WINDOWS,
format_duration, print_warning, count, plural)
from .worker import create_worker_process, USE_PROCESS_GROUP
def create_json_file(self, stack: contextlib.ExitStack) -> tuple[JsonFile, TextIO | None]:
"""Create JSON file."""
- json_file_use_filename = self.runtests.json_file_use_filename()
- if json_file_use_filename:
- # create an empty file to make the creation atomic
- # (to prevent races with other worker threads)
- prefix = TMP_PREFIX + 'json_'
- json_fd, json_filename = tempfile.mkstemp(prefix=prefix)
- os.close(json_fd)
-
- stack.callback(os_helper.unlink, json_filename)
- json_file = JsonFile(json_filename, JsonFileType.FILENAME)
+ json_file_use_stdout = self.runtests.json_file_use_stdout()
+ if json_file_use_stdout:
+ json_file = JsonFile(None, JsonFileType.STDOUT)
json_tmpfile = None
else:
json_tmpfile = tempfile.TemporaryFile('w+', encoding='utf8')
f"Cannot read process stdout: {exc}", None)
def read_json(self, json_file: JsonFile, json_tmpfile: TextIO | None,
- stdout: str) -> TestResult:
+ stdout: str) -> tuple[TestResult, str]:
try:
if json_tmpfile is not None:
json_tmpfile.seek(0)
worker_json: StrJSON = json_tmpfile.read()
+ elif json_file.file_type == JsonFileType.STDOUT:
+ stdout, _, worker_json = stdout.rpartition("\n")
+ stdout = stdout.rstrip()
else:
with json_file.open(encoding='utf8') as json_fp:
worker_json: StrJSON = json_fp.read()
raise WorkerError(self.test_name, "empty JSON", stdout)
try:
- return TestResult.from_json(worker_json)
+ result = TestResult.from_json(worker_json)
except Exception as exc:
# gh-101634: Catch UnicodeDecodeError if stdout cannot be
# decoded from encoding
raise WorkerError(self.test_name, err_msg, stdout,
state=State.MULTIPROCESSING_ERROR)
+ return (result, stdout)
+
def _runtest(self, test_name: TestName) -> MultiprocessResult:
with contextlib.ExitStack() as stack:
stdout_file = self.create_stdout(stack)
if retcode is None:
raise WorkerError(self.test_name, None, stdout, state=State.TIMEOUT)
- result = self.read_json(json_file, json_tmpfile, stdout)
+ result, stdout = self.read_json(json_file, json_tmpfile, stdout)
if retcode != 0:
raise WorkerError(self.test_name, f"Exit code {retcode}", stdout)
class JsonFileType:
UNIX_FD = "UNIX_FD"
WINDOWS_HANDLE = "WINDOWS_HANDLE"
- FILENAME = "FILENAME"
+ STDOUT = "STDOUT"
@dataclasses.dataclass(slots=True, frozen=True)
class JsonFile:
- # See RunTests.json_file_use_filename()
- file: int | StrPath
+ # file type depends on file_type:
+ # - UNIX_FD: file descriptor (int)
+ # - WINDOWS_HANDLE: handle (int)
+ # - STDOUT: use process stdout (None)
+ file: int | None
file_type: str
def configure_subprocess(self, popen_kwargs: dict) -> None:
startupinfo = subprocess.STARTUPINFO()
startupinfo.lpAttributeList = {"handle_list": [self.file]}
popen_kwargs['startupinfo'] = startupinfo
- case JsonFileType.FILENAME:
- # Filename: nothing to do to
- pass
@contextlib.contextmanager
def inherit_subprocess(self):
yield
def open(self, mode='r', *, encoding):
+ if self.file_type == JsonFileType.STDOUT:
+ raise ValueError("for STDOUT file type, just use sys.stdout")
+
file = self.file
if self.file_type == JsonFileType.WINDOWS_HANDLE:
import msvcrt
def from_json(worker_json: StrJSON) -> 'RunTests':
return json.loads(worker_json, object_hook=_decode_runtests)
- def json_file_use_filename(self) -> bool:
- # json_file type depends on the platform:
- # - Unix: file descriptor (int)
- # - Windows: handle (int)
- # - Emscripten/WASI or if --python is used: filename (str)
+ def json_file_use_stdout(self) -> bool:
+ # Use STDOUT in two cases:
+ #
+ # - If --python command line option is used;
+ # - On Emscripten and WASI.
+ #
+ # On other platforms, UNIX_FD or WINDOWS_HANDLE can be used.
return (
bool(self.python_cmd)
or support.is_emscripten
from test.support import os_helper
from .setup import setup_process, setup_test_dir
-from .runtests import RunTests, JsonFile
+from .runtests import RunTests, JsonFile, JsonFileType
from .single import run_single_test
from .utils import (
StrPath, StrJSON, FilterTuple,
runtests = RunTests.from_json(worker_json)
test_name = runtests.tests[0]
match_tests: FilterTuple | None = runtests.match_tests
- # json_file type depends on the platform:
- # - Unix: file descriptor (int)
- # - Windows: handle (int)
- # - Emscripten/WASI or if --python is used: filename (str)
json_file: JsonFile = runtests.json_file
setup_test_dir(runtests.test_dir)
result = run_single_test(test_name, runtests)
- with json_file.open('w', encoding='utf-8') as json_fp:
- result.write_json_into(json_fp)
+ 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)
"ARCHFLAGS",
"ARFLAGS",
"AUDIODEV",
+ "BUILDPYTHON",
"CC",
"CFLAGS",
"COLUMNS",
"VIRTUAL_ENV",
"WAYLAND_DISPLAY",
"WINDIR",
+ "_PYTHON_HOSTRUNNER",
"_PYTHON_HOST_PLATFORM",
"_PYTHON_PROJECT_BASE",
"_PYTHON_SYSCONFIGDATA_NAME",
for name, value in os.environ.items():
uname = name.upper()
if (uname in ENV_VARS
- # Copy PYTHON* and LC_* variables
+ # Copy PYTHON* variables like PYTHONPATH
+ # Copy LC_* variables like LC_ALL
or uname.startswith(("PYTHON", "LC_"))
# Visual Studio: VS140COMNTOOLS
or (uname.startswith("VS") and uname.endswith("COMNTOOLS"))):
'CFLAGS',
'CFLAGSFORSHARED',
'CONFIG_ARGS',
+ 'HOSTRUNNER',
'HOST_GNU_TYPE',
'MACHDEP',
'MULTIARCH',
projects / thirdparty / Python / cpython.git / commitdiff
