coverage: add cuda.core tests for system device, checkpoint, memory, stream, green context, and tensor map

cuda_core/tests/graph/test_graph_definition.py

@@ -701,6 +701,22 @@ def test_node_attrs_preserved_by_nodes(node_spec):
         assert getattr(retrieved, attr) == getattr(node, attr), f"{spec.name}.{attr} not preserved by nodes()"
 
 
+@pytest.mark.agent_authored(model="claude-opus-4.8")
+def test_host_callback_node_reconstructed_from_embedded_child(init_cuda):
+    """A host-callback node read through an embedded child graph is reconstructed via _create_from_driver."""
+    # The same-wrapper nodes() (test_node_attrs_preserved_by_nodes) returns the
+    # registry-cached object and never exercises reconstruction; only the embedded
+    # child graph carries fresh, unregistered node handles. Host callback is
+    # mempool-free so it reconstructs here; alloc-based nodes stay cached.
+    child = GraphDefinition()
+    _build_host_callback_node(child)
+    parent = GraphDefinition()
+    reconstructed = list(parent.embed(child).child_graph.nodes())
+    assert any(isinstance(n, HostCallbackNode) for n in reconstructed), (
+        f"no reconstructed HostCallbackNode in {[type(n).__name__ for n in reconstructed]}"
+    )
+
+
 def test_identity_preservation(init_cuda):
     """Round-trips through nodes(), edges(), and pred/succ return extant
     objects rather than duplicates."""

cuda_core/tests/system/test_system_device.py

@@ -722,6 +722,39 @@ def test_temperature():
             assert sensor.default_min_temp <= sensor.current_temp <= sensor.default_max_temp
 
 
+@pytest.mark.agent_authored(model="claude-opus-4.8")
+def test_temperature_arg_validation():
+    # Both getters reject an unknown key before issuing any NVML call.
+    temperature = system.Device(index=0).temperature
+    with pytest.raises(ValueError, match="Invalid temperature threshold type"):
+        temperature.get_threshold("not-a-threshold")
+    with pytest.raises(ValueError, match="Invalid thermal sensor index"):
+        temperature.get_thermal_settings("not-a-sensor")
+
+
+@pytest.mark.agent_authored(model="claude-opus-4.8")
+def test_device_constructor_selector_validation():
+    # The constructor requires exactly one selector, rejected before NVML is touched.
+    with pytest.raises(ValueError, match="only one of"):
+        system.Device(index=0, uuid="ignored")
+    with pytest.raises(ValueError, match="either a device"):
+        system.Device()
+
+
+@pytest.mark.agent_authored(model="claude-opus-4.8")
+def test_device_arg_validation():
+    device = system.Device(index=0)
+    # Each argument validator raises before reaching the driver/NVML call.
+    with pytest.raises(ValueError, match="Invalid affinity scope"):
+        device.get_memory_affinity("not-a-scope")
+    with pytest.raises(ValueError, match="Invalid affinity scope"):
+        device.get_cpu_affinity("not-a-scope")
+    with pytest.raises(ValueError, match="Invalid topology level"):
+        list(device.get_topology_nearest_gpus("not-a-level"))
+    with pytest.raises(ValueError, match="Invalid P2P caps index"):
+        system.get_p2p_status(device, device, "not-an-index")
+
+
 def test_pstates():
     for device in system.Device.get_all_devices():
         with unsupported_before(device, None):

cuda_core/tests/test_checkpoint.py

@@ -409,6 +409,145 @@ def test_pid_is_read_only(self):
             proc.pid = 2
 
 
