test: share one helper for isolated-subprocess test snippets

cuda_bindings/tests/test_cuda.py

@@ -2,15 +2,13 @@
 # SPDX-License-Identifier: Apache-2.0
 
 import ctypes
-import os.path
 import shutil
-import subprocess
-import sys
 import textwrap
 
 import numpy as np
 import pytest
 from cuda_python_test_helpers.mempool import xfail_if_mempool_oom
+from cuda_python_test_helpers.subprocess_runner import run_python_snippet
 
 import cuda.bindings.driver as cuda
 import cuda.bindings.runtime as cudart
@@ -1289,10 +1287,7 @@ def test_array_setter_no_double_free_after_clearing_with_empty_list():
         params.attrs = [cuda.CUlaunchAttribute() for _ in range(8)]
         """
     )
-    proc = subprocess.run([sys.executable, "-c", code], capture_output=True, cwd=os.path.dirname(__file__))  # noqa: S603
-    assert proc.returncode == 0, (
-        f"reproducer subprocess exited with code {proc.returncode}; stderr: {proc.stderr.decode(errors='replace')}"
-    )
+    run_python_snippet(code)
 
 
 def test_dealloc_clears_array_field_in_external_struct():

cuda_core/tests/graph/test_graph_definition_lifetime.py

@@ -5,15 +5,14 @@
 
 import ctypes
 import gc
-import subprocess
-import sys
 import textwrap
 import threading
 import time
 import weakref
 
 import pytest
 from conftest import xfail_on_graph_mempool_oom
+from cuda_python_test_helpers.subprocess_runner import run_python_snippet
 from helpers.graph_kernels import compile_common_kernels
 from helpers.misc import try_create_condition
 
@@ -704,7 +703,7 @@ def test_user_object_cleanup_is_coalesced_on_python_thread(init_cuda):
 
 
 @pytest.mark.agent_authored(model="gpt-5.6")
-def test_pending_call_queue_saturation_preserves_cleanup(tmp_path):
+def test_pending_call_queue_saturation_preserves_cleanup():
     """A full CPython queue neither strands nor mis-threads cleanup."""
     code = f"timeout = {_FINALIZE_TIMEOUT!r}\n" + textwrap.dedent(
         """
@@ -786,19 +785,11 @@ def fill_queue_and_destroy():
         assert set(finalized_threads) == {main_thread}
         """
     )
-    result = subprocess.run(  # noqa: S603 - controlled interpreter probe
-        [sys.executable, "-c", code],
-        capture_output=True,
-        text=True,
-        timeout=60,
-        # Isolate the process-global pending-call queue from parallel tests.
-        cwd=tmp_path,
-    )
-    assert result.returncode == 0, result.stderr
+    run_python_snippet(code, timeout=60)
 
 
 @pytest.mark.agent_authored(model="gpt-5.6")
-def test_pending_cleanup_is_safe_during_python_shutdown(init_cuda, tmp_path):
+def test_pending_cleanup_is_safe_during_python_shutdown(init_cuda):
     """Outstanding graph attachments neither call Python nor hang at shutdown."""
     code = textwrap.dedent(
         """
@@ -815,15 +806,7 @@ def __call__(self):
         graph.callback(Callback())
         """
     )
-    result = subprocess.run(  # noqa: S603 - controlled interpreter probe
-        [sys.executable, "-c", code],
-        capture_output=True,
-        text=True,
-        timeout=20,
-        # Avoid shadowing the installed package with cuda_core/cuda/core.
-        cwd=tmp_path,
-    )
-    assert result.returncode == 0, result.stderr
+    run_python_snippet(code, timeout=20)
 
 
 def test_python_callable_callback_survives_del(init_cuda):

cuda_core/tests/test_rlcompleter_patch.py

@@ -16,13 +16,12 @@
 `CUDA_CORE_DONT_FIX_TAB_COMPLETION`.
 """
 
-import os
 import subprocess
 import sys
-import tempfile
 import textwrap
 
 import pytest
+from cuda_python_test_helpers.subprocess_runner import run_python_snippet
 
 from cuda.core import Device
 
@@ -59,30 +58,14 @@ def _gpu_with_mempool_or_skip():
 
 
 def _run_probe(*, pythoninspect: bool, opt_out: bool = False) -> subprocess.CompletedProcess:
