Update VT_ARRAY | VT_RECORD tests.

comtypes/_safearray.py

@@ -2,9 +2,15 @@
 
 from ctypes import POINTER, Structure, WinDLL, byref, c_uint, c_ushort, c_void_p
 from ctypes.wintypes import DWORD, LONG, UINT, USHORT
+from typing import TYPE_CHECKING
 
 from comtypes import GUID, HRESULT
 
+if TYPE_CHECKING:
+    from ctypes import _Pointer
+
+    from comtypes import hints  # type: ignore
+
 ################################################################
 # if __debug__:
 #     from ctypeslib.dynamic_module import include
@@ -77,17 +83,20 @@ def SafeArrayGetVartype(pa):
     _SafeArrayGetVartype(pa, result)
     return result.value
 
-def SafeArrayGetRecordInfo(pa):
+
+_SafeArrayGetRecordInfo = _oleaut32.SafeArrayGetRecordInfo
+_SafeArrayGetRecordInfo.restype = HRESULT
+_SafeArrayGetRecordInfo.argtypes = [POINTER(SAFEARRAY), POINTER(c_void_p)]
+
+
+def SafeArrayGetRecordInfo(pa: "_Pointer[tagSAFEARRAY]") -> "hints.IRecordInfo":
     # Defined here to avoid a circular import
     from comtypes.typeinfo import IRecordInfo
 
-    _SafeArrayGetRecordInfo = _oleaut32.SafeArrayGetRecordInfo
-    _SafeArrayGetRecordInfo.restype = HRESULT
-    _SafeArrayGetRecordInfo.argtypes = [POINTER(SAFEARRAY), POINTER(POINTER(IRecordInfo))]
-
     result = POINTER(IRecordInfo)()
     _SafeArrayGetRecordInfo(pa, byref(result))
-    return result.value
+    return result  # type: ignore
+
 
 SafeArrayGetElement = _oleaut32.SafeArrayGetElement
 SafeArrayGetElement.restype = HRESULT

comtypes/hints.pyi

@@ -36,7 +36,9 @@ from comtypes import IUnknown as IUnknown, COMObject as COMObject, GUID as GUID
 from comtypes.automation import IDispatch as IDispatch, VARIANT as VARIANT
 from comtypes.server import IClassFactory as IClassFactory
 from comtypes.server import localserver as localserver
-from comtypes.typeinfo import ITypeInfo as ITypeInfo, ITypeLib as ITypeLib
+from comtypes.typeinfo import IRecordInfo as IRecordInfo
+from comtypes.typeinfo import ITypeInfo as ITypeInfo
+from comtypes.typeinfo import ITypeLib as ITypeLib
 from comtypes._safearray import tagSAFEARRAY as tagSAFEARRAY
 
 Incomplete: TypeAlias = Any

comtypes/test/test_record_safearray_bstr.py

