bpo-29734: nt._getfinalpathname handle leak (GH-740)

Make sure that failure paths call CloseHandle outside of the function that failed

1 files changed, 56 insertions, 0 deletions

diff --git a/Lib/test/test_os.py b/Lib/test/test_os.py
index aca445f..9e0bef5 100644
--- a/Lib/test/test_os.py
+++ b/Lib/test/test_os.py
@@ -2325,6 +2325,62 @@ class Win32JunctionTests(unittest.TestCase):
         os.unlink(self.junction)
         self.assertFalse(os.path.exists(self.junction))
 
+@unittest.skipUnless(sys.platform == "win32", "Win32 specific tests")
+class Win32NtTests(unittest.TestCase):
+    def setUp(self):
+        from test import support
+        self.nt = support.import_module('nt')
+        pass
+
+    def tearDown(self):
+        pass
+
+    def test_getfinalpathname_handles(self):
+        try:
+            import ctypes, ctypes.wintypes
+        except ImportError:
+            raise unittest.SkipTest('ctypes module is required for this test')
+
+        kernel = ctypes.WinDLL('Kernel32.dll', use_last_error=True)
+        kernel.GetCurrentProcess.restype = ctypes.wintypes.HANDLE
+
+        kernel.GetProcessHandleCount.restype = ctypes.wintypes.BOOL
+        kernel.GetProcessHandleCount.argtypes = (ctypes.wintypes.HANDLE,
+                                                 ctypes.wintypes.LPDWORD)
+
+        # This is a pseudo-handle that doesn't need to be closed
+        hproc = kernel.GetCurrentProcess()
+
+        handle_count = ctypes.wintypes.DWORD()
+        ok = kernel.GetProcessHandleCount(hproc, ctypes.byref(handle_count))
+        self.assertEqual(1, ok)
+
+        before_count = handle_count.value
+
+        # The first two test the error path, __file__ tests the success path
+        filenames = [ r'\\?\C:',
+                      r'\\?\NUL',
+                      r'\\?\CONIN',
+                      __file__ ]
+
+        for i in range(10):
+            for name in filenames:
+                try:
+                    tmp = self.nt._getfinalpathname(name)
+                except:
+                    # Failure is expected
+                    pass
+                try:
+                    tmp = os.stat(name)
+                except:
+                    pass
+
+        ok = kernel.GetProcessHandleCount(hproc, ctypes.byref(handle_count))
+        self.assertEqual(1, ok)
+
+        handle_delta = handle_count.value - before_count
+
+        self.assertEqual(0, handle_delta)
 
 @support.skip_unless_symlink
 class NonLocalSymlinkTests(unittest.TestCase):