Add test cases that delete a file during incremental checking

mypy/test/helpers.py

@@ -1,8 +1,9 @@
-import sys
-import re
 import os
+import re
+import sys
+import time
 
-from typing import List, Dict, Tuple
+from typing import List, Dict, Tuple, TypeVar, Callable
 
 from mypy import defaults
 from mypy.myunit import AssertionFailure
@@ -283,3 +284,28 @@ def normalize_error_messages(messages: List[str]) -> List[str]:
     for m in messages:
         a.append(m.replace(os.sep, '/'))
     return a
+
+
+_T = TypeVar('_T')
+
+
+def retry_on_error(func: Callable[[], _T], max_wait: float = 1.0) -> _T:
+    """Retry callback with exponential backoff when it raises OSError.
+
+    If the function still generates an error after max_wait seconds, propagate
+    the exception.
+
+    This can be effective against random file system operation failures on
+    Windows.
+    """
+    t0 = time.time()
+    wait_time = 0.01
+    while True:
+        try:
+            return func()
+        except OSError:
+            wait_time = min(wait_time * 2, t0 + max_wait - time.time())
+            if wait_time <= 0.01:
+                # Done enough waiting, the error seems persistent.
+                raise
+            time.sleep(wait_time)

mypy/test/testcheck.py

@@ -17,7 +17,7 @@
 from mypy.test.data import parse_test_cases, DataDrivenTestCase, DataSuite
 from mypy.test.helpers import (
     assert_string_arrays_equal, normalize_error_messages,
-    testcase_pyversion, update_testcase_output,
+    retry_on_error, testcase_pyversion, update_testcase_output,
 )
 from mypy.errors import CompileError
 from mypy.options import Options
@@ -147,15 +147,17 @@ def run_case_once(self, testcase: DataDrivenTestCase, incremental_step: int = 0)
                         if file.endswith('.' + str(incremental_step)):
                             full = os.path.join(dn, file)
                             target = full[:-2]
-                            shutil.copy(full, target)
+                            # Use retries to work around potential flakiness on Windows (AppVeyor).
+                            retry_on_error(lambda: shutil.copy(full, target))
 
                             # In some systems, mtime has a resolution of 1 second which can cause
                             # annoying-to-debug issues when a file has the same size after a
                             # change. We manually set the mtime to circumvent this.
                             new_time = os.stat(target).st_mtime + 1
                             os.utime(target, times=(new_time, new_time))
                 for path in testcase.deleted_paths.get(incremental_step, set()):
-                    os.remove(path)
+                    # Use retries to work around potential flakiness on Windows (AppVeyor).
+                    retry_on_error(lambda: os.remove(path))
 
         # Parse options after moving files (in case mypy.ini is being moved).
         options = self.parse_options(original_program_text, testcase, incremental_step)