test(cuda.core): cover API validation and resource lifecycles

Exercise validation failures, resource ownership, and recovery paths using real CUDA resources. Verify tensor-map dtype encoding and exporter lifetime without requiring runtime kernel compilation.

Signed-off-by: Rui Luo <ruluo@nvidia.com>

Author: rluo8

cuda_core/tests/graph/test_graph_definition_errors.py

@@ -6,13 +6,16 @@
 import ctypes
 
 import pytest
+from helpers.contexts import current_context_handle, use_context
 from helpers.graph_kernels import compile_common_kernels
 from helpers.misc import try_create_condition
 
-from cuda.core import Device, LaunchConfig
-from cuda.core._utils.cuda_utils import CUDAError
+from cuda.core import ContextOptions, Device, LaunchConfig, SMResourceOptions
+from cuda.core._utils.cuda_utils import CUDAError, driver, handle_return
+from cuda.core._utils.version import driver_version
 from cuda.core.graph import (
     EmptyNode,
+    GraphCompleteOptions,
     GraphCondition,
     GraphDefinition,
 )
@@ -150,6 +153,118 @@ def test_multiple_instantiation(init_cuda):
     assert g1 is not g2
 
 
+@pytest.mark.agent_authored(model="gpt-6-sol")
+def test_instantiate_rejects_cycle_and_accepts_repaired_graph(init_cuda):
+    """Removing a cycle makes an invalid graph executable again."""
+    definition = GraphDefinition()
+    called = []
+    first = definition.empty()
+    last = first.callback(lambda: called.append(True))
+    last.succ.add(first)
+    stream = init_cuda.create_stream()
+    executable = None
+    try:
+        with pytest.raises(RuntimeError, match="invalid structure, such as cycles"):
+            definition.instantiate()
+        assert called == []
+        last.succ.discard(first)
+        executable = definition.instantiate()
+        executable.launch(stream)
+        stream.sync()
+        assert called == [True]
+    finally:
+        if executable is not None:
+            executable.close()
+        last.succ.discard(first)
+        last.destroy()
+        first.destroy()
+        stream.close()
+
+
+@pytest.mark.agent_authored(model="gpt-6-sol")
+def test_device_launch_rejects_host_callback(init_cuda):
+    """Device-launch graphs reject host callbacks before execution."""
+    definition = GraphDefinition()
+    called = []
+    node = definition.callback(lambda: called.append(True))
+    try:
+        with pytest.raises(
+            RuntimeError, match="device launch failed because the graph contained an unsupported operation"
+        ):
+            definition.instantiate(GraphCompleteOptions(device_launch=True))
+        assert called == []
+    finally:
+        node.destroy()
+
+
+@pytest.mark.agent_authored(model="gpt-6-sol")
+def test_device_launch_rejects_nodes_from_different_contexts(init_cuda):
+    """Device-launch instantiation rejects mixed contexts and accepts a repaired graph."""
+    if driver_version() < (12, 4, 0):
+        pytest.skip("green contexts require CUDA driver 12.4 or newer")
+    if not init_cuda.properties.unified_addressing:
+        pytest.skip("device launch requires unified addressing")
+    original_context = current_context_handle()
+    definition = GraphDefinition()
+    green_context = first = foreign = executable = stream = None
+    try:
+        groups, _ = init_cuda.resources.sm.split(SMResourceOptions(count=None))
+        green_context = init_cuda.create_context(ContextOptions(resources=[groups[0]]))
+        module = compile_common_kernels()
+        kernel = module.get_kernel("empty_kernel")
+        config = LaunchConfig(grid=1, block=1)
+        first = definition.launch(config, kernel)
+        with use_context(init_cuda, green_context):
+            foreign = definition.launch(config, kernel)
+
+        # CUkernel handles are context-independent; check each node's resolved context.
+        first_params = handle_return(driver.cuGraphKernelNodeGetParams(first.handle))
+        foreign_params = handle_return(driver.cuGraphKernelNodeGetParams(foreign.handle))
+        assert int(first_params.ctx) == original_context
+        assert int(foreign_params.ctx) == int(green_context.handle)
+        assert int(foreign_params.ctx) != original_context
+        assert current_context_handle() == original_context
+        with pytest.raises(RuntimeError, match="nodes belonging to different contexts"):
+            definition.instantiate(GraphCompleteOptions(device_launch=True))
+        assert current_context_handle() == original_context
+
+        # A single-context graph remains valid after the failed instantiation.
+        foreign.destroy()
+        foreign = None
+        executable = definition.instantiate(GraphCompleteOptions(device_launch=True))
+        stream = init_cuda.create_stream()
+        executable.launch(stream)
+        stream.sync()
+        assert current_context_handle() == original_context
+    finally:
+        if executable is not None:
+            executable.close()
+        if foreign is not None:
+            foreign.destroy()
+        if first is not None:
+            first.destroy()
+        if stream is not None:
+            stream.close()
+        if green_context is not None:
+            green_context.close()
+
+
+@pytest.mark.agent_authored(model="gpt-6-sol")
+def test_instantiate_rejects_unassociated_condition(init_cuda):
+    """Instantiation rejects a condition that no conditional node uses."""
+    if driver_version() < (12, 8, 0):
+        pytest.skip("Unused graph-condition diagnostics require CUDA driver 12.8 or newer")
+    definition = GraphDefinition()
+    # Deliberately leave this graph-owned condition unassociated with a node.
+    definition.create_condition(default_value=1)
+    node = definition.empty()
+    try:
+        with pytest.raises(RuntimeError, match="conditional handles are not associated with conditional builders"):
+            definition.instantiate()
+    finally:
+        node.destroy()
+
+
 def test_unmatched_alloc_succeeds(init_cuda):
     """Alloc without corresponding free is valid (graph-scoped lifetime)."""
     _skip_if_no_mempool()

