diff --git a/cuda_pathfinder/cuda/pathfinder/_testing/__init__.py b/cuda_pathfinder/cuda/pathfinder/_testing/__init__.py new file mode 100644 index 0000000000..52a7a9daf0 --- /dev/null +++ b/cuda_pathfinder/cuda/pathfinder/_testing/__init__.py @@ -0,0 +1,2 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 diff --git a/cuda_pathfinder/cuda/pathfinder/_testing/load_nvidia_dynamic_lib_subprocess.py b/cuda_pathfinder/cuda/pathfinder/_testing/load_nvidia_dynamic_lib_subprocess.py new file mode 100644 index 0000000000..dbd5e862f9 --- /dev/null +++ b/cuda_pathfinder/cuda/pathfinder/_testing/load_nvidia_dynamic_lib_subprocess.py @@ -0,0 +1,83 @@ +#!/usr/bin/env python +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +from __future__ import annotations + +import json +import os +import sys +import traceback +from collections.abc import Sequence + +DYNAMIC_LIB_NOT_FOUND_MARKER = "CHILD_LOAD_NVIDIA_DYNAMIC_LIB_HELPER_DYNAMIC_LIB_NOT_FOUND_ERROR:" + + +def _validate_abs_path(abs_path: str) -> None: + assert abs_path, f"empty path: {abs_path=!r}" + assert os.path.isabs(abs_path), f"not absolute: {abs_path=!r}" + assert os.path.isfile(abs_path), f"not a file: {abs_path=!r}" + + +def _load_nvidia_dynamic_lib_for_test(libname: str) -> str: + # Keep imports inside the subprocess body so startup stays focused on the + # code under test rather than the parent test module. + from cuda.pathfinder import load_nvidia_dynamic_lib + from cuda.pathfinder._dynamic_libs.load_dl_common import LoadedDL + from cuda.pathfinder._dynamic_libs.load_nvidia_dynamic_lib import _load_lib_no_cache + from cuda.pathfinder._dynamic_libs.supported_nvidia_libs import ( + SUPPORTED_LINUX_SONAMES, + SUPPORTED_WINDOWS_DLLS, + ) + from cuda.pathfinder._utils.platform_aware import IS_WINDOWS + + def require_abs_path(loaded_dl: LoadedDL) -> str: + abs_path = loaded_dl.abs_path + if not isinstance(abs_path, str): + raise RuntimeError(f"loaded dynamic library is missing abs_path: {loaded_dl!r}") + _validate_abs_path(abs_path) + return abs_path + + loaded_dl_fresh: LoadedDL = load_nvidia_dynamic_lib(libname) + if loaded_dl_fresh.was_already_loaded_from_elsewhere: + raise RuntimeError("loaded_dl_fresh.was_already_loaded_from_elsewhere") + + fresh_abs_path = require_abs_path(loaded_dl_fresh) + assert loaded_dl_fresh.found_via is not None + + loaded_dl_from_cache: LoadedDL = load_nvidia_dynamic_lib(libname) + if loaded_dl_from_cache is not loaded_dl_fresh: + raise RuntimeError("loaded_dl_from_cache is not loaded_dl_fresh") + + loaded_dl_no_cache = _load_lib_no_cache(libname) + no_cache_abs_path = require_abs_path(loaded_dl_no_cache) + supported_libs = SUPPORTED_WINDOWS_DLLS if IS_WINDOWS else SUPPORTED_LINUX_SONAMES + if not loaded_dl_no_cache.was_already_loaded_from_elsewhere and libname in supported_libs: + raise RuntimeError("not loaded_dl_no_cache.was_already_loaded_from_elsewhere") + if not os.path.samefile(no_cache_abs_path, fresh_abs_path): + raise RuntimeError(f"not os.path.samefile({no_cache_abs_path=!r}, {fresh_abs_path=!r})") + return fresh_abs_path + + +def probe_load_nvidia_dynamic_lib_and_print_json(libname: str) -> None: + from cuda.pathfinder import DynamicLibNotFoundError + + try: + abs_path = _load_nvidia_dynamic_lib_for_test(libname) + except DynamicLibNotFoundError: + sys.stdout.write(f"{DYNAMIC_LIB_NOT_FOUND_MARKER}\n") + traceback.print_exc(file=sys.stdout) + return + sys.stdout.write(f"{json.dumps(abs_path)}\n") + + +def main(argv: Sequence[str] | None = None) -> int: + args = list(sys.argv[1:] if argv is None else argv) + if len(args) != 1: + raise SystemExit("Usage: python -m cuda.pathfinder._testing.load_nvidia_dynamic_lib_subprocess ") + probe_load_nvidia_dynamic_lib_and_print_json(args[0]) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/cuda_pathfinder/cuda/pathfinder/_utils/spawned_process_runner.py b/cuda_pathfinder/cuda/pathfinder/_utils/spawned_process_runner.py deleted file mode 100644 index 5ddfa10a59..0000000000 --- a/cuda_pathfinder/cuda/pathfinder/_utils/spawned_process_runner.py +++ /dev/null @@ -1,131 +0,0 @@ -# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. -# SPDX-License-Identifier: Apache-2.0 - -import contextlib -import multiprocessing -import queue # for Empty -import sys -import traceback -from collections.abc import Callable, Sequence -from dataclasses import dataclass -from io import StringIO -from typing import Any - -PROCESS_KILLED = -9 -PROCESS_NO_RESULT = -999 - - -# Similar to https://docs.python.org/3/library/subprocess.html#subprocess.CompletedProcess -# (args, check_returncode() are intentionally not supported here.) -@dataclass -class CompletedProcess: - returncode: int - stdout: str - stderr: str - - -class ChildProcessWrapper: - def __init__( - self, - result_queue: Any, - target: Callable[..., None], - args: Sequence[Any] | None, - kwargs: dict[str, Any] | None, - ) -> None: - self.target = target - self.args = () if args is None else args - self.kwargs = {} if kwargs is None else kwargs - self.result_queue = result_queue - - def __call__(self) -> None: - # Capture stdout/stderr - old_stdout = sys.stdout - old_stderr = sys.stderr - sys.stdout = StringIO() - sys.stderr = StringIO() - - try: - self.target(*self.args, **self.kwargs) - returncode = 0 - except SystemExit as e: # Handle sys.exit() - returncode = e.code if isinstance(e.code, int) else 0 - except BaseException: - traceback.print_exc() - returncode = 1 - finally: - # Collect outputs and restore streams - stdout = sys.stdout.getvalue() - stderr = sys.stderr.getvalue() - sys.stdout = old_stdout - sys.stderr = old_stderr - with contextlib.suppress(Exception): - self.result_queue.put((returncode, stdout, stderr)) - - -def run_in_spawned_child_process( - target: Callable[..., None], - *, - args: Sequence[Any] | None = None, - kwargs: dict[str, Any] | None = None, - timeout: float | None = None, - rethrow: bool = False, -) -> CompletedProcess: - """Run `target` in a spawned child process, capturing stdout/stderr. - - The provided `target` must be defined at the top level of a module, and must - be importable in the spawned child process. Lambdas, closures, or interactively - defined functions (e.g., in Jupyter notebooks) will not work. - - If `rethrow=True` and the child process exits with a nonzero code, - raises ChildProcessError with the captured stderr. - """ - ctx = multiprocessing.get_context("spawn") - result_queue = ctx.Queue() - process = ctx.Process(target=ChildProcessWrapper(result_queue, target, args, kwargs)) - process.start() - - try: - process.join(timeout) - if process.is_alive(): - process.terminate() - process.join() - result = CompletedProcess( - returncode=PROCESS_KILLED, - stdout="", - stderr=f"Process timed out after {timeout} seconds and was terminated.", - ) - else: - try: - returncode, stdout, stderr = result_queue.get(timeout=1.0) - except (queue.Empty, EOFError): - result = CompletedProcess( - returncode=PROCESS_NO_RESULT, - stdout="", - stderr="Process exited or crashed before returning results.", - ) - else: - result = CompletedProcess( - returncode=returncode, - stdout=stdout, - stderr=stderr, - ) - - if rethrow and result.returncode != 0: - raise ChildProcessError( - f"Child process exited with code {result.returncode}.\n" - "--- stderr-from-child-process ---\n" - f"{result.stderr}" - "\n" - ) - - return result - - finally: - try: - result_queue.close() - result_queue.join_thread() - except Exception: # noqa: S110 - pass - if process.is_alive(): - process.kill() - process.join() diff --git a/cuda_pathfinder/tests/child_load_nvidia_dynamic_lib_helper.py b/cuda_pathfinder/tests/child_load_nvidia_dynamic_lib_helper.py index 245552e874..b1dfcadec4 100644 --- a/cuda_pathfinder/tests/child_load_nvidia_dynamic_lib_helper.py +++ b/cuda_pathfinder/tests/child_load_nvidia_dynamic_lib_helper.py @@ -1,17 +1,24 @@ -# SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 -# This helper is factored out so spawned child processes only import this -# lightweight module. That avoids re-importing the test module (and -# repeating its potentially expensive setup) in every child process. +from __future__ import annotations -import json -import os +import subprocess import sys -import traceback +import tempfile +from pathlib import Path +from cuda.pathfinder._testing.load_nvidia_dynamic_lib_subprocess import DYNAMIC_LIB_NOT_FOUND_MARKER -def build_child_process_failed_for_libname_message(libname, result): +LOAD_NVIDIA_DYNAMIC_LIB_SUBPROCESS_MODULE = "cuda.pathfinder._testing.load_nvidia_dynamic_lib_subprocess" +# Launch the child from a neutral directory so `python -m cuda.pathfinder...` +# resolves the installed package instead of the source checkout. In CI the +# checkout does not contain the generated `_version.py` file. +LOAD_NVIDIA_DYNAMIC_LIB_SUBPROCESS_CWD = Path(tempfile.gettempdir()) +PROCESS_TIMED_OUT = -9 + + +def build_child_process_failed_for_libname_message(libname: str, result: subprocess.CompletedProcess[str]) -> str: return ( f"Child process failed for {libname=!r} with exit code {result.returncode}\n" f"--- stdout-from-child-process ---\n{result.stdout}\n" @@ -19,43 +26,29 @@ def build_child_process_failed_for_libname_message(libname, result): ) -def validate_abs_path(abs_path): - assert abs_path, f"empty path: {abs_path=!r}" - assert os.path.isabs(abs_path), f"not absolute: {abs_path=!r}" - assert os.path.isfile(abs_path), f"not a file: {abs_path=!r}" +def child_process_reported_dynamic_lib_not_found(result: subprocess.CompletedProcess[str]) -> bool: + return result.stdout.startswith(DYNAMIC_LIB_NOT_FOUND_MARKER) -def child_process_func(libname): - from cuda.pathfinder import DynamicLibNotFoundError, load_nvidia_dynamic_lib - from cuda.pathfinder._dynamic_libs.load_nvidia_dynamic_lib import _load_lib_no_cache - from cuda.pathfinder._dynamic_libs.supported_nvidia_libs import ( - SUPPORTED_LINUX_SONAMES, - SUPPORTED_WINDOWS_DLLS, - ) - from cuda.pathfinder._utils.platform_aware import IS_WINDOWS - +def run_load_nvidia_dynamic_lib_in_subprocess( + libname: str, + *, + timeout: float, +) -> subprocess.CompletedProcess[str]: + command = [sys.executable, "-m", LOAD_NVIDIA_DYNAMIC_LIB_SUBPROCESS_MODULE, libname] try: - loaded_dl_fresh = load_nvidia_dynamic_lib(libname) - except DynamicLibNotFoundError: - print("CHILD_LOAD_NVIDIA_DYNAMIC_LIB_HELPER_DYNAMIC_LIB_NOT_FOUND_ERROR:") - traceback.print_exc(file=sys.stdout) - return - if loaded_dl_fresh.was_already_loaded_from_elsewhere: - raise RuntimeError("loaded_dl_fresh.was_already_loaded_from_elsewhere") - validate_abs_path(loaded_dl_fresh.abs_path) - assert loaded_dl_fresh.found_via is not None - - loaded_dl_from_cache = load_nvidia_dynamic_lib(libname) - if loaded_dl_from_cache is not loaded_dl_fresh: - raise RuntimeError("loaded_dl_from_cache is not loaded_dl_fresh") - - loaded_dl_no_cache = _load_lib_no_cache(libname) - # check_if_already_loaded_from_elsewhere relies on these: - supported_libs = SUPPORTED_WINDOWS_DLLS if IS_WINDOWS else SUPPORTED_LINUX_SONAMES - if not loaded_dl_no_cache.was_already_loaded_from_elsewhere and libname in supported_libs: - raise RuntimeError("not loaded_dl_no_cache.was_already_loaded_from_elsewhere") - if not os.path.samefile(loaded_dl_no_cache.abs_path, loaded_dl_fresh.abs_path): - raise RuntimeError(f"not os.path.samefile({loaded_dl_no_cache.abs_path=!r}, {loaded_dl_fresh.abs_path=!r})") - validate_abs_path(loaded_dl_no_cache.abs_path) - - print(json.dumps(loaded_dl_fresh.abs_path)) + return subprocess.run( # noqa: S603 - trusted argv: current interpreter + internal test helper module + command, + capture_output=True, + text=True, + timeout=timeout, + check=False, + cwd=LOAD_NVIDIA_DYNAMIC_LIB_SUBPROCESS_CWD, + ) + except subprocess.TimeoutExpired: + return subprocess.CompletedProcess( + args=command, + returncode=PROCESS_TIMED_OUT, + stdout="", + stderr=f"Process timed out after {timeout} seconds and was terminated.", + ) diff --git a/cuda_pathfinder/tests/test_driver_lib_loading.py b/cuda_pathfinder/tests/test_driver_lib_loading.py index d8d463599a..8221a1b9b9 100644 --- a/cuda_pathfinder/tests/test_driver_lib_loading.py +++ b/cuda_pathfinder/tests/test_driver_lib_loading.py @@ -12,7 +12,11 @@ import os import pytest -from child_load_nvidia_dynamic_lib_helper import build_child_process_failed_for_libname_message, child_process_func +from child_load_nvidia_dynamic_lib_helper import ( + build_child_process_failed_for_libname_message, + child_process_reported_dynamic_lib_not_found, + run_load_nvidia_dynamic_lib_in_subprocess, +) from cuda.pathfinder._dynamic_libs.lib_descriptor import LIB_DESCRIPTORS from cuda.pathfinder._dynamic_libs.load_dl_common import DynamicLibNotFoundError, LoadedDL @@ -22,7 +26,6 @@ _load_lib_no_cache, ) from cuda.pathfinder._utils.platform_aware import IS_WINDOWS, quote_for_shell -from cuda.pathfinder._utils.spawned_process_runner import run_in_spawned_child_process STRICTNESS = os.environ.get("CUDA_PATHFINDER_TEST_LOAD_NVIDIA_DYNAMIC_LIB_STRICTNESS", "see_what_works") assert STRICTNESS in ("see_what_works", "all_must_work") @@ -119,19 +122,19 @@ def test_load_lib_no_cache_does_not_dispatch_ctk_lib_to_driver_path(mocker): # --------------------------------------------------------------------------- -# Real loading tests (spawned child process for isolation) +# Real loading tests (dedicated subprocess for isolation) # --------------------------------------------------------------------------- @pytest.mark.parametrize("libname", sorted(_DRIVER_ONLY_LIBNAMES)) def test_real_load_driver_lib(info_summary_append, libname): - """Load a real driver library in a child process. + """Load a real driver library in a dedicated subprocess. This complements the mock tests above: it exercises the actual OS loader path and logs results via INFO for CI/QA inspection. """ timeout = 120 if IS_WINDOWS else 30 - result = run_in_spawned_child_process(child_process_func, args=(libname,), timeout=timeout) + result = run_load_nvidia_dynamic_lib_in_subprocess(libname, timeout=timeout) def raise_child_process_failed(): raise RuntimeError(build_child_process_failed_for_libname_message(libname, result)) @@ -139,7 +142,7 @@ def raise_child_process_failed(): if result.returncode != 0: raise_child_process_failed() assert not result.stderr - if result.stdout.startswith("CHILD_LOAD_NVIDIA_DYNAMIC_LIB_HELPER_DYNAMIC_LIB_NOT_FOUND_ERROR:"): + if child_process_reported_dynamic_lib_not_found(result): if STRICTNESS == "all_must_work": raise_child_process_failed() info_summary_append(f"Not found: {libname=!r}") diff --git a/cuda_pathfinder/tests/test_load_nvidia_dynamic_lib.py b/cuda_pathfinder/tests/test_load_nvidia_dynamic_lib.py index 6de2bff097..ccdd58d5b2 100644 --- a/cuda_pathfinder/tests/test_load_nvidia_dynamic_lib.py +++ b/cuda_pathfinder/tests/test_load_nvidia_dynamic_lib.py @@ -6,14 +6,17 @@ import platform import pytest -from child_load_nvidia_dynamic_lib_helper import build_child_process_failed_for_libname_message, child_process_func +from child_load_nvidia_dynamic_lib_helper import ( + build_child_process_failed_for_libname_message, + child_process_reported_dynamic_lib_not_found, + run_load_nvidia_dynamic_lib_in_subprocess, +) from local_helpers import have_distribution from cuda.pathfinder import DynamicLibNotAvailableError, DynamicLibUnknownError, load_nvidia_dynamic_lib from cuda.pathfinder._dynamic_libs import load_nvidia_dynamic_lib as load_nvidia_dynamic_lib_module from cuda.pathfinder._dynamic_libs import supported_nvidia_libs from cuda.pathfinder._utils.platform_aware import IS_WINDOWS, quote_for_shell -from cuda.pathfinder._utils.spawned_process_runner import run_in_spawned_child_process STRICTNESS = os.environ.get("CUDA_PATHFINDER_TEST_LOAD_NVIDIA_DYNAMIC_LIB_STRICTNESS", "see_what_works") assert STRICTNESS in ("see_what_works", "all_must_work") @@ -107,12 +110,10 @@ def _is_expected_load_nvidia_dynamic_lib_failure(libname): supported_nvidia_libs.SUPPORTED_WINDOWS_DLLS if IS_WINDOWS else supported_nvidia_libs.SUPPORTED_LINUX_SONAMES, ) def test_load_nvidia_dynamic_lib(info_summary_append, libname): - # We intentionally run each dynamic library operation in a child process - # to ensure isolation of global dynamic linking state (e.g., dlopen handles). - # Without child processes, loading/unloading libraries during testing could - # interfere across test cases and lead to nondeterministic or platform-specific failures. + # Use a fresh Python subprocess for each load to isolate global dynamic + # loader state and keep the tests aligned with the canary probe model. timeout = 120 if IS_WINDOWS else 30 - result = run_in_spawned_child_process(child_process_func, args=(libname,), timeout=timeout) + result = run_load_nvidia_dynamic_lib_in_subprocess(libname, timeout=timeout) def raise_child_process_failed(): raise RuntimeError(build_child_process_failed_for_libname_message(libname, result)) @@ -120,7 +121,7 @@ def raise_child_process_failed(): if result.returncode != 0: raise_child_process_failed() assert not result.stderr - if result.stdout.startswith("CHILD_LOAD_NVIDIA_DYNAMIC_LIB_HELPER_DYNAMIC_LIB_NOT_FOUND_ERROR:"): + if child_process_reported_dynamic_lib_not_found(result): if STRICTNESS == "all_must_work" and not _is_expected_load_nvidia_dynamic_lib_failure(libname): raise_child_process_failed() info_summary_append(f"Not found: {libname=!r}") diff --git a/cuda_pathfinder/tests/test_load_nvidia_dynamic_lib_subprocess.py b/cuda_pathfinder/tests/test_load_nvidia_dynamic_lib_subprocess.py new file mode 100644 index 0000000000..df4a3db156 --- /dev/null +++ b/cuda_pathfinder/tests/test_load_nvidia_dynamic_lib_subprocess.py @@ -0,0 +1,71 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +import subprocess +import sys + +from child_load_nvidia_dynamic_lib_helper import ( + LOAD_NVIDIA_DYNAMIC_LIB_SUBPROCESS_CWD, + LOAD_NVIDIA_DYNAMIC_LIB_SUBPROCESS_MODULE, + PROCESS_TIMED_OUT, + run_load_nvidia_dynamic_lib_in_subprocess, +) + +from cuda.pathfinder._dynamic_libs.load_dl_common import DynamicLibNotFoundError +from cuda.pathfinder._testing import load_nvidia_dynamic_lib_subprocess as subprocess_mod + +_HELPER_MODULE = "child_load_nvidia_dynamic_lib_helper" + + +def test_run_load_nvidia_dynamic_lib_in_subprocess_invokes_dedicated_module(mocker): + result = subprocess.CompletedProcess(args=[], returncode=0, stdout='"/tmp/libcudart.so.13"\n', stderr="") + run_mock = mocker.patch(f"{_HELPER_MODULE}.subprocess.run", return_value=result) + + assert run_load_nvidia_dynamic_lib_in_subprocess("cudart", timeout=12.5) is result + run_mock.assert_called_once_with( + [sys.executable, "-m", LOAD_NVIDIA_DYNAMIC_LIB_SUBPROCESS_MODULE, "cudart"], + capture_output=True, + text=True, + timeout=12.5, + check=False, + cwd=LOAD_NVIDIA_DYNAMIC_LIB_SUBPROCESS_CWD, + ) + + +def test_run_load_nvidia_dynamic_lib_in_subprocess_returns_timeout_result(mocker): + mocker.patch( + f"{_HELPER_MODULE}.subprocess.run", + side_effect=subprocess.TimeoutExpired(cmd=["python"], timeout=3.0), + ) + + result = run_load_nvidia_dynamic_lib_in_subprocess("nvvm", timeout=3.0) + + assert result.args == [sys.executable, "-m", LOAD_NVIDIA_DYNAMIC_LIB_SUBPROCESS_MODULE, "nvvm"] + assert result.returncode == PROCESS_TIMED_OUT + assert result.stdout == "" + assert result.stderr == "Process timed out after 3.0 seconds and was terminated." + + +def test_probe_load_nvidia_dynamic_lib_and_print_json(mocker, capsys): + mocker.patch.object(subprocess_mod, "_load_nvidia_dynamic_lib_for_test", return_value="/usr/lib/libcudart.so.13") + + subprocess_mod.probe_load_nvidia_dynamic_lib_and_print_json("cudart") + + captured = capsys.readouterr() + assert captured.out == '"/usr/lib/libcudart.so.13"\n' + assert captured.err == "" + + +def test_probe_load_nvidia_dynamic_lib_and_prints_not_found_traceback(mocker, capsys): + mocker.patch.object( + subprocess_mod, + "_load_nvidia_dynamic_lib_for_test", + side_effect=DynamicLibNotFoundError("not found"), + ) + + subprocess_mod.probe_load_nvidia_dynamic_lib_and_print_json("cudart") + + captured = capsys.readouterr() + assert captured.out.startswith(f"{subprocess_mod.DYNAMIC_LIB_NOT_FOUND_MARKER}\nTraceback") + assert "DynamicLibNotFoundError: not found" in captured.out + assert captured.err == "" diff --git a/cuda_pathfinder/tests/test_spawned_process_runner.py b/cuda_pathfinder/tests/test_spawned_process_runner.py deleted file mode 100644 index 83ec8aaad7..0000000000 --- a/cuda_pathfinder/tests/test_spawned_process_runner.py +++ /dev/null @@ -1,22 +0,0 @@ -# SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved. -# SPDX-License-Identifier: Apache-2.0 - -# Note: This only covers what is not covered already in test_nvidia_dynamic_libs_load_lib.py - -import pytest - -from cuda.pathfinder._utils.spawned_process_runner import run_in_spawned_child_process - - -def child_crashes(): - raise RuntimeError("this is an intentional failure") - - -def test_rethrow_child_exception(): - with pytest.raises(ChildProcessError) as excinfo: - run_in_spawned_child_process(child_crashes, rethrow=True) - - msg = str(excinfo.value) - assert "Child process exited with code 1" in msg - assert "this is an intentional failure" in msg - assert "--- stderr-from-child-process ---" in msg