test: add tests for arrow_c_stream_capsule destructor behavior

Author: kosiew

python/tests/test_dataframe.py

@@ -1749,6 +1749,48 @@ def test_arrow_c_stream_capsule_released(ctx):
     gc.collect()
 
 
+def test_arrow_c_stream_capsule_clear_destructor(ctx):
+    df = ctx.from_pydict({"a": [1]})
+
+    capsule = df.__arrow_c_stream__()
+
+    set_destructor = ctypes.pythonapi.PyCapsule_SetDestructor
+    set_destructor.restype = ctypes.c_int
+    set_destructor.argtypes = [ctypes.py_object, ctypes.c_void_p]
+    set_destructor(capsule, ctypes.c_void_p(0))
+
+    reader = pa.RecordBatchReader._import_from_c_capsule(capsule)
+
+    del capsule
+    gc.collect()
+
+    table = pa.Table.from_batches(reader)
+    expected = pa.table({"a": [1]})
+    assert table.equals(expected)
+
+
+def test_arrow_c_stream_capsule_manual_destructor_noop(ctx):
+    df = ctx.from_pydict({"a": [1]})
+
+    capsule = df.__arrow_c_stream__()
+
+    get_destructor = ctypes.pythonapi.PyCapsule_GetDestructor
+    get_destructor.restype = ctypes.c_void_p
+    get_destructor.argtypes = [ctypes.py_object]
+    destructor_ptr = get_destructor(capsule)
+    destructor = ctypes.CFUNCTYPE(None, ctypes.py_object)(destructor_ptr)
+
+    reader = pa.RecordBatchReader._import_from_c_capsule(capsule)
+    reader.read_all()
+
+    destructor(capsule)
+    destructor(capsule)
+
+    del reader
+    del capsule
+    gc.collect()
+
+
 def test_to_pylist(df):
     # Convert datafusion dataframe to Python list
     pylist = df.to_pylist()