diff --git a/cuda_pathfinder/cuda/pathfinder/__init__.py b/cuda_pathfinder/cuda/pathfinder/__init__.py index dc818dfd08..57790a61c3 100644 --- a/cuda_pathfinder/cuda/pathfinder/__init__.py +++ b/cuda_pathfinder/cuda/pathfinder/__init__.py @@ -11,6 +11,9 @@ find_nvidia_binary_utility as find_nvidia_binary_utility, ) from cuda.pathfinder._binaries.supported_nvidia_binaries import SUPPORTED_BINARIES as _SUPPORTED_BINARIES +from cuda.pathfinder._dynamic_libs.find_nvidia_dynamic_lib import ( + find_nvidia_dynamic_lib as find_nvidia_dynamic_lib, +) from cuda.pathfinder._dynamic_libs.load_dl_common import ( DynamicLibNotAvailableError as DynamicLibNotAvailableError, ) diff --git a/cuda_pathfinder/cuda/pathfinder/_dynamic_libs/dynamic_lib_subprocess.py b/cuda_pathfinder/cuda/pathfinder/_dynamic_libs/dynamic_lib_subprocess.py index ba5d05242f..4e48295226 100644 --- a/cuda_pathfinder/cuda/pathfinder/_dynamic_libs/dynamic_lib_subprocess.py +++ b/cuda_pathfinder/cuda/pathfinder/_dynamic_libs/dynamic_lib_subprocess.py @@ -13,6 +13,7 @@ from cuda.pathfinder._dynamic_libs.platform_loader import LOADER from cuda.pathfinder._dynamic_libs.subprocess_protocol import ( MODE_CANARY, + MODE_FIND, MODE_LOAD, STATUS_NOT_FOUND, STATUS_OK, @@ -20,12 +21,9 @@ format_dynamic_lib_subprocess_payload, ) -# NOTE: The main entrypoint (below) serves both production (canary probe) -# and tests (full loader). Keeping them together ensures a single subprocess -# protocol and CLI surface, so the test subprocess stays aligned with the -# production flow while avoiding a separate test-only module. -# Any production-code impact is negligible since the extra logic only runs -# in the subprocess entrypoint and only in test mode. +# The main entrypoint serves three modes — canary probe, find-without-load, +# and (test-only) full-loader exercise — behind a single subprocess protocol +# so test and production flows stay aligned. def _probe_canary_abs_path(libname: str) -> str | None: @@ -94,6 +92,22 @@ def probe_dynamic_lib_and_print_json(libname: str, mode: str) -> None: print(format_dynamic_lib_subprocess_payload(status, abs_path)) return + if mode == MODE_FIND: + from cuda.pathfinder import load_nvidia_dynamic_lib + + try: + loaded = load_nvidia_dynamic_lib(libname) + except DynamicLibNotFoundError as exc: + error = {"type": exc.__class__.__name__, "message": str(exc)} + print(format_dynamic_lib_subprocess_payload(STATUS_NOT_FOUND, None, error=error)) + return + abs_path = loaded.abs_path + if not isinstance(abs_path, str): + raise RuntimeError(f"loaded.abs_path is not a string: {abs_path!r}") + _validate_abs_path(abs_path) + print(format_dynamic_lib_subprocess_payload(STATUS_OK, abs_path)) + return + if mode == MODE_LOAD: # Test-only path: exercises full loader behavior in isolation. try: diff --git a/cuda_pathfinder/cuda/pathfinder/_dynamic_libs/find_nvidia_dynamic_lib.py b/cuda_pathfinder/cuda/pathfinder/_dynamic_libs/find_nvidia_dynamic_lib.py new file mode 100644 index 0000000000..0fd00d115b --- /dev/null +++ b/cuda_pathfinder/cuda/pathfinder/_dynamic_libs/find_nvidia_dynamic_lib.py @@ -0,0 +1,88 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Locate an NVIDIA dynamic library on disk without loading it in this process. + +Resolution is delegated to ``load_nvidia_dynamic_lib`` running in a fresh +Python subprocess. The full loader runs (including ``dlopen`` / +``LoadLibraryExW``) but only inside the child, so the caller's process is left +untouched. +""" + +from __future__ import annotations + +import functools + +from cuda.pathfinder._dynamic_libs import load_nvidia_dynamic_lib as _load_module +from cuda.pathfinder._dynamic_libs.load_dl_common import ( + DynamicLibNotAvailableError, + DynamicLibNotFoundError, + DynamicLibUnknownError, +) +from cuda.pathfinder._dynamic_libs.subprocess_protocol import ( + MODE_FIND, + STATUS_OK, + run_dynamic_lib_subprocess, +) +from cuda.pathfinder._utils.platform_aware import IS_WINDOWS + +# The subprocess runs the full loader (site-packages / conda / CUDA_PATH / +# canary cascade), which can be substantially slower than a single canary +# probe. Bound it so a wedged child cannot hang the caller indefinitely. +_FIND_SUBPROCESS_TIMEOUT_SECONDS = 120.0 if IS_WINDOWS else 30.0 + + +@functools.cache +def find_nvidia_dynamic_lib(libname: str) -> str: + """Return the absolute path to an NVIDIA dynamic library without loading it. + + Resolution is performed by running :func:`load_nvidia_dynamic_lib` in a + fresh Python subprocess and reporting back the resolved absolute path. + The caller's process does **not** dlopen / LoadLibrary the library. + + Args: + libname: Short name of the library (e.g., ``"cufile"``, + ``"nvJitLink"``, ``"cudart"``). + + Returns: + The absolute path the loader would have used in the caller's process. + + Raises: + DynamicLibUnknownError: If ``libname`` is not a recognized library. + DynamicLibNotAvailableError: If ``libname`` is recognized but not + supported on this platform. + DynamicLibNotFoundError: If the library cannot be located. + + Notes: + Because resolution happens in a separate process, results may differ + from an in-process ``load_nvidia_dynamic_lib`` if the caller's process + has DSOs loaded with custom ``RPATH`` entries or has already loaded a matching + library by some other mechanism. The intent is to report the path the + loader would pick when not influenced by other DSOs in the caller. + """ + # Indirect attribute access (not `from ... import`) so tests can + # monkeypatch the source-of-truth tables in `load_nvidia_dynamic_lib`. + if libname not in _load_module._ALL_KNOWN_LIBNAMES: + raise DynamicLibUnknownError( + f"Unknown library name: {libname!r}. Known names: {sorted(_load_module._ALL_KNOWN_LIBNAMES)}" + ) + if libname not in _load_module._ALL_SUPPORTED_LIBNAMES: + raise DynamicLibNotAvailableError( + f"Library name {libname!r} is known but not available on {_load_module._PLATFORM_NAME}. " + f"Supported names on {_load_module._PLATFORM_NAME}: {sorted(_load_module._ALL_SUPPORTED_LIBNAMES)}" + ) + + payload = run_dynamic_lib_subprocess( + MODE_FIND, + libname, + timeout=_FIND_SUBPROCESS_TIMEOUT_SECONDS, + error_label=f"find_nvidia_dynamic_lib subprocess for {libname!r}", + ) + if payload.status == STATUS_OK: + abs_path: str | None = payload.abs_path + assert abs_path is not None + return abs_path + + error = payload.error + message = error["message"] if error and "message" in error else f"could not locate {libname!r}" + raise DynamicLibNotFoundError(message) diff --git a/cuda_pathfinder/cuda/pathfinder/_dynamic_libs/load_nvidia_dynamic_lib.py b/cuda_pathfinder/cuda/pathfinder/_dynamic_libs/load_nvidia_dynamic_lib.py index a7a8965d2e..c935649c57 100644 --- a/cuda_pathfinder/cuda/pathfinder/_dynamic_libs/load_nvidia_dynamic_lib.py +++ b/cuda_pathfinder/cuda/pathfinder/_dynamic_libs/load_nvidia_dynamic_lib.py @@ -5,7 +5,6 @@ import functools import struct -import subprocess import sys from typing import TYPE_CHECKING @@ -27,12 +26,9 @@ run_find_steps, ) from cuda.pathfinder._dynamic_libs.subprocess_protocol import ( - DYNAMIC_LIB_SUBPROCESS_CWD, MODE_CANARY, STATUS_OK, - DynamicLibSubprocessPayload, - build_dynamic_lib_subprocess_command, - parse_dynamic_lib_subprocess_payload, + run_dynamic_lib_subprocess, ) from cuda.pathfinder._utils.platform_aware import IS_WINDOWS @@ -74,30 +70,6 @@ def _load_driver_lib_no_cache(desc: LibDescriptor) -> LoadedDL: ) -def _coerce_subprocess_output(output: str | bytes | None) -> str: - if isinstance(output, bytes): - return output.decode(errors="replace") - return "" if output is None else output - - -def _raise_canary_probe_child_process_error( - *, - returncode: int | None = None, - timeout: float | None = None, - stderr: str | bytes | None = None, -) -> None: - if timeout is None: - error_line = f"Canary probe child process exited with code {returncode}." - else: - error_line = f"Canary probe child process timed out after {timeout} seconds." - raise ChildProcessError( - f"{error_line}\n" - "--- stderr-from-child-process ---\n" - f"{_coerce_subprocess_output(stderr)}" - "\n" - ) - - @functools.cache def _resolve_system_loaded_abs_path_in_subprocess( libname: str, @@ -105,30 +77,10 @@ def _resolve_system_loaded_abs_path_in_subprocess( timeout: float = _CANARY_PROBE_TIMEOUT_SECONDS, ) -> str | None: """Resolve a canary library's absolute path in a fresh Python subprocess.""" - try: - result = subprocess.run( # noqa: S603 - trusted argv: current interpreter + internal probe module - build_dynamic_lib_subprocess_command(MODE_CANARY, libname), - capture_output=True, - text=True, - timeout=timeout, - check=False, - cwd=DYNAMIC_LIB_SUBPROCESS_CWD, - ) - except subprocess.TimeoutExpired as exc: - _raise_canary_probe_child_process_error(timeout=exc.timeout, stderr=exc.stderr) - - if result.returncode != 0: - _raise_canary_probe_child_process_error(returncode=result.returncode, stderr=result.stderr) - - payload: DynamicLibSubprocessPayload = parse_dynamic_lib_subprocess_payload( - result.stdout, - libname=libname, - error_label="Canary probe child process", + payload = run_dynamic_lib_subprocess( + MODE_CANARY, libname, timeout=timeout, error_label="Canary probe child process" ) - abs_path: str | None = payload.abs_path - if payload.status == STATUS_OK: - return abs_path - return None + return payload.abs_path if payload.status == STATUS_OK else None def _loadable_via_canary_subprocess(libname: str, *, timeout: float = _CANARY_PROBE_TIMEOUT_SECONDS) -> bool: diff --git a/cuda_pathfinder/cuda/pathfinder/_dynamic_libs/subprocess_protocol.py b/cuda_pathfinder/cuda/pathfinder/_dynamic_libs/subprocess_protocol.py index 4404d667f5..f405098f63 100644 --- a/cuda_pathfinder/cuda/pathfinder/_dynamic_libs/subprocess_protocol.py +++ b/cuda_pathfinder/cuda/pathfinder/_dynamic_libs/subprocess_protocol.py @@ -4,14 +4,16 @@ from __future__ import annotations import json +import subprocess import sys from dataclasses import dataclass from pathlib import Path -from typing import Literal +from typing import Literal, NoReturn MODE_CANARY: Literal["canary"] = "canary" MODE_LOAD: Literal["load"] = "load" -VALID_MODES: tuple[Literal["canary"], Literal["load"]] = (MODE_CANARY, MODE_LOAD) +MODE_FIND: Literal["find"] = "find" +VALID_MODES: tuple[Literal["canary"], Literal["load"], Literal["find"]] = (MODE_CANARY, MODE_LOAD, MODE_FIND) STATUS_OK: Literal["ok"] = "ok" STATUS_NOT_FOUND: Literal["not-found"] = "not-found" @@ -24,6 +26,7 @@ class DynamicLibSubprocessPayload: status: Literal["ok", "not-found"] abs_path: str | None + error: dict[str, str] | None = None def format_dynamic_lib_subprocess_payload( @@ -60,12 +63,78 @@ def parse_dynamic_lib_subprocess_payload( raise RuntimeError(f"{error_label} emitted unexpected payload for {libname!r}: {payload!r}") status = payload.get("status") abs_path = payload.get("abs_path") - if status == STATUS_OK: - if not isinstance(abs_path, str): - raise RuntimeError(f"{error_label} emitted unexpected payload for {libname!r}: {payload!r}") - return DynamicLibSubprocessPayload(status=STATUS_OK, abs_path=abs_path) - if status == STATUS_NOT_FOUND: - if abs_path is not None: - raise RuntimeError(f"{error_label} emitted unexpected payload for {libname!r}: {payload!r}") - return DynamicLibSubprocessPayload(status=STATUS_NOT_FOUND, abs_path=None) - raise RuntimeError(f"{error_label} emitted unexpected payload for {libname!r}: {payload!r}") + error = payload.get("error") + + def reject() -> NoReturn: + raise RuntimeError(f"{error_label} emitted unexpected payload for {libname!r}: {payload!r}") + + if error is not None and not ( + isinstance(error, dict) and all(isinstance(k, str) and isinstance(v, str) for k, v in error.items()) + ): + reject() + # STATUS_OK carries a path and never an error; STATUS_NOT_FOUND has no + # path but may carry a structured error from the child loader. + if status == STATUS_OK and isinstance(abs_path, str) and error is None: + return DynamicLibSubprocessPayload(status=STATUS_OK, abs_path=abs_path, error=None) + if status == STATUS_NOT_FOUND and abs_path is None: + return DynamicLibSubprocessPayload(status=STATUS_NOT_FOUND, abs_path=None, error=error) + reject() + + +def _coerce_subprocess_output(output: str | bytes | None) -> str: + if isinstance(output, bytes): + return output.decode(errors="replace") + return "" if output is None else output + + +def raise_subprocess_child_process_error( + error_label: str, + *, + returncode: int | None = None, + timeout: float | None = None, + stdout: str | bytes | None = None, + stderr: str | bytes | None = None, +) -> NoReturn: + if timeout is not None: + first_line = f"{error_label} timed out after {timeout} seconds." + else: + first_line = f"{error_label} exited with code {returncode}." + raise ChildProcessError( + f"{first_line}\n" + "--- stdout-from-child-process ---\n" + f"{_coerce_subprocess_output(stdout)}\n" + "--- stderr-from-child-process ---\n" + f"{_coerce_subprocess_output(stderr)}\n" + ) + + +def run_dynamic_lib_subprocess( + mode: str, + libname: str, + *, + timeout: float, + error_label: str, +) -> DynamicLibSubprocessPayload: + """Run the dynamic-lib subprocess and parse its payload. + + Raises ``ChildProcessError`` if the child times out or exits non-zero; + otherwise returns the parsed payload (which may itself be ``STATUS_NOT_FOUND``). + """ + try: + result = subprocess.run( # noqa: S603 - trusted argv: current interpreter + internal probe module + build_dynamic_lib_subprocess_command(mode, libname), + capture_output=True, + text=True, + timeout=timeout, + check=False, + cwd=DYNAMIC_LIB_SUBPROCESS_CWD, + ) + except subprocess.TimeoutExpired as exc: + raise_subprocess_child_process_error(error_label, timeout=exc.timeout, stdout=exc.stdout, stderr=exc.stderr) + + if result.returncode != 0: + raise_subprocess_child_process_error( + error_label, returncode=result.returncode, stdout=result.stdout, stderr=result.stderr + ) + + return parse_dynamic_lib_subprocess_payload(result.stdout, libname=libname, error_label=error_label) diff --git a/cuda_pathfinder/docs/source/api.rst b/cuda_pathfinder/docs/source/api.rst index e49478c09e..635cbeb5a1 100644 --- a/cuda_pathfinder/docs/source/api.rst +++ b/cuda_pathfinder/docs/source/api.rst @@ -20,6 +20,7 @@ CUDA bitcode and static libraries. SUPPORTED_NVIDIA_LIBNAMES load_nvidia_dynamic_lib + find_nvidia_dynamic_lib LoadedDL DynamicLibNotFoundError DynamicLibUnknownError diff --git a/cuda_pathfinder/tests/test_ctk_root_discovery.py b/cuda_pathfinder/tests/test_ctk_root_discovery.py index 19cbe847d2..b6c156a668 100644 --- a/cuda_pathfinder/tests/test_ctk_root_discovery.py +++ b/cuda_pathfinder/tests/test_ctk_root_discovery.py @@ -33,6 +33,7 @@ from cuda.pathfinder._utils.platform_aware import IS_WINDOWS _MODULE = "cuda.pathfinder._dynamic_libs.load_nvidia_dynamic_lib" +_PROTOCOL_MODULE = "cuda.pathfinder._dynamic_libs.subprocess_protocol" _STEPS_MODULE = "cuda.pathfinder._dynamic_libs.search_steps" _PACKAGE_ROOT = DYNAMIC_LIB_SUBPROCESS_CWD @@ -201,7 +202,7 @@ def test_subprocess_probe_returns_abs_path_on_string_payload(mocker): stdout='{"status": "ok", "abs_path": "/usr/local/cuda/lib64/libcudart.so.13"}\n', stderr="", ) - run_mock = mocker.patch(f"{_MODULE}.subprocess.run", return_value=result) + run_mock = mocker.patch(f"{_PROTOCOL_MODULE}.subprocess.run", return_value=result) assert _resolve_system_loaded_abs_path_in_subprocess("cudart") == "/usr/local/cuda/lib64/libcudart.so.13" run_mock.assert_called_once_with( @@ -221,14 +222,14 @@ def test_subprocess_probe_returns_none_on_null_payload(mocker): stdout='{"status": "not-found", "abs_path": null}\n', stderr="", ) - mocker.patch(f"{_MODULE}.subprocess.run", return_value=result) + mocker.patch(f"{_PROTOCOL_MODULE}.subprocess.run", return_value=result) assert _resolve_system_loaded_abs_path_in_subprocess("cudart") is None def test_subprocess_probe_raises_on_child_failure(mocker): result = subprocess.CompletedProcess(args=[], returncode=1, stdout="", stderr="child failed\n") - mocker.patch(f"{_MODULE}.subprocess.run", return_value=result) + mocker.patch(f"{_PROTOCOL_MODULE}.subprocess.run", return_value=result) with pytest.raises(ChildProcessError, match="child failed"): _resolve_system_loaded_abs_path_in_subprocess("cudart") @@ -236,7 +237,7 @@ def test_subprocess_probe_raises_on_child_failure(mocker): def test_subprocess_probe_raises_on_timeout(mocker): mocker.patch( - f"{_MODULE}.subprocess.run", + f"{_PROTOCOL_MODULE}.subprocess.run", side_effect=subprocess.TimeoutExpired(cmd=["python"], timeout=10.0, stderr="probe hung\n"), ) with pytest.raises(ChildProcessError, match="timed out after 10.0 seconds"): @@ -245,7 +246,7 @@ def test_subprocess_probe_raises_on_timeout(mocker): def test_subprocess_probe_raises_on_empty_stdout(mocker): result = subprocess.CompletedProcess(args=[], returncode=0, stdout=" \n \n", stderr="") - mocker.patch(f"{_MODULE}.subprocess.run", return_value=result) + mocker.patch(f"{_PROTOCOL_MODULE}.subprocess.run", return_value=result) with pytest.raises(RuntimeError, match="produced no stdout payload"): _resolve_system_loaded_abs_path_in_subprocess("cudart") @@ -253,7 +254,7 @@ def test_subprocess_probe_raises_on_empty_stdout(mocker): def test_subprocess_probe_raises_on_invalid_json_payload(mocker): result = subprocess.CompletedProcess(args=[], returncode=0, stdout="not-json\n", stderr="") - mocker.patch(f"{_MODULE}.subprocess.run", return_value=result) + mocker.patch(f"{_PROTOCOL_MODULE}.subprocess.run", return_value=result) with pytest.raises(RuntimeError, match="invalid JSON payload"): _resolve_system_loaded_abs_path_in_subprocess("cudart") @@ -266,7 +267,7 @@ def test_subprocess_probe_raises_on_unexpected_json_payload(mocker): stdout='{"path": "/usr/local/cuda/lib64/libcudart.so.13"}\n', stderr="", ) - mocker.patch(f"{_MODULE}.subprocess.run", return_value=result) + mocker.patch(f"{_PROTOCOL_MODULE}.subprocess.run", return_value=result) with pytest.raises(RuntimeError, match="unexpected payload"): _resolve_system_loaded_abs_path_in_subprocess("cudart") diff --git a/cuda_pathfinder/tests/test_find_nvidia_dynamic_lib.py b/cuda_pathfinder/tests/test_find_nvidia_dynamic_lib.py new file mode 100644 index 0000000000..ac5fbae3d4 --- /dev/null +++ b/cuda_pathfinder/tests/test_find_nvidia_dynamic_lib.py @@ -0,0 +1,154 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +import os +import platform +import subprocess +import sys +import textwrap + +import pytest +from child_load_nvidia_dynamic_lib_helper import run_load_nvidia_dynamic_lib_in_subprocess + +from cuda.pathfinder import ( + DynamicLibNotAvailableError, + DynamicLibNotFoundError, + DynamicLibUnknownError, + find_nvidia_dynamic_lib, +) +from cuda.pathfinder._dynamic_libs import find_nvidia_dynamic_lib as find_nvidia_dynamic_lib_module +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._dynamic_libs.subprocess_protocol import ( + DYNAMIC_LIB_SUBPROCESS_CWD, + STATUS_NOT_FOUND, + STATUS_OK, + DynamicLibSubprocessPayload, + parse_dynamic_lib_subprocess_payload, +) +from cuda.pathfinder._utils.platform_aware import IS_WINDOWS, quote_for_shell + +STRICTNESS = os.environ.get("CUDA_PATHFINDER_TEST_LOAD_NVIDIA_DYNAMIC_LIB_STRICTNESS", "see_what_works") +assert STRICTNESS in ("see_what_works", "all_must_work") + + +def test_unknown_libname_raises_dynamic_lib_unknown_error(): + with pytest.raises(DynamicLibUnknownError, match=r"Unknown library name: 'not_a_real_lib'.*cudart"): + find_nvidia_dynamic_lib("not_a_real_lib") + + +def test_known_but_platform_unavailable_libname_raises_dynamic_lib_not_available_error(monkeypatch): + find_nvidia_dynamic_lib.cache_clear() + monkeypatch.setattr(load_nvidia_dynamic_lib_module, "_ALL_KNOWN_LIBNAMES", frozenset(("known_but_unavailable",))) + monkeypatch.setattr(load_nvidia_dynamic_lib_module, "_ALL_SUPPORTED_LIBNAMES", frozenset()) + monkeypatch.setattr(load_nvidia_dynamic_lib_module, "_PLATFORM_NAME", "TestOS") + with pytest.raises( + DynamicLibNotAvailableError, + match=r"known_but_unavailable.*not available on TestOS", + ): + find_nvidia_dynamic_lib("known_but_unavailable") + + +def _is_expected_find_failure(libname: str) -> bool: + # Mirror load-side strictness: libnames known to fail loading on this + # platform are also allowed to fail finding. + return libname == "nvpl_fftw" and platform.machine().lower() != "aarch64" + + +@pytest.mark.parametrize( + "libname", + supported_nvidia_libs.SUPPORTED_WINDOWS_DLLS if IS_WINDOWS else supported_nvidia_libs.SUPPORTED_LINUX_SONAMES, +) +def test_find_nvidia_dynamic_lib_returns_existing_path_without_loading(info_summary_append, libname): + find_nvidia_dynamic_lib.cache_clear() + try: + abs_path = find_nvidia_dynamic_lib(libname) + except DynamicLibNotFoundError: + if STRICTNESS == "all_must_work" and not _is_expected_find_failure(libname): + raise + info_summary_append(f"Not found: {libname=!r}") + return + + info_summary_append(f"abs_path={quote_for_shell(abs_path)}") + assert os.path.isabs(abs_path) + assert os.path.isfile(abs_path) + + +def test_find_matches_load_in_subprocess(info_summary_append): + libname = "cudart" + find_nvidia_dynamic_lib.cache_clear() + timeout = 120 if IS_WINDOWS else 30 + load_result = run_load_nvidia_dynamic_lib_in_subprocess(libname, timeout=timeout) + if load_result.returncode != 0: + pytest.skip(f"load subprocess failed for {libname!r}; consistency comparison N/A") + + load_payload = parse_dynamic_lib_subprocess_payload( + load_result.stdout, + libname=libname, + error_label="Load subprocess child process", + ) + if load_payload.status != STATUS_OK: + pytest.skip(f"{libname} not loadable on this host; nothing to compare against") + + find_abs_path = find_nvidia_dynamic_lib(libname) + assert load_payload.abs_path is not None + info_summary_append( + f"{libname}: load={quote_for_shell(load_payload.abs_path)} find={quote_for_shell(find_abs_path)}" + ) + assert os.path.samefile(find_abs_path, load_payload.abs_path) + + +def test_find_nvidia_dynamic_lib_propagates_subprocess_not_found_message(monkeypatch): + find_nvidia_dynamic_lib.cache_clear() + expected = "child loader said: cudart could not be located" + + def fake_run(mode, libname, *, timeout, error_label): + return DynamicLibSubprocessPayload( + status=STATUS_NOT_FOUND, + abs_path=None, + error={"type": "DynamicLibNotFoundError", "message": expected}, + ) + + monkeypatch.setattr(find_nvidia_dynamic_lib_module, "run_dynamic_lib_subprocess", fake_run) + with pytest.raises(DynamicLibNotFoundError, match=expected): + find_nvidia_dynamic_lib("cudart") + + +_DOES_NOT_LOAD_PROBE = textwrap.dedent( + """ + import sys + from cuda.pathfinder import DynamicLibNotFoundError, find_nvidia_dynamic_lib + libname = sys.argv[1] + try: + find_nvidia_dynamic_lib(libname) + except DynamicLibNotFoundError: + print("not-found"); sys.exit(0) + with open("/proc/self/maps") as f: + print("loaded" if ("lib" + libname) in f.read() else "ok") + """ +).strip() + + +def test_find_nvidia_dynamic_lib_does_not_load_in_caller_process(): + if IS_WINDOWS or not os.path.exists("/proc/self/maps"): + pytest.skip("Requires /proc/self/maps for in-process load detection") + + # Run in a fresh interpreter so other pathfinder tests in the same + # pytest process can't have pre-loaded the library. + libname = "cudart" + # Match the cwd used by the production subprocess helper so the child + # resolves the installed ``cuda.pathfinder`` (with ``_version.py``) rather + # than a source tree shadow on ``sys.path[0]``. + result = subprocess.run( # noqa: S603 - trusted argv: current interpreter + inline probe + [sys.executable, "-c", _DOES_NOT_LOAD_PROBE, libname], + capture_output=True, + text=True, + timeout=60, + check=False, + cwd=DYNAMIC_LIB_SUBPROCESS_CWD, + ) + assert result.returncode == 0, f"probe failed:\nstdout={result.stdout!r}\nstderr={result.stderr!r}" + verdict = result.stdout.strip().splitlines()[-1] + if verdict == "not-found": + pytest.skip(f"{libname} not available on this host") + assert verdict == "ok", f"find_nvidia_dynamic_lib must not load the library into the caller process ({verdict=})"