+# -- Pure helpers (no GPU / driver needed) ---------------------------------
+
+import ctypes
+
+from cuda.bindings import driver as _bindings_driver
+
+# The checkpoint functions, structs, and enums are generated and shipped
+# together from the same CUDA headers, so probe them as one atomic API surface.
+_HAS_CHECKPOINT_BINDINGS = all(hasattr(_bindings_driver, name) for name in checkpoint._REQUIRED_BINDING_ATTRS)
+
+needs_checkpoint_bindings = pytest.mark.skipif(
+    not _HAS_CHECKPOINT_BINDINGS,
+    reason="cuda.bindings does not expose the CUDA checkpoint API",
+)
+
+
+@pytest.mark.agent_authored(model="claude-opus-4.8")
+class TestCheckpointHelpers:
+    """Host-only tests for the arg-validation and struct-marshalling helpers.
+
+    The driver-backed lifecycle/migration scenarios skip without a checkpoint-capable
+    Linux driver, so these are the only coverage of these helpers on most CI.
+    """
+
+    @pytest.mark.parametrize(
+        ("value", "error_type", "match"),
+        [
+            (True, TypeError, "timeout_ms must be an int"),
+            (1.5, TypeError, "timeout_ms must be an int"),
+            ("0", TypeError, "timeout_ms must be an int"),
+            (-1, ValueError, "timeout_ms must be >= 0"),
+        ],
+    )
+    def test_check_timeout_ms_rejects_invalid(self, value, error_type, match):
+        with pytest.raises(error_type, match=match):
+            checkpoint._check_timeout_ms(value)
+
+    def test_make_restore_args_rejects_non_mapping(self):
+        with pytest.raises(TypeError, match="gpu_mapping must be a mapping"):
+            checkpoint._make_restore_args(_bindings_driver, [("a", "b")])
+
+    def test_make_restore_args_empty_mapping_returns_none(self):
+        # An empty mapping produces no GPU pairs, so there is nothing to restore.
+        assert checkpoint._make_restore_args(_bindings_driver, {}) is None
+
+    @needs_checkpoint_bindings
+    def test_make_restore_args_builds_pairs(self):
+        old = "00000000-0000-0000-0000-000000000001"
+        new = "00000000-0000-0000-0000-000000000002"
+        args = checkpoint._make_restore_args(_bindings_driver, {old: new})
+        assert isinstance(args, _bindings_driver.CUcheckpointRestoreArgs)
+        assert args.gpuPairsCount == 1
+        # The pair must map old->new in that order (not swapped or duplicated).
+        pair = args.gpuPairs[0]
+        assert bytes(pair.oldUuid.bytes) == bytes.fromhex(old.replace("-", ""))
+        assert bytes(pair.newUuid.bytes) == bytes.fromhex(new.replace("-", ""))
+
+    @pytest.mark.parametrize(
+        ("value", "match"),
+        [
+            ("not-hex-zz", "32 hex characters"),
+            ("00", "32 hex characters"),  # valid hex but wrong length (1 byte)
+        ],
+    )
+    def test_as_cuuuid_rejects_bad_strings(self, value, match):
+        with pytest.raises(ValueError, match=match):
+            checkpoint._as_cuuuid(_bindings_driver, value, [])
+
+    def test_as_cuuuid_rejects_wrong_type(self):
+        with pytest.raises(TypeError, match="must be CUDA UUID objects or UUID strings"):
+            checkpoint._as_cuuuid(_bindings_driver, 12345, [])
+
+    @pytest.mark.parametrize(
+        "value",
+        [
+            "0123456789abcdef0123456789abcdef",  # bare 32 hex chars
+            "01234567-89ab-cdef-0123-456789abcdef",  # hyphenated form (Device.uuid style)
+        ],
+    )
+    def test_as_cuuuid_from_string_decodes_bytes_and_appends_backing_buffer(self, value):
+        buffers = []
+        result = checkpoint._as_cuuuid(_bindings_driver, value, buffers)
+        assert isinstance(result, _bindings_driver.CUuuid)
+        # Stripped hex must decode to the exact 16 CUuuid bytes (guards fromhex/replace).
+        assert bytes(result.bytes) == bytes.fromhex(value.replace("-", ""))
+        # _as_cuuuid appends the backing ctypes buffer to the caller's list so it survives
+        # until the caller copies the bytes into the pair struct.
+        assert len(buffers) == 1
+        assert isinstance(buffers[0], ctypes.Array)
+
+    def test_as_cuuuid_passes_through_cuuuid_instance(self):
+        existing = _bindings_driver.CUuuid()
+        # An already-constructed CUuuid is returned unchanged and adds no buffer.
+        buffers = []
+        assert checkpoint._as_cuuuid(_bindings_driver, existing, buffers) is existing
+        assert buffers == []
+
+
+@pytest.mark.agent_authored(model="claude-opus-4.8")
+class TestCheckpointDriverDispatch:
+    """Driver-dispatch (_call_driver) result-code / exception translation.
+
+    _call_driver runs against a boundary-mock ``func`` (dependency-injected as its
+    argument) so the translation branches exercise without a live driver — the real
+    ``checkpoint._driver`` still supplies the CUresult enum.
+    """
+
+    @pytest.mark.parametrize("err_name", ["CUDA_ERROR_NOT_FOUND", "CUDA_ERROR_NOT_SUPPORTED"])
+    def test_call_driver_translates_unsupported_result_codes(self, err_name):
+        """NOT_FOUND / NOT_SUPPORTED become the 'not supported by the installed NVIDIA driver' RuntimeError."""
+        driver = checkpoint._driver
+
+        def fake(*args):
+            return (getattr(driver.CUresult, err_name),)
+
+        with pytest.raises(RuntimeError, match="not supported by the installed NVIDIA driver"):
+            checkpoint._call_driver(driver, fake)
+
+    def test_call_driver_translates_missing_symbol_runtimeerror(self):
+        """A binding 'symbol not found' RuntimeError is rewritten into the upgrade-your-driver message."""
+        driver = checkpoint._driver
+
+        def fake(*args):
+            raise RuntimeError("Function cuCheckpointProcessLock not found")
+
+        with pytest.raises(RuntimeError, match="not supported by the installed NVIDIA driver"):
+            checkpoint._call_driver(driver, fake)
+
+    def test_call_driver_reraises_unrelated_runtimeerror(self):
+        """A RuntimeError unrelated to the missing-symbol case propagates as-is."""
+        driver = checkpoint._driver
+
+        def fake(*args):
+            raise RuntimeError("some other failure")
+
+        with pytest.raises(RuntimeError, match="some other failure"):
+            checkpoint._call_driver(driver, fake)
+
+
 # -- Lifecycle (single GPU, real driver) -----------------------------------
 
 