-    env = os.environ.copy()
-    # Don't let parent-environment values bleed into the subprocess.
-    env.pop("CUDA_CORE_DONT_FIX_TAB_COMPLETION", None)
-    # Drop PYTHONPATH so the subprocess can't find a source-tree cuda.core
-    # via an inherited path entry; we want it to import the installed wheel.
-    env.pop("PYTHONPATH", None)
-    if opt_out:
-        env["CUDA_CORE_DONT_FIX_TAB_COMPLETION"] = "1"
-    # `python -c` puts the parent's CWD at the head of sys.path. If pytest is
-    # run from `cuda_core/` (which contains a `cuda/core/` source tree), that
-    # source tree shadows the installed package. Run the subprocess from a
-    # neutral temp dir to avoid this.
-    with tempfile.TemporaryDirectory() as tmpdir:
-        return subprocess.run(  # noqa: S603
-            [sys.executable, "-c", _PROBE_SCRIPT],
-            capture_output=True,
-            text=True,
-            env=env,
-            check=False,
-            # PYTHONINSPECT keeps the interpreter alive after `-c`; close stdin
-            # so the implicit REPL exits immediately.
-            stdin=subprocess.DEVNULL,
-            cwd=tmpdir,
-        )
+    return run_python_snippet(
+        _PROBE_SCRIPT,
+        # The child must import the installed wheel, not a source tree
+        # reachable through an inherited PYTHONPATH.
+        unset_env=("CUDA_CORE_DONT_FIX_TAB_COMPLETION", "PYTHONPATH"),
+        extra_env={"CUDA_CORE_DONT_FIX_TAB_COMPLETION": "1"} if opt_out else None,
+        check=False,
+    )
 
 
 def test_patched_completion_succeeds_on_non_ipc_resource():
@@ -148,20 +131,9 @@ def test_opt_out_env_var_values(value, expect_patched):
     `ValueError: invalid literal for int() with base 10: ''` out of
     `cuda/core/__init__.py` and made the package unimportable.
     """
-    env = os.environ.copy()
-    env.pop("PYTHONPATH", None)
-    env["CUDA_CORE_DONT_FIX_TAB_COMPLETION"] = value
-    # Run from a neutral directory so a source tree next to the test run
-    # cannot shadow the installed package (see _run_probe).
-    with tempfile.TemporaryDirectory() as tmpdir:
-        result = subprocess.run(  # noqa: S603
-            [sys.executable, "-c", _OPT_OUT_PROBE_SCRIPT],
-            capture_output=True,
-            text=True,
-            env=env,
-            check=False,
-            stdin=subprocess.DEVNULL,
-            cwd=tmpdir,
-        )
-    assert result.returncode == 0, f"stderr: {result.stderr}\nstdout: {result.stdout}"
+    result = run_python_snippet(
+        _OPT_OUT_PROBE_SCRIPT,
+        unset_env=("PYTHONPATH",),
+        extra_env={"CUDA_CORE_DONT_FIX_TAB_COMPLETION": value},
+    )
     assert result.stdout.strip() == f"patched: {expect_patched}", result.stdout

cuda_python_test_helpers/cuda_python_test_helpers/subprocess_runner.py

@@ -0,0 +1,73 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Shared runner for tests that must execute a snippet in a fresh interpreter."""
+
+from __future__ import annotations
+
+import os
+import subprocess
+import sys
+import tempfile
+from collections.abc import Mapping
+
+__all__ = ["run_python_snippet"]
+
+
+def run_python_snippet(
+    code: str,
+    *,
+    cwd: str | os.PathLike[str] | None = None,
+    timeout: float | None = None,
+    extra_env: Mapping[str, str] | None = None,
+    unset_env: tuple[str, ...] = (),
+    check: bool = True,
+) -> subprocess.CompletedProcess[str]:
+    """Run ``code`` with ``sys.executable -c`` and return the completed process.
+
+    Output is captured as text. ``stdin`` is closed, so a child left
+    interactive by ``PYTHONINSPECT`` exits instead of waiting.
+
+    Args:
+        code: Python source for the child interpreter.
+        cwd: Directory to run from. When omitted, an empty temporary directory
+            is created and cleaned up after the child exits.
+        timeout: Seconds before ``subprocess.TimeoutExpired`` is raised.
+        extra_env: Environment entries to set on top of the parent environment.
+        unset_env: Environment variable names to remove from the child. Applied
+            before ``extra_env``.
+        check: Fail the calling test if the child exits non-zero, quoting the
+            exit code and both streams. Pass ``False`` when the caller asserts
+            on the exit code itself.
+
+    Returns:
+        The completed process, with ``stdout`` and ``stderr`` as ``str``.
+    """
+    env = os.environ.copy()
+    for name in unset_env:
+        env.pop(name, None)
+    if extra_env:
+        env.update(extra_env)
+
+    def run(cwd: str | os.PathLike[str]) -> subprocess.CompletedProcess[str]:
+        return subprocess.run(  # noqa: S603
+            [sys.executable, "-c", code],
+            capture_output=True,
+            text=True,
+            env=env,
+            check=False,
+            cwd=os.fspath(cwd),
+            timeout=timeout,
+            stdin=subprocess.DEVNULL,
+        )
+
+    if cwd is None:
+        with tempfile.TemporaryDirectory() as tmpdir:
+            result = run(tmpdir)
+    else:
+        result = run(cwd)
+    if check:
+        assert result.returncode == 0, (
+            f"subprocess exited with code {result.returncode}\nstdout: {result.stdout}\nstderr: {result.stderr}"
+        )
+    return result