From a94c136cc68284f288efa5067e55afeeeb02208a Mon Sep 17 00:00:00 2001 From: Rui Luo Date: Fri, 9 Oct 2026 16:21:00 +0800 Subject: [PATCH 1/3] coverage: expand resource, cache, and DLPack regression coverage Signed-off-by: Rui Luo --- .../graph/test_graph_definition_lifetime.py | 75 ++- .../graph/test_graph_definition_mutation.py | 3 + .../tests/graph/test_graph_node_update.py | 53 ++ cuda_core/tests/memory/test_managed_ops.py | 50 +- cuda_core/tests/system/test_system_events.py | 12 + cuda_core/tests/test_green_context.py | 54 ++ cuda_core/tests/test_memory.py | 8 + cuda_core/tests/test_module.py | 14 + cuda_core/tests/test_program.py | 43 ++ cuda_core/tests/test_program_cache.py | 482 ++++++++++++++++++ cuda_core/tests/test_tensor_map.py | 1 + cuda_core/tests/test_texture_surface.py | 37 ++ cuda_core/tests/test_utils_dlpack.py | 143 +++++- 13 files changed, 962 insertions(+), 13 deletions(-) diff --git a/cuda_core/tests/graph/test_graph_definition_lifetime.py b/cuda_core/tests/graph/test_graph_definition_lifetime.py index a33b5497ae8..99afe7d5386 100644 --- a/cuda_core/tests/graph/test_graph_definition_lifetime.py +++ b/cuda_core/tests/graph/test_graph_definition_lifetime.py @@ -85,8 +85,9 @@ def _wait_until(predicate, timeout=None, interval=0.02): VirtualMemoryResourceOptions, ) from cuda.core._utils.cuda_utils import CUDAError -from cuda.core._utils.version import driver_version +from cuda.core._utils.version import BUILD_CUDA_MAJOR, driver_version from cuda.core.graph import ( + AllocNode, ChildGraphNode, ConditionalNode, EventRecordNode, @@ -97,6 +98,7 @@ def _wait_until(predicate, timeout=None, interval=0.02): KernelNode, MemcpyNode, ) +from cuda.core.typing import GraphMemoryType def _skip_if_no_mempool(): @@ -124,6 +126,12 @@ def _virtual_memory_resource(dev): ] +def _skip_if_no_managed_mempool(): + _skip_if_no_mempool() + if not Device(0).properties.concurrent_managed_access: + pytest.skip("Device does not support managed memory pool operations") + + # ============================================================================= # Conditional body graph lifetime # ============================================================================= @@ -1274,12 +1282,8 @@ def test_kernel_node_reconstruction_preserves_validity(init_cuda): stream.sync() -def _pred_chain_memcpy(g, bufs): - memory_resource = LegacyPinnedMemoryResource() - src = memory_resource.allocate(8) - dst = memory_resource.allocate(8) - bufs.extend((src, dst)) - node = g.memcpy(dst, src, 8) +def _make_pred_chain_memcpy(builder, src, dst): + node = builder.memcpy(dst, src, 8) src_ptr, dst_ptr, size = node.src, node.dst, node.size succ = node.record(Device().create_event()) @@ -1292,6 +1296,21 @@ def check(reconstructed): return node, succ, check +def _pred_chain_memcpy_host(g, bufs): + memory_resource = LegacyPinnedMemoryResource() + src = memory_resource.allocate(8) + dst = memory_resource.allocate(8) + bufs.extend((src, dst)) + return _make_pred_chain_memcpy(g, src, dst) + + +def _pred_chain_memcpy_device(g, bufs): + _skip_if_no_mempool() + src = g.allocate(8) + dst = g.allocate(8) + return _make_pred_chain_memcpy(g.join(src, dst), src.dptr, dst.dptr) + + def _pred_chain_event_record(g, bufs): event = Device().create_event() node = g.record(event) @@ -1316,6 +1335,35 @@ def check(reconstructed): return node, succ, check +def _pred_chain_alloc_peer_access(g, bufs): + _skip_if_no_mempool() + device = Device() + node = g.allocate(64, device=device, peer_access=[device]) + expected_peer_access = (device.device_id,) + succ = node.record(device.create_event()) + + def check(reconstructed): + assert isinstance(reconstructed, AllocNode) + assert reconstructed.peer_access == expected_peer_access + + return node, succ, check + + +def _pred_chain_alloc_managed(g, bufs): + _skip_if_no_managed_mempool() + node = g.allocate(64, memory_type=GraphMemoryType.MANAGED) + expected_dptr = node.dptr + succ = node.record(Device().create_event()) + + def check(reconstructed): + assert isinstance(reconstructed, AllocNode) + assert reconstructed.memory_type == "managed" + assert reconstructed.dptr == expected_dptr + assert reconstructed.dptr != 0 + + return node, succ, check + + def _pred_chain_free(g, bufs): _skip_if_no_mempool() alloc = g.allocate(64) @@ -1333,12 +1381,21 @@ def check(reconstructed): @pytest.mark.parametrize( "factory", [ - _pred_chain_memcpy, + _pred_chain_memcpy_host, + _pred_chain_memcpy_device, _pred_chain_event_record, _pred_chain_event_wait, + _pred_chain_alloc_peer_access, + pytest.param( + _pred_chain_alloc_managed, + marks=pytest.mark.skipif( + BUILD_CUDA_MAJOR < 13, + reason="managed alloc requires a CUDA 13.0+ build", + ), + ), _pred_chain_free, ], - ids=["memcpy", "event_record", "event_wait", "free"], + ids=["memcpy_host", "memcpy_device", "event_record", "event_wait", "alloc_peer_access", "alloc_managed", "free"], ) @pytest.mark.agent_authored(model="gpt-5.6-sol") def test_graph_nodes_reconstructed_via_pred_chain(init_cuda, factory): diff --git a/cuda_core/tests/graph/test_graph_definition_mutation.py b/cuda_core/tests/graph/test_graph_definition_mutation.py index 3b4e5e1185e..57b7a85b93c 100644 --- a/cuda_core/tests/graph/test_graph_definition_mutation.py +++ b/cuda_core/tests/graph/test_graph_definition_mutation.py @@ -401,6 +401,9 @@ def test_discard_wrong_type_is_noop(init_cuda): owner.succ.discard(42) assert owner.succ == {neighbor} + with pytest.raises(KeyError): + owner.succ.remove("not a node") + assert owner.succ == {neighbor} @pytest.mark.agent_authored(model="gpt-5.6") diff --git a/cuda_core/tests/graph/test_graph_node_update.py b/cuda_core/tests/graph/test_graph_node_update.py index c3170599477..9f31df6a118 100644 --- a/cuda_core/tests/graph/test_graph_node_update.py +++ b/cuda_core/tests/graph/test_graph_node_update.py @@ -1065,6 +1065,59 @@ def test_executable_node_enable_state(init_cuda, node_kind): assert view.is_enabled +@pytest.mark.parametrize("operation", ["update", "is-enabled", "enable"]) +@pytest.mark.agent_authored(model="gpt-6-sol") +def test_executable_kernel_node_rejects_destroyed_source_node(init_cuda, operation): + """Executable-node operations reject a destroyed source node.""" + if driver_version() < (12, 2, 0): + pytest.skip("individual graph node updates require CUDA 12.2+") + + kernel = compile_common_kernels().get_kernel("empty_kernel") + config = LaunchConfig(grid=1, block=1) + graph_def = GraphDefinition() + node = graph_def.launch(config, kernel) + graph = graph_def.instantiate() + view = graph[node] + + node.destroy() + + operations = { + "update": lambda: view.update(config=config, kernel=kernel, args=()), + "is-enabled": lambda: view.is_enabled, + "enable": view.enable, + } + with pytest.raises(RuntimeError, match="GraphNode has been destroyed"): + operations[operation]() + + +@pytest.mark.agent_authored(model="gpt-6-sol") +def test_kernel_update_rejects_existing_cooperative_node(init_cuda): + """Definition updates reject a node already marked cooperative by the driver.""" + if driver_version() < (12, 2, 0): + pytest.skip("individual graph node updates require CUDA 12.2+") + + kernel = compile_common_kernels().get_kernel("empty_kernel") + config = LaunchConfig(grid=1, block=1) + graph_def = GraphDefinition() + node = graph_def.launch(config, kernel) + + attr = driver.CUkernelNodeAttrValue() + attr.cooperative = 1 + handle_return( + driver.cuGraphKernelNodeSetAttribute( + node.handle, + driver.CUkernelNodeAttrID.CU_LAUNCH_ATTRIBUTE_COOPERATIVE, + attr, + ) + ) + + with pytest.raises( + NotImplementedError, + match="updating clustered or cooperative kernel nodes is not supported", + ): + node.update(config=config) + + @pytest.mark.agent_authored(model="gpt-5.6") def test_executable_node_view_rejects_unsupported_and_destroyed_nodes( init_cuda, diff --git a/cuda_core/tests/memory/test_managed_ops.py b/cuda_core/tests/memory/test_managed_ops.py index 6915d261e1d..0500ad57df8 100644 --- a/cuda_core/tests/memory/test_managed_ops.py +++ b/cuda_core/tests/memory/test_managed_ops.py @@ -61,12 +61,12 @@ def _skip_if_managed_discard_prefetch_unsupported(device): # Fixture set: # * location_ops_device / location_ops_mr — concurrent_managed_access tier; -# covers advise/prefetch (every test in this file needs it). +# gates advise/prefetch tests that request these fixtures. # * discard_prefetch_device — adds cuMemDiscardAndPrefetchBatchAsync + # multi-GPU concurrent-managed-access check. # * managed_buffer — parametrized over pool-allocated (ManagedMemoryResource) -# vs external (DummyUnifiedMemoryResource + from_handle); used by -# TestManagedBuffer so each method runs against both buffer sources. +# vs external (DummyUnifiedMemoryResource + from_handle); tests requesting +# this fixture run against both buffer sources. @pytest.fixture @@ -472,6 +472,50 @@ def test_accessed_by_set_assignment_validates_kind_before_mutation( assert device in buf.accessed_by assert Host() in buf.accessed_by + @pytest.mark.thread_unsafe(reason="external_managed_buffer is shared between threads") + @pytest.mark.agent_authored(model="gpt-6-sol") + def test_accessed_by_rejects_non_location_types(self, location_ops_device, external_managed_buffer): + """``accessed_by`` handles invalid location types without mutating valid entries.""" + device = location_ops_device + buf = external_managed_buffer + accessed_by = buf.accessed_by + invalid = "not-a-location" + + buf.accessed_by = {device} + assert set(accessed_by) == {device} + + assert invalid not in accessed_by + with pytest.raises(TypeError, match="expected Device or Host"): + accessed_by.add(invalid) + assert set(accessed_by) == {device} + + accessed_by.discard(invalid) + assert set(accessed_by) == {device} + + with pytest.raises(TypeError, match="accessed_by entries must be Device or Host"): + # A new valid entry precedes the invalid one, exposing partial mutation. + buf.accessed_by = [Host(), invalid] + assert set(accessed_by) == {device} + + @pytest.mark.agent_authored(model="gpt-6-sol") + def test_managed_attribute_queries_reject_closed_buffer(self, init_cuda): + """Managed attribute queries reject closed buffers, including cached views.""" + # A non-owning handle exercises closed guards without a managed allocation. + with ManagedBuffer.from_handle(0, 0) as buf: + assert not buf.is_closed + accessed_by = buf.accessed_by + assert buf.is_closed + + for operation in ( + lambda: buf.read_mostly, + lambda: buf.preferred_location, + lambda: buf.last_prefetch_location, + lambda: buf.accessed_by, + lambda: len(accessed_by), + ): + with pytest.raises(RuntimeError, match="Buffer has been closed"): + operation() + def test_instance_prefetch(self, location_ops_device, managed_buffer): device = location_ops_device buf = managed_buffer diff --git a/cuda_core/tests/system/test_system_events.py b/cuda_core/tests/system/test_system_events.py index 92443365923..19e7e1f9f57 100644 --- a/cuda_core/tests/system/test_system_events.py +++ b/cuda_core/tests/system/test_system_events.py @@ -94,3 +94,15 @@ def test_register_events(): with pytest.raises(system.TimeoutError): events.wait(timeout_ms=500, buffer_size=1) + + +@pytest.mark.agent_authored(model="gpt-6-sol") +def test_register_events_rejects_invalid_event_type(): + with pytest.raises(ValueError, match="Invalid event type: not-a-system-event"): + system.register_events(["not-a-system-event"]) + + +@pytest.mark.agent_authored(model="gpt-6-sol") +def test_register_events_rejects_non_event_container(): + with pytest.raises(TypeError, match="events must be"): + system.register_events(0) diff --git a/cuda_core/tests/test_green_context.py b/cuda_core/tests/test_green_context.py index 773e0f4f42a..f0f56a9d353 100644 --- a/cuda_core/tests/test_green_context.py +++ b/cuda_core/tests/test_green_context.py @@ -437,6 +437,47 @@ def test_discovery_respects_explicit_coscheduled_sm_count(self, sm_resource): assert groups[0].sm_count % alignment == 0 + @pytest.mark.agent_authored(model="gpt-6-sol") + def test_discovery_accepts_preferred_coscheduled_sm_count(self, init_cuda): + """Discovery with a preferred co-scheduling hint matches native-driver metadata.""" + if BUILD_CUDA_MAJOR < 13 or driver_version() < (13, 1, 0): + pytest.skip("preferred co-scheduled SM split requires CUDA 13.1+") + if init_cuda.compute_capability < (10, 0): + pytest.skip("distinct preferred co-scheduling requires Blackwell+") + + sm_resource = init_cuda.resources.sm + alignment = sm_resource.coscheduled_alignment + preferred = 2 * alignment + if preferred > min(32, sm_resource.sm_count): + pytest.skip("device cannot support a distinct preferred count") + + # Probe the same request independently. Once the native API accepts it, + # any cuda.core failure must propagate instead of becoming a skip. + native_input = driver.CUdevResource(_ptr=sm_resource.handle) + params = driver.CU_DEV_SM_RESOURCE_GROUP_PARAMS() + params.smCount = 0 + params.coscheduledSmCount = alignment + params.preferredCoscheduledSmCount = preferred + status, expected_groups, expected_remaining = driver.cuDevSmResourceSplit(1, native_input, 0, [params]) + if status == driver.CUresult.CUDA_ERROR_INVALID_RESOURCE_CONFIGURATION: + pytest.skip("native driver cannot satisfy the preferred co-scheduling request") + handle_return((status,)) + + groups, remaining = sm_resource.split( + SMResourceOptions( + count=None, + coscheduled_sm_count=alignment, + preferred_coscheduled_sm_count=preferred, + ) + ) + assert len(groups) == len(expected_groups) == 1 + assert groups[0].sm_count == expected_groups[0].sm.smCount + if expected_groups[0].sm.smCoscheduledAlignment: + assert groups[0].coscheduled_alignment == expected_groups[0].sm.smCoscheduledAlignment + assert groups[0].sm_count % groups[0].coscheduled_alignment == 0 + if expected_remaining.type == driver.CUdevResourceType.CU_DEV_RESOURCE_TYPE_SM: + assert remaining.sm_count == expected_remaining.sm.smCount + def test_two_groups(self, sm_resource): """Two-group split succeeds for a supported explicit request.""" split = _find_supported_split(sm_resource, n_groups=2) @@ -632,6 +673,19 @@ def test_primary_context_stream_sm_resources(self, init_cuda, sm_resource): finally: stream.close() + @pytest.mark.agent_authored(model="gpt-6-sol") + def test_primary_context_stream_workqueue_resources(self, init_cuda, wq_resource): + """A primary-context stream reports the full device workqueue resources.""" + stream = init_cuda.create_stream() + try: + stream_wq = stream.resources.workqueue + assert stream_wq.handle != 0 + assert stream_wq.sharing_scope is wq_resource.sharing_scope + assert stream_wq.concurrency_limit == wq_resource.concurrency_limit + assert stream_wq.device.device_id == init_cuda.device_id + finally: + stream.close() + # --------------------------------------------------------------------------- # Kernel launch in green context (explicit model) diff --git a/cuda_core/tests/test_memory.py b/cuda_core/tests/test_memory.py index f33e17ddccf..a4d40a84932 100644 --- a/cuda_core/tests/test_memory.py +++ b/cuda_core/tests/test_memory.py @@ -1899,6 +1899,14 @@ def test_vmm_deallocate_frees_wrapped_pointer(init_cuda): _vmm_assert_released([(int(ptr), size)]) +@pytest.mark.agent_authored(model="gpt-6-sol") +def test_vmm_modify_allocation_rejects_non_buffer(init_cuda): + """``modify_allocation`` rejects inputs that are not Buffer objects.""" + mr = _vmm_resource(_vmm_device_or_skip()) + with pytest.raises(TypeError, match=r"Argument 'buf' has incorrect type .*Buffer, got object"): + mr.modify_allocation(object(), 4096) + + @pytest.mark.agent_authored(model="claude-fable-5-1") def test_vmm_buffers_alive_at_shutdown_are_freed_quietly(init_cuda): """Buffers still referenced at interpreter exit are freed without a warning or an error.""" diff --git a/cuda_core/tests/test_module.py b/cuda_core/tests/test_module.py index 3b1fe9f7a9d..a3ffa6d4e75 100644 --- a/cuda_core/tests/test_module.py +++ b/cuda_core/tests/test_module.py @@ -902,3 +902,17 @@ def get_kernel_only(): stream.sync() assert result[0] == 42, f"Expected 42, got {result[0]}" + + +@pytest.mark.agent_authored(model="gpt-6-sol") +def test_kernel_equality_hash_and_repr(get_saxpy_kernel_cubin): + """Kernel identity follows the underlying CUkernel handle.""" + kernel, objcode = get_saxpy_kernel_cubin + same = Kernel.from_handle(int(kernel.handle), objcode) + other = objcode.get_kernel("saxpy") + + assert kernel == same + assert hash(kernel) == hash(same) + assert kernel != other + assert kernel != object() + assert repr(kernel).startswith("= 13\.3"): + ProgramOptions(arch="sm_80", use_bundled_headers=True) + assert calls == [True] diff --git a/cuda_core/tests/test_program_cache.py b/cuda_core/tests/test_program_cache.py index bd327eb0d05..9e5a4277d75 100644 --- a/cuda_core/tests/test_program_cache.py +++ b/cuda_core/tests/test_program_cache.py @@ -2564,3 +2564,485 @@ def test_program_cache_preexisting_shared_root_used_as_is(tmp_path): assert stat.S_IMODE(os.stat(root).st_mode) == 0o777 assert stat.S_IMODE(os.stat(root / "tmp").st_mode) == 0o700 + + +@pytest.mark.thread_unsafe(reason="monkeypatches the process-global NVVM fingerprint probe") +@pytest.mark.agent_authored(model="gpt-6-sol") +def test_make_program_cache_key_nvvm_extra_sources_order_and_content(monkeypatch): + """NVVM cache keys distinguish source order, names, and content while normalizing byte-like bodies.""" + from cuda.core.utils import _program_cache + + monkeypatch.setattr(_program_cache._keys, "_nvvm_fingerprint", lambda: "lib=13.3;ir=2.0.0.0") + src_a = ("mod_a", "int a = 1;") + src_b = ("mod_b", "int b = 2;") + # Caller order is part of the NVVM cache-key fingerprint. + k_ab = _make_key( + code="abc", + code_type="nvvm", + target_type="ptx", + options=_opts(extra_sources=[src_a, src_b]), + ) + k_ba = _make_key( + code="abc", + code_type="nvvm", + target_type="ptx", + options=_opts(extra_sources=[src_b, src_a]), + ) + assert k_ab != k_ba + + k_renamed = _make_key( + code="abc", + code_type="nvvm", + target_type="ptx", + options=_opts(extra_sources=[("mod_a2", "int a = 1;"), src_b]), + ) + assert k_renamed != k_ab + + k_changed = _make_key( + code="abc", + code_type="nvvm", + target_type="ptx", + options=_opts(extra_sources=[("mod_a", "int a = 99;"), src_b]), + ) + assert k_changed != k_ab + + k_str = _make_key( + code="abc", + code_type="nvvm", + target_type="ptx", + options=_opts(extra_sources=[("mod", "hello")]), + ) + k_bytes = _make_key( + code="abc", + code_type="nvvm", + target_type="ptx", + options=_opts(extra_sources=[("mod", b"hello")]), + ) + k_bytearray = _make_key( + code="abc", + code_type="nvvm", + target_type="ptx", + options=_opts(extra_sources=[("mod", bytearray(b"hello"))]), + ) + assert k_str == k_bytes == k_bytearray + + +@pytest.mark.agent_authored(model="gpt-6-sol") +def test_make_program_cache_key_nvvm_empty_extra_sources_equals_none(): + """NVVM cache keys treat empty ``extra_sources`` like no extra sources.""" + k_none = _make_key(code="abc", code_type="nvvm", target_type="ptx", options=_opts()) + k_empty = _make_key( + code="abc", + code_type="nvvm", + target_type="ptx", + options=_opts(extra_sources=[]), + ) + k_one = _make_key( + code="abc", + code_type="nvvm", + target_type="ptx", + options=_opts(extra_sources=[("mod", "int x;")]), + ) + assert k_none == k_empty + assert k_one != k_none + + +@pytest.mark.agent_authored(model="gpt-6-sol") +def test_make_program_cache_key_nvvm_rejects_non_str_bytes_code(): + """NVVM cache keys reject source code that is not string or bytes-like.""" + with pytest.raises(TypeError, match="str or bytes"): + _make_key(code=123, code_type="nvvm", target_type="ptx") + + +@pytest.mark.agent_authored(model="gpt-6-sol") +def test_make_program_cache_key_nvvm_accepts_bytearray_code(): + """``bytearray`` is encoded the same as ``bytes`` (matching ``Program()``).""" + k_bytes = _make_key(code=b"abc", code_type="nvvm", target_type="ptx") + k_bytearray = _make_key(code=bytearray(b"abc"), code_type="nvvm", target_type="ptx") + assert k_bytes == k_bytearray + + +@pytest.mark.thread_unsafe(reason="monkeypatches the process-global NVVM fingerprint probe") +@pytest.mark.agent_authored(model="gpt-6-sol") +def test_make_program_cache_key_nvvm_probe_failure_is_stable(monkeypatch): + """NVVM cache keys remain stable and distinct when compiler fingerprinting fails.""" + from cuda.core.utils import _program_cache + + monkeypatch.setattr(_program_cache._keys, "_nvvm_fingerprint", lambda: "lib=13.3,ir=2.0") + k_ok = _make_key(code="abc", code_type="nvvm", target_type="ptx") + calls = 0 + + def _broken(): + nonlocal calls + calls += 1 + raise RuntimeError(f"nvvm probe failed {calls}") + + monkeypatch.setattr(_program_cache._keys, "_nvvm_fingerprint", _broken) + k_broken1 = _make_key(code="abc", code_type="nvvm", target_type="ptx") + k_broken2 = _make_key(code="abc", code_type="nvvm", target_type="ptx") + assert calls == 2 + assert k_ok != k_broken1 + assert k_broken1 == k_broken2 + + +@pytest.mark.thread_unsafe(reason="monkeypatches process-global linker probes") +@pytest.mark.agent_authored(model="gpt-6-sol") +def test_make_program_cache_key_linker_decision_failure_is_stable(monkeypatch): + """Failed linker decisions yield stable PTX keys distinct from working backends.""" + from cuda.core import _linker + from cuda.core.utils import _program_cache + + monkeypatch.setattr(_program_cache._keys, "_linker_backend_and_version", lambda _use_driver: ("nvJitLink", "13030")) + monkeypatch.setattr(_linker, "_decide_nvjitlink_or_driver", lambda: False) + k_ok = _make_key(code=".version 7.0", code_type="ptx") + calls = 0 + + def _broken(): + nonlocal calls + calls += 1 + raise RuntimeError("linker decision failed") + + monkeypatch.setattr(_linker, "_decide_nvjitlink_or_driver", _broken) + k1 = _make_key(code=".version 7.0", code_type="ptx") + k2 = _make_key(code=".version 7.0", code_type="ptx") + assert calls == 2 + assert k_ok != k1 + assert k1 == k2 + # The probe cannot know which linker would run, so it cannot apply the + # cuLink-specific unsupported-option gate. + k_ptxas = _make_key( + code=".version 7.0", + code_type="ptx", + options=_opts(ptxas_options="-v"), + ) + assert calls == 3 + assert k_ptxas != k1 + + +@pytest.mark.thread_unsafe(reason="monkeypatches process-global linker, module, and import state") +@pytest.mark.agent_authored(model="gpt-6-sol") +def test_make_program_cache_key_cold_nvjitlink_import_tracks_version(monkeypatch): + """PTX keys are stable after cold nvJitLink imports and change with its version.""" + import builtins + import sys + from types import SimpleNamespace + + from cuda import bindings + + observed = set() + real_import = builtins.__import__ + + def _use_nvjitlink(): + observed.add("backend") + return False + + def _version(): + observed.add("version") + return version + + fake_module = SimpleNamespace(version=_version) + + def _import(name, globals=None, locals=None, fromlist=(), level=0): + if name == "cuda.bindings" and fromlist and "nvjitlink" in fromlist: + observed.add("import") + return SimpleNamespace(nvjitlink=fake_module) + return real_import(name, globals, locals, fromlist, level) + + keys = [] + for version in ((13, 3), (13, 3), (13, 4)): + observed.clear() + with monkeypatch.context() as patch: + # Each key must discover the module afresh, including the repeated version. + patch.delitem(sys.modules, "cuda.bindings.nvjitlink", raising=False) + patch.delattr(bindings, "nvjitlink", raising=False) + patch.setattr(_linker, "_decide_nvjitlink_or_driver", _use_nvjitlink) + patch.setattr(builtins, "__import__", _import) + keys.append(_make_key(code=".version 7.0", code_type="ptx")) + assert {"backend", "import", "version"} <= observed + + assert keys[0] == keys[1] + assert keys[0] != keys[2] + + +@pytest.mark.thread_unsafe(reason="monkeypatches pathlib.Path.stat process-wide") +@pytest.mark.agent_authored(model="gpt-6-sol") +def test_filestream_clear_tolerates_entry_vanishing_before_prune(tmp_path, monkeypatch): + """``clear`` tolerates an entry vanishing between its snapshot and prune.""" + from pathlib import Path + + from cuda.core.utils import FileStreamProgramCache + + with FileStreamProgramCache(tmp_path / "fc") as cache: + cache[b"k"] = _fake_object_code(b"v") + path = cache._path_for_key(b"k") + real_stat = Path.stat + calls = 0 + + def _stat(self, *args, **kwargs): + nonlocal calls + if self == path: + calls += 1 + if calls == 2: + path.unlink() + raise FileNotFoundError(self) + return real_stat(self, *args, **kwargs) + + monkeypatch.setattr(Path, "stat", _stat) + cache.clear() + assert calls >= 2, "race injection never reached clear()'s prune re-stat" + assert not path.exists() + + +@pytest.mark.thread_unsafe(reason="monkeypatches pathlib.Path.unlink process-wide") +@pytest.mark.agent_authored(model="gpt-6-sol") +def test_filestream_clear_tolerates_unlink_already_gone(tmp_path, monkeypatch): + """``clear`` treats a racing unlink as a successful prune.""" + from pathlib import Path + + from cuda.core.utils import FileStreamProgramCache + + with FileStreamProgramCache(tmp_path / "fc") as cache: + cache[b"k"] = _fake_object_code(b"v") + path = cache._path_for_key(b"k") + real_unlink = Path.unlink + unlink_injected = False + + def _already_gone(self, *args, **kwargs): + nonlocal unlink_injected + if self != path: + return real_unlink(self, *args, **kwargs) + unlink_injected = True + real_unlink(self, *args, **kwargs) + raise FileNotFoundError(self) + + monkeypatch.setattr(Path, "unlink", _already_gone) + cache.clear() + assert unlink_injected, "race injection never reached clear()'s unlink" + assert not path.exists() + + +@pytest.mark.thread_unsafe(reason="monkeypatches pathlib.Path.unlink process-wide") +@pytest.mark.agent_authored(model="gpt-6-sol") +def test_filestream_clear_reraises_non_sharing_permission_error(tmp_path, monkeypatch): + """File-cache clear propagates non-sharing permission errors.""" + import errno + from pathlib import Path + + from cuda.core.utils import FileStreamProgramCache, _program_cache + + monkeypatch.setattr(_program_cache._file_stream, "_IS_WINDOWS", True) + + with FileStreamProgramCache(tmp_path / "fc") as cache: + cache[b"k"] = _fake_object_code(b"v") + + def _denied(self, *args, **kwargs): + exc = PermissionError("user mapped file") + exc.winerror = 1224 # ERROR_USER_MAPPED_FILE, not a sharing violation + exc.errno = errno.EACCES + raise exc + + monkeypatch.setattr(Path, "unlink", _denied) + with pytest.raises(PermissionError, match="user mapped file"): + cache.clear() + + +@pytest.mark.thread_unsafe(reason="monkeypatches pathlib.Path.stat process-wide") +@pytest.mark.agent_authored(model="gpt-6-sol") +def test_filestream_read_exhausted_sharing_retry_is_cache_miss(tmp_path, monkeypatch): + """Exhausted Windows sharing retries turn file-cache reads into misses.""" + from pathlib import Path + + from cuda.core.utils import FileStreamProgramCache, _program_cache + + monkeypatch.setattr(_program_cache._file_stream, "_IS_WINDOWS", True) + monkeypatch.setattr(_program_cache._file_stream, "_REPLACE_RETRY_DELAYS", (0.0, 0.0)) + + with FileStreamProgramCache(tmp_path / "fc") as cache: + cache[b"k"] = _fake_object_code(b"v") + path = cache._path_for_key(b"k") + assert cache[b"k"] == b"v" + real_stat = Path.stat + attempts = 0 + + def _always_sharing(self, *args, **kwargs): + nonlocal attempts + if self != path: + return real_stat(self, *args, **kwargs) + attempts += 1 + exc = PermissionError("sharing violation") + exc.winerror = 32 + raise exc + + monkeypatch.setattr(Path, "stat", _always_sharing) + assert cache.get(b"k") is None + assert attempts == 2 + + +@pytest.mark.thread_unsafe(reason="monkeypatches os.scandir process-wide") +@pytest.mark.agent_authored(model="gpt-6-sol") +def test_filestream_iter_skips_shard_dir_that_vanishes(tmp_path, monkeypatch): + """File-cache iteration skips shard directories that vanish during traversal.""" + import os as _os + from pathlib import Path + + from cuda.core.utils import FileStreamProgramCache + + with FileStreamProgramCache(tmp_path / "fc") as cache: + cache[b"a"] = _fake_object_code(b"A") + cache[b"b"] = _fake_object_code(b"B") + assert cache._path_for_key(b"a").parent != cache._path_for_key(b"b").parent + real_scandir = _os.scandir + entries = cache._entries.resolve() + shard_scans = 0 + + def _scandir(path): + nonlocal shard_scans + path_p = Path(_os.fspath(path)).resolve() + if path_p.parent == entries: + shard_scans += 1 + if shard_scans == 1: + raise FileNotFoundError(path) + return real_scandir(path) + + monkeypatch.setattr(_os, "scandir", _scandir) + assert len(cache) == 1 + assert shard_scans == 2 + + +@pytest.mark.thread_unsafe(reason="monkeypatches pathlib.Path.stat process-wide") +@pytest.mark.agent_authored(model="gpt-6-sol") +def test_filestream_open_with_size_cap_skips_vanished_entries(tmp_path, monkeypatch): + """Opening a capped file cache skips entries that vanish during size accounting.""" + from pathlib import Path + + from cuda.core.utils import FileStreamProgramCache + + root = tmp_path / "fc" + values = {b"a": b"A" * 32, b"b": b"B" * 48} + with FileStreamProgramCache(root) as cache: + for key, value in values.items(): + cache[key] = value + paths = {cache._path_for_key(key): key for key in values} + + real_stat = Path.stat + vanished = None + + def _stat(self, *args, **kwargs): + nonlocal vanished + if self in paths and vanished is None: + vanished = self + self.unlink() + raise FileNotFoundError(self) + return real_stat(self, *args, **kwargs) + + monkeypatch.setattr(Path, "stat", _stat) + with FileStreamProgramCache(root, max_size_bytes=10_000) as cache: + assert vanished is not None, "race injection never reached the startup size scan" + assert not vanished.exists() + survivor_key = next(key for path, key in paths.items() if path != vanished) + assert cache._tracked_size_bytes == len(values[survivor_key]) + assert cache[survivor_key] == values[survivor_key] + cache[b"new"] = b"new" + assert cache[b"new"] == b"new" + assert cache._tracked_size_bytes == len(values[survivor_key]) + len(b"new") + + +@pytest.mark.thread_unsafe(reason="monkeypatches os.unlink process-wide") +@pytest.mark.agent_authored(model="gpt-6-sol") +def test_filestream_sweep_skips_permission_error_on_stale_tmp(tmp_path, monkeypatch): + """File-cache startup ignores permission errors while pruning stale temporary files.""" + import os as _os + from pathlib import Path + + from cuda.core.utils import FileStreamProgramCache, _program_cache + + root = tmp_path / "fc" + with FileStreamProgramCache(root): + pass + ancient = root / "tmp" / "entry-ancient" + ancient.write_bytes(b"crashed") + ancient_mtime = time.time() - _program_cache._file_stream._TMP_STALE_AGE_SECONDS - 60 + _os.utime(ancient, (ancient_mtime, ancient_mtime)) + + real_unlink = _os.unlink + denied = [] + + def _denied(path): + if Path(path) == ancient: + denied.append(Path(path)) + raise PermissionError("locked tmp") + return real_unlink(path) + + monkeypatch.setattr(_os, "unlink", _denied) + with FileStreamProgramCache(root): + pass + assert denied == [ancient] + assert ancient.exists() + + +@pytest.mark.thread_unsafe(reason="monkeypatches pathlib.Path.stat process-wide") +@pytest.mark.agent_authored(model="gpt-6-sol") +def test_filestream_clear_skips_entry_that_vanishes_during_snapshot(tmp_path, monkeypatch): + """File-cache clear skips entries that vanish during its snapshot scan.""" + from pathlib import Path + + from cuda.core.utils import FileStreamProgramCache + + with FileStreamProgramCache(tmp_path / "fc") as cache: + cache[b"a"] = _fake_object_code(b"A") + cache[b"b"] = _fake_object_code(b"B") + gone = cache._path_for_key(b"a") + remaining = cache._path_for_key(b"b") + real_stat = Path.stat + vanished = False + + def _stat(self, *args, **kwargs): + nonlocal vanished + if self == gone and not vanished: + vanished = True + self.unlink() + raise FileNotFoundError(self) + return real_stat(self, *args, **kwargs) + + monkeypatch.setattr(Path, "stat", _stat) + cache.clear() + assert vanished, "race injection never reached clear()'s snapshot stat" + assert not gone.exists() + assert not remaining.exists() + assert cache.get(b"b") is None + + +@pytest.mark.thread_unsafe(reason="monkeypatches pathlib.Path.stat process-wide") +@pytest.mark.agent_authored(model="gpt-6-sol") +def test_filestream_write_over_cap_skips_vanished_files(tmp_path, monkeypatch): + """File-cache eviction skips entries that vanish before unlink.""" + from pathlib import Path + + from cuda.core.utils import FileStreamProgramCache + + with FileStreamProgramCache(tmp_path / "fc", max_size_bytes=250) as cache: + cache[b"a"] = b"a" * 100 + cache[b"b"] = b"b" * 100 + gone = cache._path_for_key(b"a") + st = gone.stat() + os.utime(gone, ns=(st.st_atime_ns - 10_000_000_000, st.st_mtime_ns)) + real_stat = Path.stat + n_stat = {"n": 0} + + def _stat(self, *args, **kwargs): + # Let the scan see the file, then remove it on the eviction re-stat. + if self == gone: + n_stat["n"] += 1 + if n_stat["n"] == 2: + self.unlink() + raise FileNotFoundError(self) + return real_stat(self, *args, **kwargs) + + monkeypatch.setattr(Path, "stat", _stat) + cache[b"c"] = b"c" * 100 # must not raise + assert n_stat["n"] > 1, "race injection never reached eviction's re-stat" + monkeypatch.setattr(Path, "stat", real_stat) + assert not gone.exists() + assert cache.get(b"b") == b"b" * 100 + assert cache.get(b"c") == b"c" * 100 + assert cache._tracked_size_bytes == 200 diff --git a/cuda_core/tests/test_tensor_map.py b/cuda_core/tests/test_tensor_map.py index 06433792a61..bab91c32ed1 100644 --- a/cuda_core/tests/test_tensor_map.py +++ b/cuda_core/tests/test_tensor_map.py @@ -320,6 +320,7 @@ def test_as_tensor_map_host_view_rejected_without_tma(self): box_dim=(32, 32), data_type=TensorMapDataType.FLOAT32, element_strides=(1, 1), + interleave=TensorMapInterleave.NONE, swizzle=TensorMapSwizzle.SWIZZLE_128B, l2_promotion=TensorMapL2Promotion.L2_128B, oob_fill=TensorMapOOBFill.NAN_REQUEST_ZERO_FMA, diff --git a/cuda_core/tests/test_texture_surface.py b/cuda_core/tests/test_texture_surface.py index 42ae716e8f3..23e1e2976fb 100644 --- a/cuda_core/tests/test_texture_surface.py +++ b/cuda_core/tests/test_texture_surface.py @@ -1207,3 +1207,40 @@ def test_surface_object_keeps_backing_array_alive(init_cuda): assert len(arr_refs) == 1, "SurfaceObject should still reference its backing OpaqueArray via the ResourceDescriptor" assert surf.handle != 0 surf.close() + + +@pytest.mark.agent_authored(model="gpt-6-sol") +def test_array_options_rejects_unknown_format_string(): + """OpaqueArrayOptions rejects unknown format strings.""" + with pytest.raises(ValueError, match="format must be an ArrayFormatType or one of"): + OpaqueArrayOptions(shape=(8,), format="not_a_format", num_channels=1) + + +@pytest.mark.agent_authored(model="gpt-6-sol") +def test_array_repr_includes_shape_format_and_channels(init_cuda): + arr = Device().create_opaque_array(OpaqueArrayOptions(shape=(8, 4), format=ArrayFormatType.FLOAT32, num_channels=1)) + try: + text = repr(arr) + assert "OpaqueArray(" in text + assert "shape=(8, 4)" in text + assert "FLOAT32" in text + assert "num_channels=1" in text + finally: + arr.close() + + +@pytest.mark.agent_authored(model="gpt-6-sol") +def test_array_copy_rejects_non_buffer(init_cuda): + """copy_from/copy_to reject objects that do not provide the buffer protocol.""" + arr = Device().create_opaque_array(OpaqueArrayOptions(shape=(8,), format=ArrayFormatType.UINT8, num_channels=1)) + try: + stream = Device().create_stream() + try: + with pytest.raises(TypeError, match=r"buffer-protocol object|bytes-like object"): + arr.copy_from(object(), stream=stream) + with pytest.raises(TypeError, match=r"buffer-protocol object|bytes-like object"): + arr.copy_to(object(), stream=stream) + finally: + stream.close() + finally: + arr.close() diff --git a/cuda_core/tests/test_utils_dlpack.py b/cuda_core/tests/test_utils_dlpack.py index 724c2f9d95d..fa92248cf04 100644 --- a/cuda_core/tests/test_utils_dlpack.py +++ b/cuda_core/tests/test_utils_dlpack.py @@ -8,6 +8,9 @@ """ import ctypes +import subprocess +import sys +from pathlib import Path try: import ml_dtypes @@ -27,7 +30,6 @@ _Py_DecRef.argtypes = (ctypes.c_void_p,) _Py_DecRef.restype = None - _NUMPY_NATIVE_DLPACK_DTYPES = ( np.uint8, np.uint16, @@ -268,6 +270,10 @@ class _DLManagedTensorVersioned(ctypes.Structure): ] +# --------------------------------------------------------------------------- +# DLPack capsule validation +# --------------------------------------------------------------------------- + # DLPACK_FLAG_BITMASK_READ_ONLY in dlpack.h. _FLAG_READ_ONLY = 1 << 0 @@ -284,6 +290,52 @@ def __dlpack__(self, **kwargs): return self.capsule +def _retag_capsule_device(capsule, device_type): + """Retag an unconsumed DLPack capsule as CUDA host or managed memory.""" + # Keep the fake producer self-consistent: DLPack requires the capsule's + # device type to agree with __dlpack_device__. + if _PyCapsule_IsValid(capsule, b"dltensor_versioned"): + name, managed_cls = b"dltensor_versioned", _DLManagedTensorVersioned + else: + name, managed_cls = b"dltensor", _DLManagedTensor + dlm = ctypes.cast(_PyCapsule_GetPointer(capsule, name), ctypes.POINTER(managed_cls)) + dlm.contents.dl_tensor.device.device_type = int(device_type) + return capsule + + +@pytest.mark.parametrize( + "device_type", + [ + pytest.param(DLDeviceType.kDLCUDAHost, id="cuda-host"), + pytest.param(DLDeviceType.kDLCUDAManaged, id="cuda-managed"), + ], +) +@pytest.mark.agent_authored(model="gpt-6-sol") +def test_from_dlpack_cuda_host_or_managed_stream_none_passthrough(device_type): + """CUDA-host and managed producers receive stream_ptr=None unchanged.""" + src = np.arange(4, dtype=np.float32) + received_streams = [] + + class _Export: + def __dlpack_device__(self): + return (int(device_type), 0) + + def __dlpack__(self, stream=None, max_version=None, **kwargs): + received_streams.append(stream) + # Host memory retagged as pinned/managed, so the capsule reports the + # same device as __dlpack_device__ and the view stays importable here. + base = StridedMemoryView.from_array_interface(src) + capsule = base.__dlpack__(stream=stream, max_version=max_version, **kwargs) + return _retag_capsule_device(capsule, device_type) + + view = StridedMemoryView.from_dlpack(_Export(), stream_ptr=None) + assert received_streams == [None] + assert view.is_device_accessible is True + assert view.device_id == 0 + assert view.__dlpack_device__() == (int(device_type), 0) + assert np.array_equal(np.from_dlpack(view), src) + + @pytest.mark.agent_authored(model="gpt-5.6-sol") def test_dlpack_versioned_readonly_export_and_import(init_cuda): """The versioned readonly flag survives a StridedMemoryView round-trip.""" @@ -334,6 +386,81 @@ def test_from_dlpack_malformed_dtype_rejected_on_access(code, bits, lanes, excep _ = imported.dtype +def _assert_bfloat16_without_ml_dtypes(): + from cuda.core import _memoryview + + assert ml_dtypes is None + assert _memoryview.bfloat16 is None + base = StridedMemoryView.from_any_interface(np.arange(4, dtype=np.int32), stream_ptr=-1) + capsule = base.__dlpack__(max_version=(1, 0)) + dlm = ctypes.cast( + _PyCapsule_GetPointer(capsule, b"dltensor_versioned"), + ctypes.POINTER(_DLManagedTensorVersioned), + ) + dlm.contents.dl_tensor.dtype = _DLDataType(4, 16, 1) + imported = StridedMemoryView.from_dlpack( + _VersionedCapsuleExport(base, capsule), + stream_ptr=-1, + ) + with pytest.raises(NotImplementedError, match=r"requires `ml_dtypes`"): + _ = imported.dtype + + +@pytest.mark.agent_authored(model="gpt-6-sol") +def test_from_dlpack_bfloat16_requires_ml_dtypes(): + """kDLBfloat without ml_dtypes raises on ``.dtype`` access.""" + check = """ +import numpy +import runpy +import sys + +assert "cuda.core._memoryview" not in sys.modules +sys.modules.pop("ml_dtypes", None) + +class _BlockMlDtypes: + def find_spec(self, fullname, path=None, target=None): + if fullname == "ml_dtypes" or fullname.startswith("ml_dtypes."): + raise ModuleNotFoundError("No module named 'ml_dtypes'") + +sys.meta_path.insert(0, _BlockMlDtypes()) +runpy.run_path(sys.argv[1])["_assert_bfloat16_without_ml_dtypes"]() +""" + subprocess.run( # noqa: S603 - trusted interpreter and in-tree test file + (sys.executable, "-I", "-c", check, str(Path(__file__).resolve())), + check=True, + ) + + +@pytest.mark.parametrize( + ("lanes", "expected_shape"), + [ + pytest.param(1, None, id="non-byte-aligned"), + pytest.param(2, (4,), id="byte-aligned"), + ], +) +@pytest.mark.agent_authored(model="gpt-6-sol") +def test_from_dlpack_shape_validates_total_dtype_width(lanes, expected_shape): + """Shape access validates the total DLPack element width, including lanes.""" + base = StridedMemoryView.from_any_interface(np.arange(4, dtype=np.int32), stream_ptr=-1) + capsule = base.__dlpack__(max_version=(1, 0)) + dlm = ctypes.cast( + _PyCapsule_GetPointer(capsule, b"dltensor_versioned"), + ctypes.POINTER(_DLManagedTensorVersioned), + ) + dlm.contents.dl_tensor.dtype = _DLDataType(0, 12, lanes) + imported = StridedMemoryView.from_dlpack( + _VersionedCapsuleExport(base, capsule), + stream_ptr=-1, + ) + if expected_shape is None: + with pytest.raises(ValueError, match="multiple of 8"): + _ = imported.shape + else: + assert imported.shape == expected_shape + # Reinterpretation checks the inferred 3-byte width through the public API. + assert imported.view(dtype=np.dtype("V3")).shape == expected_shape + + @pytest.mark.agent_authored(model="cursor-grok-4.5") @pytest.mark.parametrize( "max_version, capsule_name, managed_cls", @@ -501,6 +628,20 @@ def test_dlpack_c_exchange_api_dltensor_from_py_object_scalar(): assert not out.strides +@pytest.mark.parametrize("entry_point", ["managed_tensor_from_py_object_no_sync", "dltensor_from_py_object_no_sync"]) +@pytest.mark.agent_authored(model="gpt-6-sol") +def test_dlpack_c_exchange_api_export_failure_sets_exception(init_cuda, entry_point): + """Exporting a view without dtype information raises BufferError through either C entry point.""" + api = _get_exchange_api() + with init_cuda.memory_resource.allocate(16, stream=init_cuda.default_stream) as buffer: + view = StridedMemoryView.from_buffer(buffer, shape=(16,), itemsize=1, dtype=None) + out = ctypes.c_void_p(123) if entry_point == "managed_tensor_from_py_object_no_sync" else _DLTensor() + with pytest.raises(BufferError, match="without dtype information"): + getattr(api, entry_point)(id(view), ctypes.byref(out)) + if entry_point == "managed_tensor_from_py_object_no_sync": + assert not out.value + + @pytest.mark.agent_authored(model="gpt-5.6-sol") def test_dlpack_c_exchange_api_managed_tensor_export_failure_sets_exception(): """A view that DLPack cannot describe reports its BufferError through the From 1bab5d940f17df2cd28cd7c941d678b266b42fca Mon Sep 17 00:00:00 2001 From: Rui Luo Date: Tue, 13 Oct 2026 15:58:52 +0800 Subject: [PATCH 2/3] test: cover non-VMM buffers and reuse subprocess runner Signed-off-by: Rui Luo --- cuda_core/tests/test_memory.py | 20 ++++++++++++++++++++ cuda_core/tests/test_utils_dlpack.py | 15 +++++++-------- 2 files changed, 27 insertions(+), 8 deletions(-) diff --git a/cuda_core/tests/test_memory.py b/cuda_core/tests/test_memory.py index a4d40a84932..2d8b1a69427 100644 --- a/cuda_core/tests/test_memory.py +++ b/cuda_core/tests/test_memory.py @@ -1907,6 +1907,26 @@ def test_vmm_modify_allocation_rejects_non_buffer(init_cuda): mr.modify_allocation(object(), 4096) +@pytest.mark.agent_authored(model="gpt-6-sol") +def test_vmm_modify_allocation_rejects_non_vmm_buffer(init_cuda): + """modify_allocation rejects non-VMM Buffers without closing them.""" + mr = _vmm_resource(_vmm_device_or_skip()) + stream = init_cuda.create_stream() + try: + buf = init_cuda.memory_resource.allocate(64, stream=stream) + try: + assert isinstance(buf, Buffer) + assert not isinstance(buf, VirtualMemoryBuffer) + with pytest.raises(TypeError, match="buf must be a VirtualMemoryBuffer"): + mr.modify_allocation(buf, 4096) + assert not buf.is_closed + finally: + buf.close(stream=stream) + stream.sync() + finally: + stream.close() + + @pytest.mark.agent_authored(model="claude-fable-5-1") def test_vmm_buffers_alive_at_shutdown_are_freed_quietly(init_cuda): """Buffers still referenced at interpreter exit are freed without a warning or an error.""" diff --git a/cuda_core/tests/test_utils_dlpack.py b/cuda_core/tests/test_utils_dlpack.py index fa92248cf04..11fa8472717 100644 --- a/cuda_core/tests/test_utils_dlpack.py +++ b/cuda_core/tests/test_utils_dlpack.py @@ -8,8 +8,6 @@ """ import ctypes -import subprocess -import sys from pathlib import Path try: @@ -18,6 +16,7 @@ ml_dtypes = None import numpy as np import pytest +from cuda_python_test_helpers.subprocess_runner import run_python_snippet from cuda.core._dlpack import DLDeviceType from cuda.core.utils import StridedMemoryView @@ -409,7 +408,10 @@ def _assert_bfloat16_without_ml_dtypes(): @pytest.mark.agent_authored(model="gpt-6-sol") def test_from_dlpack_bfloat16_requires_ml_dtypes(): """kDLBfloat without ml_dtypes raises on ``.dtype`` access.""" - check = """ + # Only the parent needs this helper; the child loads this file from an empty directory. + from helpers.child_processes import child_timeout_sec + + check = f""" import numpy import runpy import sys @@ -423,12 +425,9 @@ def find_spec(self, fullname, path=None, target=None): raise ModuleNotFoundError("No module named 'ml_dtypes'") sys.meta_path.insert(0, _BlockMlDtypes()) -runpy.run_path(sys.argv[1])["_assert_bfloat16_without_ml_dtypes"]() +runpy.run_path({str(Path(__file__).resolve())!r})["_assert_bfloat16_without_ml_dtypes"]() """ - subprocess.run( # noqa: S603 - trusted interpreter and in-tree test file - (sys.executable, "-I", "-c", check, str(Path(__file__).resolve())), - check=True, - ) + run_python_snippet(check, timeout=child_timeout_sec()) @pytest.mark.parametrize( From 9010d2665039404ff574093211355a6dd560ff22 Mon Sep 17 00:00:00 2001 From: Rui Luo Date: Tue, 13 Oct 2026 16:53:23 +0800 Subject: [PATCH 3/3] test: keep the DLPack subprocess helper import in the parent Signed-off-by: Rui Luo --- cuda_core/tests/test_utils_dlpack.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/cuda_core/tests/test_utils_dlpack.py b/cuda_core/tests/test_utils_dlpack.py index 11fa8472717..4f24b8ad1b6 100644 --- a/cuda_core/tests/test_utils_dlpack.py +++ b/cuda_core/tests/test_utils_dlpack.py @@ -16,7 +16,6 @@ ml_dtypes = None import numpy as np import pytest -from cuda_python_test_helpers.subprocess_runner import run_python_snippet from cuda.core._dlpack import DLDeviceType from cuda.core.utils import StridedMemoryView @@ -408,7 +407,8 @@ def _assert_bfloat16_without_ml_dtypes(): @pytest.mark.agent_authored(model="gpt-6-sol") def test_from_dlpack_bfloat16_requires_ml_dtypes(): """kDLBfloat without ml_dtypes raises on ``.dtype`` access.""" - # Only the parent needs this helper; the child loads this file from an empty directory. + # Import parent-only helpers here; the child loads this file from an empty directory. + from cuda_python_test_helpers.subprocess_runner import run_python_snippet from helpers.child_processes import child_timeout_sec check = f"""