Ft testing test fixes

cuda_bindings/tests/test_cufile.py

@@ -138,7 +138,7 @@ def ctx():
     (err,) = cuda.cuCtxSetCurrent(ctx)
     assert err == cuda.CUresult.CUDA_SUCCESS
 
-    yield
+    yield ctx
 
     cuda.cuDevicePrimaryCtxRelease(device)
 

cuda_bindings/tests/test_examples.py

@@ -19,6 +19,7 @@ def test_example(example):
 
     env = os.environ.copy()
     env["CUDA_BINDINGS_SKIP_EXAMPLE"] = "100"
+    env["MPLBACKEND"] = "Agg"  # avoid plt.show() from blocking
 
     process = subprocess.run([sys.executable, example], capture_output=True, env=env)  # noqa: S603
     # returncode is a special value used in the examples to indicate that system requirements are not met.

cuda_core/tests/graph/test_graph_definition_lifetime.py

@@ -685,6 +685,7 @@ def test_event_survives_graph_clone_and_execution(init_cuda):
 # =============================================================================
 
 
+@pytest.mark.thread_unsafe(reason="asserts cleanup on main thread")
 @pytest.mark.agent_authored(model="gpt-5.6")
 def test_user_object_cleanup_is_coalesced_on_python_thread(init_cuda):
     """More than 32 CUDA callbacks drain through one main-thread pending call."""
@@ -1412,6 +1413,7 @@ def test_memcpy_buffer_survives_close(init_cuda):
     assert list(out) == [0xCD] * 4
 
 
+@pytest.mark.thread_unsafe(reason="deferred cleanup on main thread which would wait")
 @pytest.mark.agent_authored(model="claude-opus-4.8")
 def test_memcpy_buffer_allocations_released_after_graph_destroyed(init_cuda):
     """Destroying the graph frees both memcpy operand allocations.

cuda_core/tests/graph/test_graph_memory_resource.py

@@ -5,7 +5,7 @@
 """Tests for GraphMemoryResource allocation and attributes during graph capture."""
 
 import pytest
-from helpers.buffers import compare_buffer_to_constant, make_scratch_buffer, set_buffer
+from helpers.buffers import compare_buffer_to_constant, make_scratch_buffer, set_buffer, thread_unsafe_on_windows
 from helpers.memory import xfail_on_graph_mempool_oom
 
 from cuda.core import (
@@ -21,6 +21,13 @@
 from cuda.core.graph import GraphCompleteOptions
 from cuda_python_test_helpers import IS_WINDOWS, IS_WSL
 
+# NOTE(seberg): "global" mode seems thread-unsafe even when working on stream
+_GRAPH_MODES = [
+    pytest.param("global", marks=pytest.mark.thread_unsafe(reason="gb instances share stream unsafely")),
+    "thread_local",
+    "relaxed",
+]
+
 
 def _common_kernels_alloc():
     code = """
@@ -80,8 +87,9 @@ def free(self, buffers):
         self.stream.sync()
 
 
-@pytest.mark.parametrize("mode", ["no_graph", "global", "thread_local", "relaxed"])
+@pytest.mark.parametrize("mode", ["no_graph"] + _GRAPH_MODES)
 @pytest.mark.parametrize("action", ["incr", "fill"])
