gh-154137: Check for Windows handle leaks in regrtest

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

@@ -797,36 +797,78 @@ def __exit__(self, *ignore_exc):
 
 try:
     if support.MS_WINDOWS:
-        import ctypes
+        import ctypes.util
         kernel32 = ctypes.WinDLL('kernel32', use_last_error=True)
-
-        ERROR_FILE_NOT_FOUND = 2
-        DDD_REMOVE_DEFINITION = 2
-        DDD_EXACT_MATCH_ON_REMOVE = 4
-        DDD_NO_BROADCAST_SYSTEM = 8
     else:
         raise AttributeError
 except (ImportError, AttributeError):
     def subst_drive(path):
         raise unittest.SkipTest('ctypes or kernel32 is not available')
+
+    def handle_count():
+        return 0
 else:
+    ERROR_FILE_NOT_FOUND = 2
+    DDD_REMOVE_DEFINITION = 2
+    DDD_EXACT_MATCH_ON_REMOVE = 4
+    DDD_NO_BROADCAST_SYSTEM = 8
+
+
+    @ctypes.util.wrap_dll_function(kernel32)
+    def DefineDosDeviceW(
+        dwFlags: ctypes.wintypes.DWORD,
+        lpDeviceName: ctypes.c_wchar_p,
+        lpTargetPath: ctypes.c_wchar_p,
+    ) -> ctypes.wintypes.BOOL:
+        pass
+
+    @ctypes.util.wrap_dll_function(kernel32)
+    def QueryDosDeviceW(
+        lpDeviceName: ctypes.c_wchar_p,
+        lpTargetPath: ctypes.c_wchar_p,
+        ucchMax: ctypes.wintypes.DWORD,
+    ) -> ctypes.wintypes.DWORD:
+        pass
+
     @contextlib.contextmanager
     def subst_drive(path):
         """Temporarily yield a substitute drive for a given path."""
         for c in reversed(string.ascii_uppercase):
             drive = f'{c}:'
-            if (not kernel32.QueryDosDeviceW(drive, None, 0) and
+            if (not QueryDosDeviceW(drive, None, 0) and
                     ctypes.get_last_error() == ERROR_FILE_NOT_FOUND):
                 break
         else:
             raise unittest.SkipTest('no available logical drive')
-        if not kernel32.DefineDosDeviceW(
-                DDD_NO_BROADCAST_SYSTEM, drive, path):
+
+        if not DefineDosDeviceW(DDD_NO_BROADCAST_SYSTEM, drive, path):
             raise ctypes.WinError(ctypes.get_last_error())
+
         try:
             yield drive
         finally:
-            if not kernel32.DefineDosDeviceW(
-                    DDD_REMOVE_DEFINITION | DDD_EXACT_MATCH_ON_REMOVE,
-                    drive, path):
+            flags = DDD_REMOVE_DEFINITION | DDD_EXACT_MATCH_ON_REMOVE
+            if not DefineDosDeviceW(flags, drive, path):
                 raise ctypes.WinError(ctypes.get_last_error())
+
+
+    @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(ctypes.get_last_error())
+
+        return handle_count.value

Lib/test/test_os/test_windows.py

@@ -457,28 +457,8 @@ def test_unlink_removes_junction(self):
 class Win32NtTests(unittest.TestCase):
     def test_getfinalpathname_handles(self):
         nt = import_helper.import_module('nt')
-        ctypes = import_helper.import_module('ctypes')
-        # Ruff false positive -- it thinks we're redefining `ctypes` here
-        import ctypes.wintypes  # noqa: F811
 
-        kernel = ctypes.WinDLL('Kernel32.dll', use_last_error=True)
-        @ctypes.util.wrap_dll_function(kernel)
-        def GetCurrentProcess() -> ctypes.wintypes.HANDLE:
-            pass
-
-        @ctypes.util.wrap_dll_function(kernel)
-        def GetProcessHandleCount(khProcess: ctypes.wintypes.HANDLE,
-                                  pdwHandleCount: ctypes.wintypes.LPDWORD) -> ctypes.wintypes.BOOL:
-            pass
-
-        # This is a pseudo-handle that doesn't need to be closed
-        hproc = GetCurrentProcess()
-
-        handle_count = ctypes.wintypes.DWORD()
-        ok = GetProcessHandleCount(hproc, ctypes.byref(handle_count))
-        self.assertEqual(1, ok)
-
-        before_count = handle_count.value
+        before_count = os_helper.handle_count()
 
         # The first two test the error path, __file__ tests the success path
         filenames = [
@@ -500,10 +480,7 @@ def GetProcessHandleCount(khProcess: ctypes.wintypes.HANDLE,
                 except Exception:
                     pass
 
-        ok = kernel.GetProcessHandleCount(hproc, ctypes.byref(handle_count))
-        self.assertEqual(1, ok)
-
-        handle_delta = handle_count.value - before_count
+        handle_delta = os_helper.handle_count() - before_count
 
         self.assertEqual(0, handle_delta)
 

Lib/test/test_regrtest.py

@@ -1382,6 +1382,35 @@ def test_leak(self):
         """)
         self.check_leak(code, 'file descriptors')
 
+    @unittest.skipUnless(support.Py_DEBUG, 'need a debug build')
+    @unittest.skipUnless(support.MS_WINDOWS, 'test specific to Windows')
+    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.