bpo-46785: Fix race condition between os.stat() and unlink on Windows (GH-31858)

1 files changed, 44 insertions, 0 deletions

diff --git a/Lib/test/test_os.py b/Lib/test/test_os.py
index 5f73af8..36ad587 100644
--- a/Lib/test/test_os.py
+++ b/Lib/test/test_os.py
@@ -24,6 +24,7 @@ import subprocess
 import sys
 import sysconfig
 import tempfile
+import textwrap
 import time
 import types
 import unittest
@@ -2883,6 +2884,49 @@ class Win32NtTests(unittest.TestCase):
 
         self.assertEqual(0, handle_delta)
 
+    @support.requires_subprocess()
+    def test_stat_unlink_race(self):
+        # bpo-46785: the implementation of os.stat() falls back to reading
+        # the parent directory if CreateFileW() fails with a permission
+        # error. If reading the parent directory fails because the file or
+        # directory are subsequently unlinked, or because the volume or
+        # share are no longer available, then the original permission error
+        # should not be restored.
+        filename =  os_helper.TESTFN
+        self.addCleanup(os_helper.unlink, filename)
+        deadline = time.time() + 5
+        command = textwrap.dedent("""\
+            import os
+            import sys
+            import time
+
+            filename = sys.argv[1]
+            deadline = float(sys.argv[2])
+
+            while time.time() < deadline:
+                try:
+                    with open(filename, "w") as f:
+                        pass
+                except OSError:
+                    pass
+                try:
+                    os.remove(filename)
+                except OSError:
+                    pass
+            """)
+
+        with subprocess.Popen([sys.executable, '-c', command, filename, str(deadline)]) as proc:
+            while time.time() < deadline:
+                try:
+                    os.stat(filename)
+                except FileNotFoundError as e:
+                    assert e.winerror == 2  # ERROR_FILE_NOT_FOUND
+            try:
+                proc.wait(1)
+            except subprocess.TimeoutExpired:
+                proc.terminate()
+
+
 @os_helper.skip_unless_symlink
 class NonLocalSymlinkTests(unittest.TestCase):