+@thread_unsafe_on_windows
 def test_graph_alloc(mempool_device, mode, action):
     """Test basic graph capture with memory allocated and deallocated by
     GraphMemoryResource.
@@ -130,7 +138,7 @@ def apply_kernels(mr, stream, out):
         assert compare_buffer_to_constant(out, 3)
     else:
         # Capture work, then upload and launch.
-        gb = device.create_graph_builder().begin_building(mode)
+        gb = stream.create_graph_builder().begin_building(mode)
         with xfail_on_graph_mempool_oom(device):
             apply_kernels(mr=gmr, stream=gb, out=out)
             graph = gb.end_building().complete()
@@ -150,7 +158,8 @@ def apply_kernels(mr, stream, out):
 
 
 @pytest.mark.skipif(IS_WINDOWS or IS_WSL, reason="auto_free_on_launch not supported on Windows")
-@pytest.mark.parametrize("mode", ["global", "thread_local", "relaxed"])
+@pytest.mark.parametrize("mode", _GRAPH_MODES)
+@thread_unsafe_on_windows
 def test_graph_alloc_with_output(mempool_device, mode):
     """Test for memory allocated in a graph being used outside the graph."""
     NBYTES = 64
@@ -168,7 +177,7 @@ def test_graph_alloc_with_output(mempool_device, mode):
     # Construct a graph to copy and increment the input. It returns a new
     # buffer allocated within the graph.  The auto_free_on_launch option
     # is required to properly use the output buffer.
-    gb = device.create_graph_builder().begin_building(mode)
+    gb = stream.create_graph_builder().begin_building(mode)
     with xfail_on_graph_mempool_oom(device):
         out = gmr.allocate(NBYTES, stream=gb)
         out.copy_from(in_, stream=gb)
@@ -195,7 +204,8 @@ def test_graph_alloc_with_output(mempool_device, mode):
     assert compare_buffer_to_constant(out, 6)
 
 
-@pytest.mark.parametrize("mode", ["global", "thread_local", "relaxed"])
+@pytest.mark.parametrize("mode", _GRAPH_MODES)
+@pytest.mark.thread_unsafe(reason="gb instances share default stream")
 def test_graph_mem_alloc_zero(mempool_device, mode):
     device = mempool_device
     gb = device.create_graph_builder().begin_building(mode)
@@ -213,7 +223,8 @@ def test_graph_mem_alloc_zero(mempool_device, mode):
     assert buffer.device_id == int(device)
 
 
-@pytest.mark.parametrize("mode", ["global", "thread_local", "relaxed"])
+@pytest.mark.parametrize("mode", _GRAPH_MODES)
+@pytest.mark.thread_unsafe(reason="GMR is shared, so high mark is global")
 def test_graph_mem_set_attributes(mempool_device, mode):
     device = mempool_device
     stream = device.create_stream()
@@ -265,7 +276,7 @@ def test_graph_mem_set_attributes(mempool_device, mode):
     mman.reset()
 
 
-@pytest.mark.parametrize("mode", ["global", "thread_local", "relaxed"])
+@pytest.mark.parametrize("mode", _GRAPH_MODES)
 def test_gmr_check_capture_state(mempool_device, mode):
     """
     Test expected errors (and non-errors) using GraphMemoryResource with graph
@@ -284,7 +295,7 @@ def test_gmr_check_capture_state(mempool_device, mode):
         gmr.allocate(1, stream=stream)
 
     # Capturing
-    gb = device.create_graph_builder().begin_building(mode=mode)
+    gb = stream.create_graph_builder().begin_building(mode=mode)
     with xfail_on_graph_mempool_oom(device):
         gmr.allocate(1, stream=gb)  # no error
         gb.end_building().complete()
@@ -320,7 +331,7 @@ def test_graph_memory_resource_attributes_repr(mempool_device):
     assert "used_mem_high=" in r
 
 