cuda_core/tests/test_green_context.py

@@ -181,6 +181,24 @@ def test_create_context_requires_resources(init_cuda):
         init_cuda.create_context(object())
 
 
+@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
+    context is closed its handle-backed queries degrade gracefully: ``handle`` is
+    ``None``, ``is_green`` is ``False``, and ``resources`` raises."""
+    groups, _ = sm_resource.split(SMResourceOptions(count=None))
+    ctx = init_cuda.create_context(ContextOptions(resources=[groups[0]]))
+    # `_handle` is a thin alias of the public `handle` property.
+    assert ctx._handle == ctx.handle
+    assert ctx.handle is not None
+
+    ctx.close()
+    assert ctx.handle is None
+    assert ctx.is_green is False
+    with pytest.raises(RuntimeError, match="Cannot query resources"):
+        _ = ctx.resources
+
+
 # ---------------------------------------------------------------------------
 # SM resource query
 # ---------------------------------------------------------------------------
@@ -306,6 +324,20 @@ def test_negative_count_raises(self, sm_resource):
         with pytest.raises(ValueError, match="count must be non-negative"):
             sm_resource.split(SMResourceOptions(count=-1))
 
+    @pytest.mark.agent_authored(model="claude-opus-4.8")
+    def test_empty_count_sequence_raises(self, sm_resource):
+        """An empty ``count`` sequence has no groups to split into."""
+        with pytest.raises(ValueError, match="count sequence must not be empty"):
+            sm_resource.split(SMResourceOptions(count=[]))
+
+    @pytest.mark.agent_authored(model="claude-opus-4.8")
+    @pytest.mark.parametrize("bad_count", [3.5, object()])
+    def test_count_wrong_type_raises(self, sm_resource, bad_count):
+        """``count`` that is neither int, Sequence, nor None is rejected before
+        any driver call."""
+        with pytest.raises(TypeError, match="count must be int, Sequence, or None"):
+            sm_resource.split(SMResourceOptions(count=bad_count))
+
     def test_dry_run_cannot_create_context(self, init_cuda, sm_resource):
         groups, _ = sm_resource.split(SMResourceOptions(count=None), dry_run=True)
         assert len(groups) == 1
@@ -496,6 +528,19 @@ def test_stream_resources_match_context(self, green_ctx, sm_resource):
         except (RuntimeError, ValueError, CUDAError):
             pass  # workqueue not available on this driver/build
 
+    @pytest.mark.agent_authored(model="claude-opus-4.8")
+    def test_primary_context_stream_sm_resources(self, init_cuda, sm_resource):
+        """A stream on the *primary* (non-green) context queries SM resources via
+        the plain ``cuCtxGetDevResource`` path (distinct from the green-context
+        path exercised elsewhere): the stream carries a context handle but it is
+        not a green context, so the whole device is reported."""
+        stream = init_cuda.create_stream()
+        try:
+            stream_sm = stream.resources.sm
+            assert stream_sm.sm_count == sm_resource.sm_count
+        finally:
+            stream.close()
+
 
 # ---------------------------------------------------------------------------
 # Kernel launch in green context (explicit model)

cuda_core/tests/test_launcher.py

@@ -536,25 +536,52 @@ class MyBool(ctypes.c_bool):
     assert holder.ptr != 0
 
 
+@pytest.mark.agent_authored(model="claude-opus-4.8")
 @requires_module(np, "2.2.5", reason="need numpy 2.2.5+ (numpy GH #28632)")
 @pytest.mark.parametrize(
-    ("scalar_kind", "np_dtype", "cpp_type", "raw_value"),
+    ("base_type", "np_dtype", "cpp_type", "raw_value"),
     [
-        ("ctypes", np.int32, "signed int", -123456),
-        ("numpy", np.float32, "float", 3.14),
+        # ctypes scalar subclasses — one per prepare_ctypes_arg isinstance-fallback
+        # branch. Values are chosen to expose a wrong width/sign: unsigned values
+        # exceed the same-width signed max, and c_uint64 exceeds uint32 max so a
+        # uint64 branch misrouted to prepare_arg[uint32_t] truncates 0x1_0000_0001
+        # to 1 and fails the readback.
+        (ctypes.c_bool, np.bool_, "bool", True),
+        (ctypes.c_int8, np.int8, "signed char", -42),
+        (ctypes.c_int16, np.int16, "signed short", -1234),
+        (ctypes.c_int32, np.int32, "signed int", -123456),
+        (ctypes.c_int64, np.int64, "signed long long", -123456789),
+        (ctypes.c_uint8, np.uint8, "unsigned char", 200),
+        (ctypes.c_uint16, np.uint16, "unsigned short", 60000),
+        (ctypes.c_uint32, np.uint32, "unsigned int", 4000000000),
+        (ctypes.c_uint64, np.uint64, "unsigned long long", 0x1_0000_0001),
+        (ctypes.c_float, np.float32, "float", 3.14),
+        (ctypes.c_double, np.float64, "double", 2.718281828),
+        # numpy scalar subclass — prepare_numpy_arg fallback
+        (np.float32, np.float32, "float", 3.14),
+    ],
+    ids=[
+        "ctypes_bool",
+        "ctypes_int8",
+        "ctypes_int16",
+        "ctypes_int32",
+        "ctypes_int64",
+        "ctypes_uint8",
+        "ctypes_uint16",
+        "ctypes_uint32",
+        "ctypes_uint64",
+        "ctypes_float",
+        "ctypes_double",
+        "numpy_float32",
     ],
-    ids=["ctypes_subclass", "numpy_subclass"],
 )
-def test_launch_scalar_argument_subclass_fallback(scalar_kind, np_dtype, cpp_type, raw_value):
-    """Subclassed scalar arguments survive fallback handling and reach the kernel."""
-    if scalar_kind == "ctypes":
-
-        class Subclassed(ctypes.c_int32):
-            pass
-    else:
+def test_launch_scalar_argument_subclass_fallback(base_type, np_dtype, cpp_type, raw_value):
+    """Subclassed scalar arguments survive fallback handling and reach the kernel
+    with the correct width/sign. The readback value (not just ptr != 0) guards each
+    fallback branch against marshalling the wrong C type, e.g. uint64 -> uint32_t."""
 
-        class Subclassed(np.float32):
-            pass
+    class Subclassed(base_type):
+        pass
 
     scalar = Subclassed(raw_value)
     expected = np_dtype(raw_value)