cuda_core/tests/test_device.py

@@ -2,12 +2,14 @@
 # SPDX-License-Identifier: Apache-2.0
 
 import contextlib
+import ctypes
 import pickle
 import subprocess
 import sys
 from concurrent.futures import ThreadPoolExecutor
 
 import pytest
+from helpers.buffers import DummyDeviceMemoryResource
 from helpers.contexts import (
     assert_device_operations_use_bound_context,
     current_context_handle,
@@ -18,7 +20,7 @@
 import cuda.core
 from cuda.bindings import driver, runtime
 from cuda.core import Device, StreamOptions
-from cuda.core._utils.cuda_utils import ComputeCapability, handle_return
+from cuda.core._utils.cuda_utils import ComputeCapability, CUDAError, handle_return
 from cuda.core._utils.version import driver_version
 
 
@@ -58,6 +60,95 @@ def test_device_set_current(deinit_cuda):
     assert handle_return(driver.cuCtxGetCurrent()) is not None
 
 
+@pytest.mark.parametrize("ordinal_kind", ["negative", "one-past-end"])
+@pytest.mark.agent_authored(model="gpt-6-sol")
+def test_device_rejects_out_of_range_ordinal(init_cuda, ordinal_kind):
+    """Invalid ordinals preserve the current context and cached Device."""
+    device_count = len(Device.get_all_devices())
+    ordinal = -1 if ordinal_kind == "negative" else device_count
+    original_context = current_context_handle()
+    with pytest.raises(ValueError):
+        Device(ordinal)
+    assert current_context_handle() == original_context
+    assert Device(init_cuda.device_id) is init_cuda
+
+
+@pytest.mark.agent_authored(model="gpt-6-sol")
+def test_device_context_requires_initialization_on_new_thread(init_cuda):
+    """Initialization on one thread does not initialize another thread's Device."""
+    device_id = init_cuda.device_id
+    original_context = current_context_handle()
+
+    def worker():
+        device = Device(device_id)
+        assert current_context_handle() == 0
+        with pytest.raises(CUDAError):
+            _ = device.context
+        with pytest.raises(CUDAError):
+            device.create_stream()
+        device.set_current()
+        stream = None
+        try:
+            stream = device.create_stream()
+            assert int(stream.context.handle) == int(device.context.handle)
+            with device.allocate(4, stream=stream) as buffer:
+                stream.sync()
+                handle_return(driver.cuMemsetD8(int(buffer.handle), 0x29, buffer.size))
+                result = (ctypes.c_ubyte * 4)()
+                handle_return(driver.cuMemcpyDtoH(ctypes.addressof(result), int(buffer.handle), 4))
+                assert bytes(result) == b"\x29" * 4
+            stream.sync()
+        finally:
+            if stream is not None:
+                stream.close()
+            handle_return(driver.cuCtxPopCurrent())
+
+    with ThreadPoolExecutor(max_workers=1) as executor:
+        executor.submit(worker).result()
+    assert current_context_handle() == original_context
+
+
+@pytest.mark.agent_authored(model="gpt-6-sol")
+def test_device_memory_resource_controls_allocation(init_cuda):
+    """Device.allocate uses the public memory resource assigned to the device."""
+    device = init_cuda
+    original_resource = device.memory_resource
+    replacement = DummyDeviceMemoryResource(device)
+    try:
+        device.memory_resource = replacement
+        assert device.memory_resource is replacement
+        with device.allocate(16, stream=device.default_stream) as buffer:
+            assert buffer.memory_resource is replacement
+            handle_return(driver.cuMemsetD8(int(buffer.handle), 0x5C, buffer.size))
+            result = (ctypes.c_ubyte * 16)()
+            handle_return(driver.cuMemcpyDtoH(ctypes.addressof(result), int(buffer.handle), 16))
+            assert bytes(result) == b"\x5c" * 16
+    finally:
+        device.memory_resource = original_resource
+    assert device.memory_resource is original_resource
+
+
+@pytest.mark.agent_authored(model="gpt-6-sol")
+def test_invalid_memory_resource_assignment_preserves_resource(init_cuda):
+    """A rejected replacement leaves the existing allocator operational."""
+    device = init_cuda
+    original_resource = device.memory_resource
+    try:
+        with pytest.raises(TypeError):
+            device.memory_resource = object()
+        assert device.memory_resource is original_resource
+        with device.allocate(4, stream=device.default_stream) as buffer:
+            device.default_stream.sync()
+            assert buffer.memory_resource is original_resource
+            handle_return(driver.cuMemsetD8(int(buffer.handle), 0x3A, buffer.size))
+            result = (ctypes.c_ubyte * 4)()
+            handle_return(driver.cuMemcpyDtoH(ctypes.addressof(result), int(buffer.handle), 4))
+            assert bytes(result) == b"\x3a" * 4
+        device.default_stream.sync()
+    finally:
+        device.memory_resource = original_resource
+
+
 @pytest.mark.agent_authored(model="gpt-5.6-sol")
 def test_primary_context_cleanup_is_safe_during_python_shutdown(init_cuda, tmp_path):
     result = subprocess.run(

cuda_core/tests/test_green_context.py

@@ -4,7 +4,7 @@
 
 import numpy as np
 import pytest
-from helpers.contexts import assert_device_operations_use_bound_context, use_context
+from helpers.contexts import assert_device_operations_use_bound_context, current_context_handle, use_context
 
 from cuda.core import (
     ContextOptions,
@@ -208,6 +208,24 @@ def test_create_context_requires_resources(init_cuda):
         init_cuda.create_context(object())
 
 
+@pytest.mark.agent_authored(model="gpt-6-sol")
+def test_create_context_rejects_empty_resources(init_cuda):
+    """An empty resource list cannot create a green context."""
+    original_context = current_context_handle()
+    with pytest.raises(ValueError):
+        init_cuda.create_context(ContextOptions(resources=[]))
+    assert current_context_handle() == original_context
+
+
+@pytest.mark.agent_authored(model="gpt-6-sol")
+def test_create_context_rejects_unsupported_resource(init_cuda):
+    """A resource of an unsupported type leaves the current context unchanged."""
+    original_context = current_context_handle()
+    with pytest.raises(TypeError):
+        init_cuda.create_context(ContextOptions(resources=[object()]))
+    assert current_context_handle() == original_context
+
+
 @pytest.mark.agent_authored(model="claude-opus-4.8")
 def test_context_handle_alias_and_closed_queries(init_cuda, sm_resource):
     """``Context._handle`` mirrors ``.handle``; after a (non-current) green

cuda_core/tests/test_launcher.py

@@ -1024,3 +1024,64 @@ class SubclassedHandle(driver.CUgraphConditionalHandle):
     dev = Device()
     dev.set_current()
     _assert_kernel_sees_ull(dev, handle, _UINT64_HANDLE_VALUE)
+
+
+@pytest.mark.parametrize("allocation", ["managed", "virtual"])
+@pytest.mark.agent_authored(model="gpt-6-sol")
+def test_launch_buffer_subclasses_preserve_kernel_arguments(init_cuda, allocation):
+    """Managed and virtual buffers marshal their device pointers into kernels."""
+    from contextlib import ExitStack
+
+    from cuda.bindings import driver
+    from cuda.core import ManagedBuffer
+    from cuda.core._utils.cuda_utils import handle_return
+
+    device = init_cuda
+    if allocation == "managed" and not device.properties.managed_memory:
+        pytest.skip("Device does not support managed memory")
+    if allocation == "virtual" and not device.properties.virtual_memory_management_supported:
+        pytest.skip("Device does not support CUDA virtual memory")
+    source = np.arange(16, dtype=np.int32)
+    actual = np.empty_like(source)
+    stream = device.create_stream()
+    buffer = program = managed_pointer = None
+    try:
+        if allocation == "managed":
+            managed_pointer = handle_return(
+                driver.cuMemAllocManaged(source.nbytes, driver.CUmemAttach_flags.CU_MEM_ATTACH_GLOBAL)
+            )
+            buffer = ManagedBuffer.from_handle(managed_pointer, source.nbytes)
+            assert isinstance(buffer, ManagedBuffer)
+        else:
+            from cuda.core import VirtualMemoryBuffer, VirtualMemoryResource, VirtualMemoryResourceOptions
+
+            resource = VirtualMemoryResource(device, VirtualMemoryResourceOptions(handle_type=None))
+            buffer = resource.allocate(source.nbytes, stream=stream)
+            assert isinstance(buffer, VirtualMemoryBuffer)
+        handle_return(driver.cuMemcpyHtoDAsync(buffer.handle, source.ctypes.data, source.nbytes, stream.handle))
+        program = Program(
+            'extern "C" __global__ void update(int* data, int offset) { '
+            "int i = threadIdx.x; data[i] = data[i] * 3 + offset; }",
+            "c++",
+        )
+        module = program.compile("cubin")
+        kernel = module.get_kernel("update")
+        launch(stream, LaunchConfig(grid=1, block=16), kernel, buffer, np.int32(7))
+        stream.sync()
+        handle_return(driver.cuMemcpyDtoH(actual.ctypes.data, buffer.handle, actual.nbytes))
+        np.testing.assert_array_equal(actual, source * 3 + 7)
+        buffer.close()
+        assert buffer.is_closed
+        with pytest.raises(RuntimeError, match="Buffer has been closed"):
+            launch(stream, LaunchConfig(grid=1, block=16), kernel, buffer, np.int32(7))
+    finally:
+        # Attempt every cleanup even if synchronization or another cleanup fails.
+        with ExitStack() as cleanup:
+            cleanup.callback(stream.close)
+            if program is not None:
+                cleanup.callback(program.close)
+            if managed_pointer is not None:
+                cleanup.callback(lambda: handle_return(driver.cuMemFree(managed_pointer)))
+            if buffer is not None:
+                cleanup.callback(buffer.close)
+            stream.sync()

cuda_core/tests/test_memory.py

@@ -1940,6 +1940,38 @@ def test_vmm_buffers_alive_at_shutdown_are_freed_quietly(init_cuda):
     assert result.stderr == ""
 
 
+@pytest.mark.parametrize(
+    "options",
+    [
+        pytest.param({"addr_align": 3}, id="invalid-alignment"),
+        pytest.param({"peers": (-1,)}, id="invalid-peer"),
+    ],
+)
+@pytest.mark.thread_unsafe(reason="records process-global warnings")
+@pytest.mark.agent_authored(model="gpt-6-sol")
+def test_failed_vmm_allocation_does_not_block_another_resource(init_cuda, options):
+    """A failed allocation leaves its config unchanged and another VMM resource usable."""
+    device = _vmm_device_or_skip()
+    invalid_resource = _vmm_resource(device, **options)
+    error_match = "CUDA_ERROR_INVALID_VALUE" if "addr_align" in options else r"CUDA_ERROR_INVALID_(VALUE|DEVICE)"
+    with assert_no_cuda_warning(), pytest.raises(CUDAError, match=error_match):
+        invalid_resource.allocate(4096)
+    for name, value in options.items():
+        assert getattr(invalid_resource.config, name) == value
+
+    # Use a separate resource with valid options; the invalid configuration is unchanged.
+    resource = _vmm_resource(device)
+    buffer = resource.allocate(4096)
+    ptr, size = int(buffer.handle), buffer.size
+    try:
+        _vmm_fill(buffer, 0x5C)
+        assert _vmm_read(buffer, 0, 16) == bytes([0x5C]) * 16
+    finally:
+        with assert_no_cuda_warning():
+            buffer.close()
+    _vmm_assert_released([(ptr, size)])
+
+
 @pytest.mark.agent_authored(model="claude-fable-5-1")
 @pytest.mark.thread_unsafe(reason="records process-global warnings")
 def test_vmm_failed_grow_leaves_input_intact(init_cuda):