-@pytest.mark.parametrize("mode", ["global", "thread_local", "relaxed"])
+@pytest.mark.parametrize("mode", _GRAPH_MODES)
 def test_dmr_check_capture_state(mempool_device, mode):
     """
     Test expected errors (and non-errors) using DeviceMemoryResource with graph
@@ -334,7 +345,7 @@ def test_dmr_check_capture_state(mempool_device, mode):
     dmr.allocate(1, stream=stream).close()  # no error
 
     # Capturing
-    gb = device.create_graph_builder().begin_building(mode=mode)
+    gb = stream.create_graph_builder().begin_building(mode=mode)
     with pytest.raises(
         RuntimeError,
         match=r"cannot perform memory operations on a capturing "

cuda_core/tests/graph/test_graph_update.py

@@ -196,6 +196,7 @@ def new_callback():
     assert called == ["new"]
 
 
+@pytest.mark.thread_unsafe(reason="deferred cleanup on main thread which would wait")
 @pytest.mark.agent_authored(model="gpt-5.6")
 def test_failed_graph_update_does_not_adopt_attachments(init_cuda):
     """A rejected source graph keeps ownership separate from the exec."""

cuda_core/tests/helpers/buffers.py

@@ -3,11 +3,13 @@
 
 import ctypes
 
+import pytest
+
 from cuda.core import Buffer, Device, MemoryResource
 from cuda.core._stream import Stream_accept
 from cuda.core._utils.cuda_utils import driver, handle_return
 
-from . import libc
+from . import IS_WINDOWS, IS_WSL, libc
 
 __all__ = [
     "DummyDeviceMemoryResource",
@@ -18,9 +20,19 @@
     "compare_equal_buffers",
     "make_instrumented_memory_resource",
     "make_scratch_buffer",
+    "thread_unsafe_on_windows",
 ]
 
 
+def thread_unsafe_on_windows(func):
+    # Tests that use these buffers and access the memory on the host are
+    # thread-unsafe on windows. On windows the GPU must be fully quiescent for host
+    # access to be safe and with threaded tests that would require a barrier.
+    if IS_WINDOWS or IS_WSL:
+        return pytest.mark.thread_unsafe(reason="windows host-access unsafe while GPU is working")
+    return func
+
+
 class StubMemoryResource(MemoryResource):
     """Device-only memory resource for tests that supply a fake pointer."""
 

cuda_core/tests/helpers/latch.py

@@ -1,4 +1,4 @@
-# SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 # SPDX-License-Identifier: Apache-2.0
 
 import ctypes

cuda_core/tests/memory/test_managed_ops.py

@@ -365,6 +365,7 @@ def test_from_handle(self, init_cuda):
         finally:
             plain.close()
 
+    @pytest.mark.thread_unsafe(reason="external_managed_buffer is shared between threads")
     def test_read_mostly_roundtrip(self, external_managed_buffer):
         buf = external_managed_buffer
         assert buf.read_mostly is False
@@ -373,6 +374,7 @@ def test_read_mostly_roundtrip(self, external_managed_buffer):
         buf.read_mostly = False
         assert buf.read_mostly is False
 
+    @pytest.mark.thread_unsafe(reason="external_managed_buffer is shared between threads")
     def test_preferred_location_roundtrip(self, location_ops_device, external_managed_buffer):
         device = location_ops_device
         buf = external_managed_buffer
@@ -387,6 +389,7 @@ def test_preferred_location_roundtrip(self, location_ops_device, external_manage
         buf.preferred_location = None
         assert buf.preferred_location is None
 
+    @pytest.mark.thread_unsafe(reason="external_managed_buffer is shared between threads")
     def test_preferred_location_roundtrip_host_numa(self, location_ops_device):
         """Host(numa_id=N) round-trips correctly on CUDA 13 builds."""
         from cuda.core._utils.version import binding_version
@@ -407,6 +410,7 @@ def test_preferred_location_roundtrip_host_numa(self, location_ops_device):
         finally:
             plain.close()
 
+    @pytest.mark.thread_unsafe(reason="external_managed_buffer is shared between threads")
     def test_accessed_by_add_discard(self, location_ops_device, external_managed_buffer):
         device = location_ops_device
         buf = external_managed_buffer
@@ -418,6 +422,7 @@ def test_accessed_by_add_discard(self, location_ops_device, external_managed_buf
         buf.accessed_by.discard(device)
         assert device not in buf.accessed_by
 
+    @pytest.mark.thread_unsafe(reason="external_managed_buffer is shared between threads")
     def test_accessed_by_mutable_set_interface(self, location_ops_device, external_managed_buffer):
         """Full MutableSet conformance pass on AccessedBySetProxy.
 
@@ -437,6 +442,7 @@ def test_accessed_by_mutable_set_interface(self, location_ops_device, external_m
             non_member=Host(numa_id=0),
         )
 
+    @pytest.mark.thread_unsafe(reason="external_managed_buffer is shared between threads")
     def test_accessed_by_set_assignment(self, location_ops_device, external_managed_buffer):
         device = location_ops_device
         buf = external_managed_buffer

cuda_core/tests/system/test_system_device.py

@@ -377,6 +377,7 @@ def test_c2c_mode_enabled(subtests):
 
 
 @pytest.mark.skipif(helpers.IS_WSL or helpers.IS_WINDOWS, reason="Persistence mode not supported on WSL or Windows")
+@pytest.mark.thread_unsafe(reason="device persistence mode is global state")
 def test_persistence_mode_enabled(subtests):
     for device in system.Device.get_all_devices():
         with subtests.test(device_index=device.index):

cuda_core/tests/test_event.py

@@ -119,6 +119,7 @@ def test_error_timing_recorded():
 
 
 @pytest.mark.skipif(Device().compute_capability.major < 7, reason="__nanosleep is only available starting Volta (sm70)")
