diff --git a/cuda_bindings/tests/test_cuda.py b/cuda_bindings/tests/test_cuda.py index 7bef2b844aa..5f7e67fb262 100644 --- a/cuda_bindings/tests/test_cuda.py +++ b/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(): diff --git a/cuda_core/tests/graph/test_graph_definition_lifetime.py b/cuda_core/tests/graph/test_graph_definition_lifetime.py index 93c1453753d..a31482c5769 100644 --- a/cuda_core/tests/graph/test_graph_definition_lifetime.py +++ b/cuda_core/tests/graph/test_graph_definition_lifetime.py @@ -5,8 +5,6 @@ import ctypes import gc -import subprocess -import sys import textwrap import threading import time @@ -14,6 +12,7 @@ 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): diff --git a/cuda_core/tests/test_rlcompleter_patch.py b/cuda_core/tests/test_rlcompleter_patch.py index 50283e62a31..92a77efee8e 100644 --- a/cuda_core/tests/test_rlcompleter_patch.py +++ b/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 diff --git a/cuda_python_test_helpers/cuda_python_test_helpers/subprocess_runner.py b/cuda_python_test_helpers/cuda_python_test_helpers/subprocess_runner.py new file mode 100644 index 00000000000..5109980cfb0 --- /dev/null +++ b/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