import gc
import numpy as np
import pytest
import scirs2
class TestToDlpackFromDlpackRoundTrip:
def test_1d_float64_round_trip(self):
a = np.array([1.0, 2.0, 3.0, 4.5, -7.25], dtype=np.float64)
capsule = scirs2.to_dlpack(a)
back = scirs2.from_dlpack(capsule)
assert np.array_equal(a, back)
def test_2d_contiguous_round_trip(self):
a = np.arange(12, dtype=np.float64).reshape(3, 4)
capsule = scirs2.to_dlpack(a)
back = scirs2.from_dlpack(capsule)
assert back.shape == (3, 4)
assert np.array_equal(a, back)
class TestFromDlpackStrideHandling:
def test_transposed_2d_view(self):
a = np.arange(12, dtype=np.float64).reshape(3, 4)
a_t = a.T
assert not a_t.flags["C_CONTIGUOUS"]
back = scirs2.from_dlpack(a_t.__dlpack__())
assert np.array_equal(a_t, back)
def test_negative_stride_reversed_slice(self):
a = np.arange(10, dtype=np.float64)[::-1]
back = scirs2.from_dlpack(a.__dlpack__())
assert np.array_equal(a, back)
def test_row_strided_2d_slice(self):
a = np.arange(20, dtype=np.float64).reshape(5, 4)[::2, :]
back = scirs2.from_dlpack(a.__dlpack__())
assert np.array_equal(a, back)
@pytest.mark.parametrize("dtype", [np.int32, np.uint8, np.int64, np.float32])
def test_supported_dtypes(self, dtype):
a = np.array([0, 1, 2, 3, 4], dtype=dtype)
back = scirs2.from_dlpack(a.__dlpack__())
assert np.array_equal(a, back.astype(dtype))
class TestCapsuleLifecycleMemorySafety:
def test_to_dlpack_alone_garbage_collected_without_consumption(self):
for _ in range(20):
arr = np.random.rand(4, 4)
capsule = scirs2.to_dlpack(arr)
del capsule
gc.collect()
def test_consumed_capsule_garbage_collected_cleanly(self):
for _ in range(20):
arr = np.random.rand(7, 5)
capsule = scirs2.to_dlpack(arr)
back = scirs2.from_dlpack(capsule)
assert np.allclose(arr, back)
del capsule, back
gc.collect()
def test_mixed_consumed_and_unconsumed_capsules(self):
kept_alive = []
for i in range(20):
arr = np.random.rand(3, 3)
capsule = scirs2.to_dlpack(arr)
if i % 2 == 0:
back = scirs2.from_dlpack(capsule)
assert np.allclose(arr, back)
else:
kept_alive.append(capsule)
del kept_alive
gc.collect()