+@pytest.mark.thread_unsafe(reason="requires a barrier wait to avoid overlapping pinned latch allocations")
 def test_error_timing_incomplete():
     device = Device()
     device.set_current()
@@ -223,6 +224,7 @@ def test_event_ipc_descriptor_non_ipc(init_cuda):
 
 
 @pytest.mark.skipif(Device().compute_capability.major < 7, reason="__nanosleep is only available starting Volta (sm70)")
+@pytest.mark.thread_unsafe(reason="requires a barrier wait to avoid overlapping pinned latch allocations")
 def test_event_is_done_false(init_cuda):
     """Event.is_done returns False when captured work has not yet completed."""
     device = Device()

cuda_core/tests/test_helpers.py

@@ -6,7 +6,7 @@
 import types
 
 import pytest
-from helpers.buffers import PatternGen, compare_equal_buffers, make_scratch_buffer
+from helpers.buffers import PatternGen, compare_equal_buffers, make_scratch_buffer, thread_unsafe_on_windows
 from helpers.latch import LatchKernel
 from helpers.logging import TimestampedLogger
 from helpers.oom_diagnostics import (
@@ -29,6 +29,7 @@
 
 
 @pytest.mark.skipif(Device().compute_capability.major < 7, reason="__nanosleep is only available starting Volta (sm70)")
+@pytest.mark.thread_unsafe(reason="requires a barrier wait to avoid overlapping pinned latch allocations")
 def test_latchkernel():
     """Test LatchKernel."""
     log = TimestampedLogger(enabled=ENABLE_LOGGING)
@@ -63,6 +64,7 @@ def test_latchkernel():
     under_compute_sanitizer(),
     reason="Too slow under compute-sanitizer (UVM-heavy test).",
 )
+@thread_unsafe_on_windows
 def test_patterngen_seeds():
     """Test PatternGen with seed argument."""
     device = Device()
@@ -81,6 +83,7 @@ def test_patterngen_seeds():
                 pgen.verify_buffer(buffer, seed=j)
 
 
+@thread_unsafe_on_windows
 def test_patterngen_values():
     """Test PatternGen with value argument, also compare_equal_buffers."""
     device = Device()

cuda_core/tests/test_memory.py

@@ -20,6 +20,7 @@
     DummyUnifiedMemoryResource,
     StubMemoryResource,
     make_instrumented_memory_resource,
+    thread_unsafe_on_windows,
 )
 from helpers.constants import POOL_SIZE
 from helpers.memory import (
@@ -283,9 +284,8 @@ def _pattern_bytes(value) -> bytes:
 
 
 @pytest.fixture(params=["device", "unified", "pinned"])
-def fill_env(request):
-    device = Device()
-    device.set_current()
+def fill_env(request, init_cuda):
+    device = init_cuda
     if request.param == "device":
         mr = DummyDeviceMemoryResource(device)
     elif request.param == "unified":
@@ -348,6 +348,7 @@ def fill_env(request):
     )
 
 
+@thread_unsafe_on_windows
 @pytest.mark.parametrize("value,size,exc", _FILL_CASES)
 def test_buffer_fill(fill_env, value, size, exc):
     device, mr = fill_env
@@ -1519,6 +1520,8 @@ def test_managed_memory_resource_with_options(init_cuda):
     device.sync()
     dst_buffer.close()
     src_buffer.close()
+    # TODO(seberg): 2026-06: mr close may be unsafe with incomplete `buf.close()`
+    device.sync()
 
 
 def test_managed_memory_resource_preferred_location_default(init_cuda):

cuda_core/tests/test_multiprocessing_warning.py

@@ -12,6 +12,7 @@
 import warnings
 from unittest.mock import patch
 
+import pytest
 from helpers.constants import POOL_SIZE
 
 from cuda.core import DeviceMemoryResource, DeviceMemoryResourceOptions, EventOptions
@@ -20,6 +21,9 @@
 from cuda.core._memory._ipc import _reduce_allocation_handle
 from cuda.core._utils.cuda_utils import check_multiprocessing_start_method, reset_fork_warning
 
+# We could move these to a (session) fixtures
+pytestmark = pytest.mark.thread_unsafe(reason="all tests use unittest.mock.patch")
+
 
 def test_warn_on_fork_method_device_memory_resource(ipc_device):
     """Test that warning is emitted when DeviceMemoryResource is pickled with fork method."""