"""Child side of the test runner's fork server (run_tests_launch.py; HERMES_TEST_FORKSERVER=1, Linux). A fresh ``python -m pytest `` spends about half a second starting the interpreter and importing pytest and the third-party libraries every test file loads. The runner's fork server pays that once: it imports those modules, then forks one child per test file. Each child gets the state a fresh process would have: its own session and process group, the per-file environment, the ``python -m pytest`` sys.path and argv, stdout and stderr in one file, and the exit code pytest returns. Product code and tests/conftest.py are never preloaded, so every file still imports them fresh. One difference remains: /proc//environ shows the environment a process was exec'd with, and a child here was never exec'd. A daemon a test forks without exec, in a session of its own, therefore carries neither the per-file temp root the runner's leftover sweep looks for nor the file's process group, and outlives the file. Daemons started through exec (subprocess) are still found. This module must stay importable on its own (the fork server unpickles ``run`` by name) and import nothing from the product. """ from __future__ import annotations import os import sys import tempfile def run(*, argv: list[str], cwd: str, env: dict[str, str], sys_path: list[str], out_path: str) -> None: """Run ``pytest.main(argv[1:])`` as ``python -m pytest`` would, then exit with its code.""" # First, before anything can spawn: a session of our own, so the runner's # killpg(pid) takes out every process this file starts and nothing else. os.setsid() # windows-footgun: ok — the fork server is Linux-only fd = os.open(out_path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) os.dup2(fd, 1) os.dup2(fd, 2) os.close(fd) os.chdir(cwd) os.environ.clear() os.environ.update(env) # gettempdir() caches its answer, and a preloaded module may have asked # in the server; this file's TMPDIR must win. tempfile.tempdir = None sys.path[:] = sys_path sys.argv[:] = argv _become_main_process() import pytest # multiprocessing's fork-server child runs atexit handlers on the way out, # as a real interpreter exit would; sys.exit carries pytest's code. sys.exit(int(pytest.main(argv[1:]))) def _become_main_process() -> None: """Drop what makes this process the fork server's child to multiprocessing. A test's own Process, Pool or spawn context must see what it sees under ``python -m pytest``: a main process with no parent, its own temp dir, and pytest as ``__main__`` (a spawned grandchild re-imports ``__main__``). """ import multiprocessing.process as mp_process import pytest.__main__ current = mp_process.current_process() current.name = "MainProcess" current._config.pop("tempdir", None) # the runner's /tmp/pymp-*; a new one lands in TMPDIR mp_process._parent_process = None # As in any process that imports multiprocessing, __mp_main__ is __main__. sys.modules["__main__"] = sys.modules["__mp_main__"] = pytest.__main__