gh-154137: Check for Windows handle leaks in regrtest

Add handle_count() to test.support.os_helper.

Author: vstinner

Lib/test/libregrtest/refleak.py

@@ -118,12 +118,15 @@ def get_pooled_int(value):
     rc_deltas = [0] * repcount
     alloc_deltas = [0] * repcount
     fd_deltas = [0] * repcount
+    handle_deltas = [0] * repcount
     getallocatedblocks = sys.getallocatedblocks
     gettotalrefcount = sys.gettotalrefcount
     getunicodeinternedsize = sys.getunicodeinternedsize
     fd_count = os_helper.fd_count
+    handle_count = os_helper.handle_count
     # initialize variables to make pyflakes quiet
     rc_before = alloc_before = fd_before = interned_immortal_before = 0
+    handle_before = 0
 
     if not quiet:
         print("beginning", repcount, "repetitions. Showing number of leaks "
@@ -160,14 +163,17 @@ def get_pooled_int(value):
         alloc_after = getallocatedblocks() - interned_immortal_after
         rc_after = gettotalrefcount()
         fd_after = fd_count()
+        handle_after = handle_count()
 
         rc_deltas[i] = get_pooled_int(rc_after - rc_before)
         alloc_deltas[i] = get_pooled_int(alloc_after - alloc_before)
         fd_deltas[i] = get_pooled_int(fd_after - fd_before)
+        handle_deltas[i] = get_pooled_int(handle_after - handle_before)
 
         if not quiet:
             # use max, not sum, so total_leaks is one of the pooled ints
-            total_leaks = max(rc_deltas[i], alloc_deltas[i], fd_deltas[i])
+            total_leaks = max(rc_deltas[i], alloc_deltas[i],
+                              fd_deltas[i], handle_deltas[i])
             if total_leaks <= 0:
                 symbol = '.'
             elif total_leaks < 10:
@@ -185,6 +191,7 @@ def get_pooled_int(value):
         alloc_before = alloc_after
         rc_before = rc_after
         fd_before = fd_after
+        handle_before = handle_after
         interned_immortal_before = interned_immortal_after
 
         restore_support_xml(xml_filename)
@@ -215,7 +222,8 @@ def check_fd_deltas(deltas):
     for deltas, item_name, checker in [
         (rc_deltas, 'references', check_rc_deltas),
         (alloc_deltas, 'memory blocks', check_rc_deltas),
-        (fd_deltas, 'file descriptors', check_fd_deltas)
+        (fd_deltas, 'file descriptors', check_fd_deltas),
+        (handle_deltas, 'handles', check_rc_deltas),
     ]:
         # ignore warmup runs
         deltas = deltas[warmups:]

Lib/test/support/os_helper.py

@@ -830,3 +830,35 @@ def subst_drive(path):
                     DDD_REMOVE_DEFINITION | DDD_EXACT_MATCH_ON_REMOVE,
                     drive, path):
                 raise ctypes.WinError(ctypes.get_last_error())
+
+
+try:
+    if support.MS_WINDOWS:
+        import ctypes.util
+        kernel32 = ctypes.WinDLL('kernel32', use_last_error=True)
+    else:
+        raise AttributeError
+except (ImportError, AttributeError):
+    def handle_count():
+        return 0
+else:
+    @ctypes.util.wrap_dll_function(kernel32)
+    def GetCurrentProcess() -> ctypes.wintypes.HANDLE:
+        pass
+
+    @ctypes.util.wrap_dll_function(kernel32)
+    def GetProcessHandleCount(khProcess: ctypes.wintypes.HANDLE,
+                              pdwHandleCount: ctypes.wintypes.LPDWORD) -> ctypes.wintypes.BOOL:
+        pass
+
+    del kernel32
+
+    def handle_count():
+        # Pseudo-handle that doesn't need to be closed
+        hproc = GetCurrentProcess()
+
+        handle_count = ctypes.wintypes.DWORD()
+        if not GetProcessHandleCount(hproc, ctypes.byref(handle_count)):
+            raise ctypes.WinError()
+
+        return handle_count.value

Lib/test/test_regrtest.py

@@ -1382,6 +1382,34 @@ def test_leak(self):
         """)
         self.check_leak(code, 'file descriptors')
 
+    @unittest.skipUnless(support.Py_DEBUG, 'need a debug build')
+    def test_huntrleaks_handle_leak(self):
+        # test --huntrleaks for Windows handle leak
+        code = textwrap.dedent("""
+            import unittest
+            import _winapi
+
+            handle = None
+
+            class HandleLeakTest(unittest.TestCase):
+                def test_leak(self):
+                    global handle
+                    if handle is None:
+                        handle = _winapi.CreateFile(
+                                        __file__, _winapi.GENERIC_READ,
+                                        0, _winapi.NULL,
+                                        _winapi.OPEN_EXISTING,
+                                        0, _winapi.NULL)
+                    else:
+                        hproc = _winapi.GetCurrentProcess()
+                        copy = _winapi.DuplicateHandle(
+                                    hproc, handle,
+                                    hproc, 0, False,
+                                    _winapi.DUPLICATE_SAME_ACCESS)
+                    # bug! the new handle is never closed
+        """)
+        self.check_leak(code, 'handles')
+
     def test_list_tests(self):
         # test --list-tests
         tests = [self.create_test() for i in range(5)]

Lib/test/test_support.py

@@ -26,6 +26,10 @@
 from test.support import socket_helper
 from test.support import warnings_helper
 
+if support.MS_WINDOWS:
+    import _winapi
+
+
 TESTFN = os_helper.TESTFN
 
 
@@ -623,6 +627,27 @@ def test_fd_count(self):
             os.close(fd)
         self.assertEqual(more - start, 1)
 
+    @unittest.skipUnless(support.MS_WINDOWS, "test specific to Windows")
+    def test_handle_count(self):
+        handle = _winapi.CreateFile(
+                        __file__, _winapi.GENERIC_READ,
+                        0, _winapi.NULL,
+                        _winapi.OPEN_EXISTING,
+                        0, _winapi.NULL)
+        self.addCleanup(_winapi.CloseHandle, handle)
+
+        start = os_helper.handle_count()
+        hproc = _winapi.GetCurrentProcess()
+        copy = _winapi.DuplicateHandle(
+                    hproc, handle,
+                    hproc, 0, False,
+                    _winapi.DUPLICATE_SAME_ACCESS)
+        try:
+            more = os_helper.handle_count()
+        finally:
+            _winapi.CloseHandle(copy)
+        self.assertEqual(more - start, 1)
+
     def check_print_warning(self, msg, expected):
         stderr = io.StringIO()
         with support.swap_attr(support.print_warning, 'orig_stderr', stderr):

Misc/NEWS.d/next/Tests/2026-07-19-15-59-41.gh-issue-154137.N689bE.rst

@@ -0,0 +1 @@
+Check for Windows handle leaks in regrtest. Patch by Victor Stinner.