Add test cases that delete a file during incremental checking

mypy/test/data.py

@@ -54,6 +54,7 @@ def parse_test_cases(
             output_files = []  # type: List[Tuple[str, str]] # path and contents for output files
             tcout = []  # type: List[str]  # Regular output errors
             tcout2 = {}  # type: Dict[int, List[str]]  # Output errors for incremental, runs 2+
+            deleted_paths = {}  # type: Dict[int, Set[str]]  # from run number of paths
             stale_modules = {}  # type: Dict[int, Set[str]]  # from run number to module names
             rechecked_modules = {}  # type: Dict[ int, Set[str]]  # from run number module names
             while i < len(p) and p[i].id != 'case':
@@ -99,6 +100,16 @@ def parse_test_cases(
                         rechecked_modules[passnum] = set()
                     else:
                         rechecked_modules[passnum] = {item.strip() for item in arg.split(',')}
+                elif p[i].id == 'delete':
+                    # File to delete during a multi-step test case
+                    arg = p[i].arg
+                    assert arg is not None
+                    m = re.match(r'(.*)\.([0-9]+)$', arg)
+                    assert m, 'Invalid delete section: {}'.format(arg)
+                    num = int(m.group(2))
+                    assert num >= 2, "Can't delete during step {}".format(num)
+                    full = join(base_path, m.group(1))
+                    deleted_paths.setdefault(num, set()).add(full)
                 elif p[i].id == 'out' or p[i].id == 'out1':
                     tcout = p[i].data
                     if native_sep and os.path.sep == '\\':
@@ -142,7 +153,7 @@ def parse_test_cases(
                 tc = DataDrivenTestCase(p[i0].arg, input, tcout, tcout2, path,
                                         p[i0].line, lastline, perform,
                                         files, output_files, stale_modules,
-                                        rechecked_modules, native_sep)
+                                        rechecked_modules, deleted_paths, native_sep)
                 out.append(tc)
         if not ok:
             raise ValueError(
@@ -180,6 +191,7 @@ def __init__(self,
                  output_files: List[Tuple[str, str]],
                  expected_stale_modules: Dict[int, Set[str]],
                  expected_rechecked_modules: Dict[int, Set[str]],
+                 deleted_paths: Dict[int, Set[str]],
                  native_sep: bool = False,
                  ) -> None:
         super().__init__(name)
@@ -194,24 +206,30 @@ def __init__(self,
         self.output_files = output_files
         self.expected_stale_modules = expected_stale_modules
         self.expected_rechecked_modules = expected_rechecked_modules
+        self.deleted_paths = deleted_paths
         self.native_sep = native_sep
 
     def set_up(self) -> None:
         super().set_up()
         encountered_files = set()
         self.clean_up = []
+        all_deleted = []  # type: List[str]
+        for paths in self.deleted_paths.values():
+            all_deleted += paths
         for path, content in self.files:
             dir = os.path.dirname(path)
             for d in self.add_dirs(dir):
                 self.clean_up.append((True, d))
             with open(path, 'w') as f:
                 f.write(content)
-            self.clean_up.append((False, path))
+            if path not in all_deleted:
+                # TODO: Don't assume that deleted files don't get reintroduced.
+                self.clean_up.append((False, path))
             encountered_files.add(path)
             if re.search(r'\.[2-9]$', path):
                 # Make sure new files introduced in the second and later runs are accounted for
                 renamed_path = path[:-2]
-                if renamed_path not in encountered_files:
+                if renamed_path not in encountered_files and renamed_path not in all_deleted:
                     encountered_files.add(renamed_path)
                     self.clean_up.append((False, renamed_path))
         for path, _ in self.output_files:

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, Callable, Any
 
 from mypy import defaults
 from mypy.myunit import AssertionFailure
@@ -283,3 +284,26 @@ def normalize_error_messages(messages: List[str]) -> List[str]:
     for m in messages:
         a.append(m.replace(os.sep, '/'))
     return a
+
+
+def retry_on_error(func: Callable[[], Any], max_wait: float = 1.0) -> None:
+    """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:
+            func()
+            return
+        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,13 +147,18 @@ 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))
+                # Delete files scheduled to be deleted in [delete <path>.num] sections.
+                for path in testcase.deleted_paths.get(incremental_step, set()):
+                    # 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)

test-data/unit/check-incremental.test

@@ -3,6 +3,9 @@
 -- Before the tests are run again, in step N any *.py.N files are copied to
 -- *.py.
 --
+-- You can add an empty section like `[delete mod.py.2]` to delete `mod.py`
+-- before the second run.
+--
 -- Errors expected in the first run should be in the `[out1]` section, and
 -- errors expected in the second run should be in the `[out2]` section, and so on.
 -- If a section is omitted, it is expected there are no errors on that run.
@@ -1947,6 +1950,34 @@ main:5: error: Revealed type is 'builtins.int'
 -- TODO: Add another test for metaclass in import cycle (reversed from the above test).
 --       This currently doesn't work.
 
+[case testDeleteFile]
+import n
+[file n.py]
+import m
+[file m.py]
+x = 1
+[delete m.py.2]
+[rechecked n]
+[stale]
+[out2]
+tmp/n.py:1: error: Cannot find module named 'm'
+tmp/n.py:1: note: (Perhaps setting MYPYPATH or using the "--ignore-missing-imports" flag would help)
+
+[case testDeleteFileWithinCycle]
+import a
+[file a.py]
+import b
+[file b.py]
+import c
+[file c.py]
+import a
+[file a.py.2]
+import c
+[delete b.py.2]
+[rechecked a, c]
+[stale a]
+[out2]
+
 [case testThreePassesBasic]
 import m
 [file m.py]