Function _run_fork_script

tests/test_read_write_fork.py:307–327

(script: str, arguments: list[str], *, timeout: float)

def _run_fork_script(script: str, arguments: list[str], *, timeout: float) -> tuple[int, str, str]:
    process = subprocess.Popen(  # executes this test's fixed interpreter and source
        [sys.executable, "-c", script, *arguments],
        start_new_session=True,
        stderr=subprocess.PIPE,
        stdout=subprocess.PIPE,
        text=True,
    )
    try:
        process_output, process_error = process.communicate(timeout=timeout)
    except subprocess.TimeoutExpired as error:
        with contextlib.suppress(ProcessLookupError):
            if sys.platform == "win32":  # pragma: win32 cover
                process.kill()
            else:  # pragma: win32 no cover
                os.killpg(process.pid, signal.SIGKILL)
        process_output, process_error = process.communicate()
        msg = f"fork script exceeded {timeout} seconds: stdout={process_output!r}, stderr={process_error!r}"
        raise AssertionError(msg) from error
    assert process.returncode is not None  # pragma: needs fork
    return process.returncode, process_output, process_error  # pragma: needs fork


def _dropped_instances_script() -> str:  # pragma: needs fd-directory