@@ -4,10 +4,18 @@
 strings) between Python and a COM server.
 """
 
+import faulthandler
 import gc
+import io
+import multiprocessing
+import re
+import sys
+import tempfile
 import unittest
 from _ctypes import COMError
-from typing import TYPE_CHECKING, ClassVar, TypeVar
+from collections.abc import Callable
+from pathlib import Path
+from typing import TYPE_CHECKING, Any, ClassVar, Optional, TypeVar
 from typing import Union as _UnionT
 
 from comtypes import CLSCTX_LOCAL_SERVER, IUnknown, _safearray
@@ -297,14 +305,12 @@ def test_wrap_record(self):
         ifc = _create_server(self.interface)
         record = _create_record(id=500, seq=0, text="WrapBstr-500-0", checksum=50708)
 
-        with self.assertRaises(
-            KeyError,
-            msg="CHARACTERIZATION(#63): "
-            "If https://github.com/enthought/comtypes/issues/63 is fixed, "
-            "this failure is expected; change it to a normal-path test.",
-        ) as cm:
-            ifc.WrapBstrRecord(record)
-        self.assertEqual(str(cm.exception), "36")
+        result = ifc.WrapBstrRecord(record)
+        self.assertEqual(len(result), 1)
+        self.assertEqual(result[0].id, 500)
+        self.assertEqual(result[0].seq, 0)
+        self.assertEqual(str(result[0].text), "WrapBstr-500-0")
+        self.assertEqual(result[0].checksum, 50708)
 
         del record
         gc.collect()
@@ -387,6 +393,7 @@ def test_first_record_empty_safearray_by_create(self):
         del records
         gc.collect()
 
+    @unittest.skipIf(sys.version_info >= (3, 14), "CRASH! CHARACTERIZATION(#63)")
     def test_reverse_records_length_3(self):
         ifc = _create_server(self.interface)
 
@@ -397,18 +404,96 @@ def test_reverse_records_length_3(self):
         ]
         lp = _get_bstr_record_sa_type().from_param(records)
 
-        with self.assertRaises(
-            KeyError,
-            msg="CHARACTERIZATION(#63): "
-            "If https://github.com/enthought/comtypes/issues/63 is fixed, "
-            "this failure is expected; change it to a normal-path test.",
-        ) as cm:
-            ifc.ReverseBstrRecords(lp)
-        self.assertEqual(str(cm.exception), "36")
+        result = ifc.ReverseBstrRecords(lp)
+        self.assertEqual(len(result), 3)
+        self.assertEqual(
+            [(r.id, r.seq, r.text, r.checksum) for r in result],
+            [(s.id, s.seq, s.text, s.checksum) for s in reversed(records)],
+        )
 
         del records
         gc.collect()
 
 
+def enable_fault_log(tmpfile: Path) -> io.TextIOWrapper:
+    """Send faulthandler output to `tmpfile` until the process exits."""
+    f = tmpfile.open(mode="w", encoding="utf-8")
+    # We keep `faulthandler` enabled and the file open until the process
+    # exits so that crash information can be captured.
+    # It is only enabled in the child process and does not affect the main
+    # process, so there is no need to worry about forgetting to close and
+    # disable.
+    faulthandler.enable(file=f)
+    return f
+
+
+def run_in_mp(fn: Callable[[Path], Any]) -> tuple[Optional[int], str]:
+    """Run a function in a separate process and return exit code and output.
+
+    Using spawn isolates the process so crashes like segfaults do not affect
+    the main test process.
+    """
+    with tempfile.TemporaryDirectory() as td:
+        tmpdir = Path(td)
+        tmpfile = tmpdir / "fault.txt"
+        ctx = multiprocessing.get_context("spawn")
+        process = ctx.Process(target=fn, args=(tmpfile,))
+        process.start()
+        process.join()
+        output = tmpfile.read_text()
+    return process.exitcode, output
+
+
+def call_reverse_bstr_records(tmpfile: Path) -> None:
+    enable_fault_log(tmpfile)
+    ifc = _create_server(ComtypesCppTestSrvLib.IDispRecordSafearrayTest)
+
+    records = [
+        _create_record(id=401, seq=0, text="RecSafeArray-401-0", checksum=40808),
+        _create_record(id=402, seq=1, text="RecSafeArray-402-1", checksum=40918),
+        _create_record(id=403, seq=2, text="RecSafeArray-403-2", checksum=41028),
+    ]
+    lp = _get_bstr_record_sa_type().from_param(records)
+
+    result = ifc.ReverseBstrRecords(lp)
+
+    del records
+    del result
+    del lp
+    gc.collect()
+
+
+# Windows NTSTATUS values (unsigned 32-bit).
+# Python's process.exitcode may return these as unsigned integers
+# (e.g., 3221226356 = 0xC0000374) depending on the platform/Python version.
+# We mask with 0xFFFFFFFF to ensure consistent comparison.
+STATUS_HEAP_CORRUPTION = 0xC0000374
+STATUS_ACCESS_VIOLATION = 0xC0000005
+STATUS_STACK_BUFFER_OVERRUN = 0xC0000409
+
+
+@unittest.skipIf(IMPORT_FAILED, "This depends on the out of process COM-server.")
+class TestOwnership(unittest.TestCase):
+    @unittest.skipIf(sys.version_info < (3, 14), "NOT CRASH! CHARACTERIZATION(#63)")
+    def test_bstr_record_safearray_ownership(self):
+        exitcode, output = run_in_mp(call_reverse_bstr_records)
+        self.assertIsNotNone(exitcode)
+        # Python's process.exitcode on Windows may return the exit code as
+        # either a signed or unsigned 32-bit integer. Masking with 0xFFFFFFFF
+        # normalizes it to the unsigned 32-bit representation for comparison
+        # with the NTSTATUS constants defined above.
+        self.assertEqual(
+            exitcode & 0xFFFFFFFF,  # type: ignore
+            STATUS_HEAP_CORRUPTION,
+            msg="CHARACTERIZATION(#63): "
+            "If https://github.com/enthought/comtypes/issues/63 is fixed, "
+            "this failure is expected; change it to a normal-path test.",
+        )
+        ptn = re.compile(
+            r'test_record_safearray_bstr\.py", line \d+ in call_reverse_bstr_records'
+        )
+        self.assertTrue(bool(ptn.search(output)))
+
+
 if __name__ == "__main__":
     unittest.main()

comtypes/test/test_record_safearray_value.py

@@ -292,14 +292,11 @@ def test_wrap_record(self):
         ifc = _create_server(self.interface)
         record = _create_record(id=300, seq=0, checksum=30042)
 
-        with self.assertRaises(
-            KeyError,
-            msg="CHARACTERIZATION(#63): "
-            "If https://github.com/enthought/comtypes/issues/63 is fixed, "
-            "this failure is expected; change it to a normal-path test.",
-        ) as cm:
-            ifc.WrapValueRecord(record)
-        self.assertEqual(str(cm.exception), "36")
+        result = ifc.WrapValueRecord(record)
+        self.assertEqual(len(result), 1)
+        self.assertEqual(result[0].id, 300)
+        self.assertEqual(result[0].seq, 0)
+        self.assertEqual(result[0].checksum, 30042)
 
         del record
         gc.collect()
@@ -391,14 +388,12 @@ def test_reverse_records_length_3(self):
         ]
         lp = _get_value_record_sa_type().from_param(records)
 
-        with self.assertRaises(
-            KeyError,
-            msg="CHARACTERIZATION(#63): "
-            "If https://github.com/enthought/comtypes/issues/63 is fixed, "
-            "this failure is expected; change it to a normal-path test.",
-        ) as cm:
-            ifc.ReverseValueRecords(lp)
-        self.assertEqual(str(cm.exception), "36")
+        result = ifc.ReverseValueRecords(lp)
+        self.assertEqual(len(result), 3)
+        self.assertEqual(
+            [(r.id, r.seq, r.checksum) for r in result],
+            [(s.id, s.seq, s.checksum) for s in reversed(records)],
+        )
 
         del records
         gc.collect()