diff --git a/cuda_core/tests/graph/test_graph_definition.py b/cuda_core/tests/graph/test_graph_definition.py index 50d4b4ac253..1d63504c7c4 100644 --- a/cuda_core/tests/graph/test_graph_definition.py +++ b/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.""" diff --git a/cuda_core/tests/system/test_system_device.py b/cuda_core/tests/system/test_system_device.py index b5fe8cccbfa..b8fe3505674 100644 --- a/cuda_core/tests/system/test_system_device.py +++ b/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): diff --git a/cuda_core/tests/test_checkpoint.py b/cuda_core/tests/test_checkpoint.py index 5e70a162320..ff727eb9fff 100644 --- a/cuda_core/tests/test_checkpoint.py +++ b/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) ----------------------------------- diff --git a/cuda_core/tests/test_green_context.py b/cuda_core/tests/test_green_context.py index 999627a901a..d0526a0231f 100644 --- a/cuda_core/tests/test_green_context.py +++ b/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) diff --git a/cuda_core/tests/test_launcher.py b/cuda_core/tests/test_launcher.py index 942952d29b8..08f2c9e041d 100644 --- a/cuda_core/tests/test_launcher.py +++ b/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) diff --git a/cuda_core/tests/test_memory.py b/cuda_core/tests/test_memory.py index 5ef48919a50..c02a88e4a28 100644 --- a/cuda_core/tests/test_memory.py +++ b/cuda_core/tests/test_memory.py @@ -1899,3 +1899,73 @@ def test_dmr_peer_accessible_by_setter_empty(mempool_device): assert set(mr.peer_accessible_by) == set() mr.peer_accessible_by = [] assert set(mr.peer_accessible_by) == set() + + +@pytest.mark.agent_authored(model="claude-opus-4.8") +def test_mempool_attributes_cannot_instantiate_directly(): + """_MemPoolAttributes cannot be instantiated directly.""" + from cuda.core._memory._memory_pool import _MemPoolAttributes + + with pytest.raises(RuntimeError, match="cannot be instantiated directly"): + _MemPoolAttributes() + + +@pytest.mark.agent_authored(model="claude-opus-4.8") +def test_dmr_handle_and_ownership(mempool_device): + """An options-created pool is handle-owning with a live handle; wrapping the device's current pool is non-owning.""" + owned = DeviceMemoryResource(mempool_device, DeviceMemoryResourceOptions(max_size=POOL_SIZE)) + assert owned.is_handle_owned is True + handle = owned.handle + assert handle is not None + assert int(handle) != 0 + + non_owned = DeviceMemoryResource(mempool_device) + assert non_owned.is_handle_owned is False + + +@pytest.mark.agent_authored(model="claude-opus-4.8") +def test_dmr_deallocate_frees_pool_pointer(mempool_device): + """Closing a Buffer.from_handle(..., mr=mr) view frees the pointer via the Python + _MemPool.deallocate path; the pool's in-use bytes drop back.""" + dev = mempool_device + stream = dev.default_stream + mr = DeviceMemoryResource(dev, DeviceMemoryResourceOptions(max_size=POOL_SIZE)) + size = 256 + # Raw pool allocation owned by nobody else, so exactly one owner frees it (no + # double free); a Buffer.from_handle view then routes teardown through the + # Python deallocate path that mr.allocate()'s C++-direct free would skip. + ptr = handle_return(driver.cuMemAllocFromPoolAsync(size, mr.handle, stream.handle)) + stream.sync() + used_after_alloc = mr.attributes.used_mem_current + assert used_after_alloc >= size + buf = Buffer.from_handle(int(ptr), size, mr=mr) + buf.close(stream) + stream.sync() + assert int(buf.handle) == 0 + # In-use bytes fell back, so the pointer was actually returned (buf.handle == 0 + # alone wouldn't prove it: the deleter callback swallows a failed free). + assert mr.attributes.used_mem_current < used_after_alloc + + +@pytest.mark.agent_authored(model="claude-opus-4.8") +def test_dmr_close_is_idempotent(mempool_device): + """Closing an owned DeviceMemoryResource twice is safe (the second close is a no-op).""" + mr = DeviceMemoryResource(mempool_device, DeviceMemoryResourceOptions(max_size=POOL_SIZE)) + assert mr.is_handle_owned is True + assert int(mr.handle) != 0 + mr.close() + # First close releases the pool handle itself, not just ownership. + assert int(mr.handle) == 0 + assert mr.is_handle_owned is False + mr.close() # no-op on the now-null handle + assert int(mr.handle) == 0 + assert mr.is_handle_owned is False + + +@pytest.mark.agent_authored(model="claude-opus-4.8") +def test_dmr_ipc_enabled_unsupported_raises(mempool_device): + """Requesting an IPC-enabled pool where memory IPC is unsupported raises RuntimeError.""" + if not IS_WINDOWS: + pytest.skip("memory IPC is supported on this platform; unsupported-raise path is Windows-only") + with pytest.raises(RuntimeError, match="IPC is not available"): + DeviceMemoryResource(mempool_device, DeviceMemoryResourceOptions(ipc_enabled=True)) diff --git a/cuda_core/tests/test_stream.py b/cuda_core/tests/test_stream.py index 49e372c9d53..39717861c51 100644 --- a/cuda_core/tests/test_stream.py +++ b/cuda_core/tests/test_stream.py @@ -303,3 +303,105 @@ def test_default_stream_consistency(init_cuda): # Should be same object (or at least equal) assert default1 == default2 assert hash(default1) == hash(default2) + + +class _BadStreamProtocol: + """Object whose __cuda_stream__ (a method) returns a malformed value.""" + + def __init__(self, value): + self._value = value + + def __cuda_stream__(self): + return self._value + + +class _AttrStreamProtocol: + """Object implementing __cuda_stream__ as an attribute (deprecated form) + rather than a method; the tuple length is wrong so resolution stops before + any GPU work.""" + + __cuda_stream__ = (0, 1, 2) + + +@pytest.mark.agent_authored(model="claude-opus-4.8") +def test_stream_init_rejects_obj_and_options(): + """Stream._init rejects supplying both a foreign object and options.""" + from cuda.core._stream import Stream + + with pytest.raises(ValueError, match="obj and options cannot be both specified"): + Stream._init(obj=_BadStreamProtocol((0, 0)), options=StreamOptions()) + + +@pytest.mark.agent_authored(model="claude-opus-4.8") +@pytest.mark.parametrize( + "value,match", + [ + ((0, 1, 2), "must return a sequence with 2 elements"), # wrong length + (5, "must return a sequence with 2 elements"), # not a sequence + ((1, 123), r"first element of the sequence.*must be 0"), # bad version + ], +) +def test_stream_init_rejects_bad_cuda_stream_protocol(value, match): + """A foreign object whose __cuda_stream__ returns a malformed value is + rejected before any handle is created.""" + from cuda.core._stream import Stream + + with pytest.raises(RuntimeError, match=match): + Stream._init(obj=_BadStreamProtocol(value)) + + +@pytest.mark.agent_authored(model="claude-opus-4.8") +def test_stream_init_warns_on_attribute_cuda_stream_protocol(): + """Implementing __cuda_stream__ as an attribute (not a method) is deprecated: + resolution emits a DeprecationWarning and then still rejects the malformed + (wrong-length) value with a RuntimeError.""" + from cuda.core._stream import Stream + + with ( + pytest.warns(DeprecationWarning, match="must be implemented as a method"), + pytest.raises(RuntimeError, match="must return a sequence with 2 elements"), + ): + Stream._init(obj=_AttrStreamProtocol()) + + +@pytest.mark.agent_authored(model="claude-opus-4.8") +def test_stream_init_from_existing_stream_object(init_cuda): + """Passing an existing Stream as the foreign object yields a borrowed stream over the same handle.""" + from cuda.core._stream import Stream + + src = Device().create_stream(options=StreamOptions()) + borrowed = Stream._init(obj=src) + assert int(borrowed.handle) == int(src.handle) + borrowed.close() + src.close() + + +@pytest.mark.agent_authored(model="claude-opus-4.8") +def test_stream_from_handle_lazy_flag_and_priority_queries(init_cuda): + """A from_handle stream reports is_nonblocking and priority read back from the + driver (matching the source stream), not the constructor defaults.""" + from cuda.core._stream import Stream + + # priority=-1 (not the default 0) so the value proves the driver was actually queried. + real = Device().create_stream(options=StreamOptions(nonblocking=True, priority=-1)) + wrapped = Stream.from_handle(int(real.handle)) + assert wrapped.is_nonblocking is True + assert wrapped.priority == -1 + wrapped.close() + real.close() + + +@pytest.mark.agent_authored(model="claude-opus-4.8") +@pytest.mark.thread_unsafe( + reason="mutates the process-global CUDA_PYTHON_CUDA_PER_THREAD_DEFAULT_STREAM env var that default_stream() reads live" +) +def test_default_stream_per_thread_when_env_set(monkeypatch): + """default_stream() returns the per-thread default stream when + CUDA_PYTHON_CUDA_PER_THREAD_DEFAULT_STREAM is set to a nonzero value, and the + legacy default stream otherwise.""" + from cuda.core._stream import default_stream + + monkeypatch.setenv("CUDA_PYTHON_CUDA_PER_THREAD_DEFAULT_STREAM", "1") + assert default_stream() is PER_THREAD_DEFAULT_STREAM + monkeypatch.delenv("CUDA_PYTHON_CUDA_PER_THREAD_DEFAULT_STREAM", raising=False) + assert default_stream() is LEGACY_DEFAULT_STREAM diff --git a/cuda_core/tests/test_tensor_map.py b/cuda_core/tests/test_tensor_map.py index 7abbaadb483..8670e910075 100644 --- a/cuda_core/tests/test_tensor_map.py +++ b/cuda_core/tests/test_tensor_map.py @@ -20,7 +20,9 @@ TensorMapL2Promotion, TensorMapOOBFill, TensorMapSwizzle, + _coerce_tensor_map_descriptor_options, _require_view_device, + _resolve_data_type, ) from cuda.core.utils import StridedMemoryView @@ -649,3 +651,67 @@ def test_from_im2col_wide_rank_validation(self, dev, skip_if_no_im2col_wide): pixels_per_column=4, data_type=TensorMapDataType.FLOAT32, ) + + +class _DtypeView: + """Minimal stand-in for a StridedMemoryView exposing only ``.dtype``. + + ``_resolve_data_type`` reads nothing else off the view, so this keeps the + host-only tests free of any GPU allocation. + """ + + def __init__(self, dtype): + self.dtype = dtype + + +@pytest.mark.agent_authored(model="claude-opus-4.8") +class TestTensorMapHelpers: + """Host-only coverage for the arg-marshalling helpers' input-validation branches. + + The happy-path normalize/coerce/resolve/stride cases are covered by the TMA + factory-method tests above once they run on TMA-capable hardware (e.g. the H200 + coverage runner). Only the rejection branches those tests never hit — real + devices never feed bad inputs — are pinned here. + """ + + # Rejected by the public TensorMapDescriptorOptions(...) constructor, whose + # __post_init__ runs the normalize/require-enum helpers. + @pytest.mark.parametrize( + ("kwargs", "match"), + [ + (dict(box_dim=5), "box_dim must be a tuple of ints"), + (dict(box_dim=(1, "x", 3)), r"box_dim\[1\] must be an int"), + (dict(box_dim=(32,), swizzle=2), "swizzle must be a TensorMapSwizzle"), + (dict(box_dim=(32,), interleave=0), "interleave must be a TensorMapInterleave"), + ], + ids=["box_dim_non_iterable", "box_dim_non_int_element", "swizzle_wrong_type", "interleave_wrong_type"], + ) + def test_options_rejects_invalid(self, kwargs, match): + with pytest.raises(TypeError, match=match): + TensorMapDescriptorOptions(**kwargs) + + @pytest.mark.parametrize( + ("view_dtype", "data_type", "match"), + [ + (None, np.complex128, "Unsupported dtype"), # explicit unsupported dtype + (None, None, "Cannot infer TMA data type"), # nothing to infer from + (np.dtype(np.complex64), None, "Unsupported dtype"), # view's dtype unsupported + ], + ids=["explicit_unsupported", "cannot_infer", "view_dtype_unsupported"], + ) + def test_resolve_data_type_rejects(self, view_dtype, data_type, match): + with pytest.raises(ValueError, match=match): + _resolve_data_type(_DtypeView(view_dtype), data_type) + + def test_coerce_requires_box_dim_without_options(self): + with pytest.raises(TypeError, match="box_dim is required unless options is provided"): + _coerce_tensor_map_descriptor_options( + None, + None, + element_strides=None, + data_type=None, + interleave=TensorMapInterleave.NONE, + swizzle=TensorMapSwizzle.NONE, + l2_promotion=TensorMapL2Promotion.NONE, + oob_fill=TensorMapOOBFill.NONE, + )