Simplify wait_for; test util.replace directly

Author: pkch

mypy/test/testutil.py

@@ -1,41 +1,78 @@
-import sys
-from typing import Type, Callable, List
+from typing import Iterator
 import time
-try:
-    import collections.abc as collections_abc
-except ImportError:
-    import collections as collections_abc  # type: ignore # PY32 and earlier
+import os
+import sys
+import tempfile
+from contextlib import contextmanager
+from threading import Thread
 from unittest import TestCase, main, skipUnless
 from mypy import util
 
 
-def create_bad_function(lag: float, exc: BaseException) -> Callable[[], None]:
-    start_time = time.perf_counter()
+WIN32 = sys.platform.startswith("win")
+
+
+@contextmanager
+def lock_file(filename: str, duration: float) -> Iterator[Thread]:
+    '''
+    Opens filename (which must exist) for reading
+    After duration sec, releases the handle
+    '''
+    def _lock_file() -> None:
+        with open(filename):
+            time.sleep(duration)
+    t = Thread(target=_lock_file, daemon=True)
+    t.start()
+    yield t
+    t.join()
+
+
+@skipUnless(WIN32, "only relevant for Windows")
+class ReliableReplace(TestCase):
+    tmpdir = tempfile.TemporaryDirectory(prefix='mypy-test-',
+                                         dir=os.path.abspath('tmp-test-dirs'))
+    src = os.path.join(tmpdir.name, 'tmp1')
+    dest = os.path.join(tmpdir.name, 'tmp2')
+
+    @classmethod
+    def tearDownClass(cls) -> None:
+        cls.tmpdir.cleanup()
+
+    def setUp(self) -> None:
+        # create two temporary files
+        for fname in (self.src, self.dest):
+            with open(fname, 'w') as f:
+                f.write(fname)
+
+    def replace_ok(self) -> None:
+        util._replace(self.src, self.dest, timeout=0.25)
+        self.assertEqual(open(self.dest).read(), self.src, 'replace failed')
 
-    def f() -> None:
-        if time.perf_counter() - start_time < lag:
-            raise exc
-        else:
-            return
-    return f
+    def test_normal(self) -> None:
+        self.replace_ok()
 
+    def test_problem_exists(self) -> None:
+        with lock_file(self.src, 0.1):
+            with self.assertRaises(PermissionError):
+                os.replace(self.src, self.dest)
 
-def create_funcs() -> List[Callable[[], None]]:
+    def test_short_lock_src(self) -> None:
+        with lock_file(self.src, 0.1):
+            self.replace_ok()
 
-    def linux_function() -> None: pass
-    windows_function1 = create_bad_function(0.1, PermissionError())
-    windows_function2 = create_bad_function(0.2, FileExistsError())
-    return [windows_function1, windows_function2, linux_function]
+    def test_short_lock_dest(self) -> None:
+        with lock_file(self.dest, 0.1):
+            self.replace_ok()
 
+    def test_long_lock_src(self) -> None:
+        with lock_file(self.src, 0.4):
+            with self.assertRaises(PermissionError):
+                self.replace_ok()
 
-class WaitRetryTests(TestCase):
-    def test_waitfor(self) -> None:
-        with self.assertRaises(OSError):
-            util.wait_for(create_funcs(), (PermissionError, FileExistsError), 0.1)
-        util.wait_for(create_funcs(), (PermissionError, FileExistsError), 1)
-        util.wait_for(create_funcs(), (OSError,), 1)
-        with self.assertRaises(FileExistsError):
-            util.wait_for(create_funcs(), (PermissionError,), 0.4)
+    def test_long_lock_dest(self) -> None:
+        with lock_file(self.dest, 0.4):
+            with self.assertRaises(PermissionError):
+                self.replace_ok()
 
 
 if __name__ == '__main__':

mypy/util.py

@@ -6,7 +6,6 @@
 import sys
 import math
 import time
-from functools import partial
 from xml.sax.saxutils import escape
 from typing import (
     TypeVar, List, Tuple, Optional, Sequence, Dict, Callable, Iterable, Type, Union, AnyStr, cast
@@ -143,49 +142,31 @@ def id(self, o: object) -> int:
         return self.id_map[o]
 
 
-# default timeout is short in case this functions is called individually for many files
-# batch processing results in better performance
-def wait_for(funcs: Iterable[Callable[[], None]],
-             exc: Iterable[Type[BaseException]] = (),
-             timeout: float = 0.1,
-             msg: str = 'file operations') -> None:
-    '''
-    Execute functions in funcs (without arguments)
-    Wait and retry all functions that raised exception that matches exc
-    Increase wait time exponentially, give up after a total wait of timeout seconds
-    Reraises the latest exception seen if timeout exceeded
-    '''
-    exc = tuple(exc)
-    last_exc = None
-    pending = set(funcs)
-    n_iter = max(1, math.ceil(math.log2(timeout / 0.001)))
-    for i in range(n_iter):
-        if not pending:
-            return
-        # last wait is ~ timeout/2, so that total wait ~ timeout
-        wait = timeout / 2 ** (n_iter - i)
-        failed = set()
-        for func in pending:
-            try:
-                func()
-            except exc as e:
-                last_exc = e
-                failed.add(func)
-        pending = failed
-        time.sleep(wait)
-    sys.stderr.write('timed out waiting for {}'.format(msg))
-    raise last_exc
-
-
 if sys.version_info >= (3, 6):
     PathType = Union[AnyStr, os.PathLike]
 else:
     PathType = AnyStr
 
 
-def _replace(src: PathType, dest: PathType) -> None:
-    repl = cast(Callable[[], None], partial(os.replace, src, dest))
-    wait_for([repl], (OSError,), 0.1)
+def _replace(src: PathType, dest: PathType, timeout: float = 10) -> None:
+    '''
+    Replace src with dest using os.replace(src, dest)
+    Wait and retry if OSError exception is raised
+    Increase wait time exponentially until total wait of timeout sec
+    On timeout, give up and reraise the last exception seen
+    '''
+    n_iter = max(1, math.ceil(math.log2(timeout / 0.001)))
+    for i in range(n_iter):
+        # last wait is ~ timeout/2, so that total wait ~ timeout
+        wait = timeout / 2 ** (n_iter - i - 1)
+        try:
+            os.replace(src, dest)
+        except PermissionError:
+            if i == n_iter - 1:
+                raise
+        else:
+            return
+        time.sleep(wait)
 
 
 if sys.platform.startswith("win"):