coverage: expand resource, cache, and DLPack regression coverage

cuda_core/tests/graph/test_graph_definition_lifetime.py

@@ -86,8 +86,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,
@@ -98,6 +99,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):

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")

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,

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

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)

cuda_core/tests/test_green_context.py

@@ -440,6 +440,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)
@@ -635,6 +676,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)

cuda_core/tests/test_memory.py

@@ -1887,6 +1887,34 @@ 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="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."""