Fix webdav tests

.flake8

@@ -17,6 +17,7 @@ count = true
 per-file-ignores =
     upath/__init__.py: F401
     upath/tests/test_pathlib.py: E127,E128,E201,E202,E203,E225,E255,E302,E303,E306,E401,E402,E501,E731
+    upath/tests/utils.py: C901
 exclude =
     .noxfile,
     .nox,

upath/tests/conftest.py

@@ -15,6 +15,7 @@
 from fsspec.registry import register_implementation
 
 from .utils import posixify
+from .utils import rmtree
 
 
 class DummyTestFS(LocalFileSystem):
@@ -314,13 +315,12 @@ def webdav_server(tmp_path_factory):
 @pytest.fixture
 def webdav_fixture(local_testdir, webdav_server):
     webdav_path, webdav_url = webdav_server
-    if os.path.isdir(webdav_path):
-        shutil.rmtree(webdav_path, ignore_errors=True)
     try:
-        shutil.copytree(local_testdir, webdav_path)
+        shutil.copytree(local_testdir, webdav_path, dirs_exist_ok=True)
         yield webdav_url
     finally:
-        shutil.rmtree(webdav_path, ignore_errors=True)
+        rmtree(webdav_path)
+        os.mkdir(webdav_path, mode=0o700)
 
 
 @pytest.fixture(scope="session")

upath/tests/utils.py

@@ -1,4 +1,8 @@
+import os
+import stat
 import sys
+import time
+import warnings
 
 import pytest
 
@@ -17,3 +21,128 @@ def only_on_windows(func):
 
 def posixify(path):
     return str(path).replace("\\", "/")
+
+
+# === vendored from test.os_support ===========================================
+# https://github.com/python/cpython/blob/7f97c8e36786/Lib/test/support/os_helper.py#LL327C1-L445C1
+# fmt: off
+if sys.platform.startswith("win"):
+    def _waitfor(func, pathname, waitall=False):
+        # Perform the operation
+        func(pathname)
+        # Now setup the wait loop
+        if waitall:
+            dirname = pathname
+        else:
+            dirname, name = os.path.split(pathname)
+            dirname = dirname or '.'
+        # Check for `pathname` to be removed from the filesystem.
+        # The exponential backoff of the timeout amounts to a total
+        # of ~1 second after which the deletion is probably an error
+        # anyway.
+        # Testing on an [email protected] shows that usually only 1 iteration is
+        # required when contention occurs.
+        timeout = 0.001
+        while timeout < 1.0:
+            # Note we are only testing for the existence of the file(s) in
+            # the contents of the directory regardless of any security or
+            # access rights.  If we have made it this far, we have sufficient
+            # permissions to do that much using Python's equivalent of the
+            # Windows API FindFirstFile.
+            # Other Windows APIs can fail or give incorrect results when
+            # dealing with files that are pending deletion.
+            L = os.listdir(dirname)
+            if not (L if waitall else name in L):
+                return
+            # Increase the timeout and try again
+            time.sleep(timeout)
+            timeout *= 2
+        warnings.warn('tests may fail, delete still pending for ' + pathname,
+                      RuntimeWarning, stacklevel=4)
+
+    def _unlink(filename):
+        _waitfor(os.unlink, filename)
+
+    def _rmdir(dirname):
+        _waitfor(os.rmdir, dirname)
+
+    def _rmtree(path):
+        from test.support import _force_run
+
+        def _rmtree_inner(path):
+            for name in _force_run(path, os.listdir, path):
+                fullname = os.path.join(path, name)
+                try:
+                    mode = os.lstat(fullname).st_mode
+                except OSError as exc:
+                    print("support.rmtree(): os.lstat(%r) failed with %s"
+                          % (fullname, exc),
+                          file=sys.__stderr__)
+                    mode = 0
+                if stat.S_ISDIR(mode):
+                    _waitfor(_rmtree_inner, fullname, waitall=True)
+                    _force_run(fullname, os.rmdir, fullname)
+                else:
+                    _force_run(fullname, os.unlink, fullname)
+        _waitfor(_rmtree_inner, path, waitall=True)
+        _waitfor(lambda p: _force_run(p, os.rmdir, p), path)
+
+    def _longpath(path):
+        try:
+            import ctypes
+        except ImportError:
+            # No ctypes means we can't expands paths.
+            pass
+        else:
+            buffer = ctypes.create_unicode_buffer(len(path) * 2)
+            length = ctypes.windll.kernel32.GetLongPathNameW(path, buffer,
+                                                             len(buffer))
+            if length:
+                return buffer[:length]
+        return path
+else:
+    _unlink = os.unlink
+    _rmdir = os.rmdir
+
+    def _rmtree(path):
+        import shutil
+        try:
+            shutil.rmtree(path)
+            return
+        except OSError:
+            pass
+
+        def _rmtree_inner(path):
+            from test.support import _force_run
+            for name in _force_run(path, os.listdir, path):
+                fullname = os.path.join(path, name)
+                try:
+                    mode = os.lstat(fullname).st_mode
+                except OSError:
+                    mode = 0
+                if stat.S_ISDIR(mode):
+                    _rmtree_inner(fullname)
+                    _force_run(path, os.rmdir, fullname)
+                else:
+                    _force_run(path, os.unlink, fullname)
+        _rmtree_inner(path)
+        os.rmdir(path)
+
+    def _longpath(path):
+        return path
+
+
+def rmdir(dirname):
+    try:
+        _rmdir(dirname)
+    except FileNotFoundError:
+        pass
+
+
+def rmtree(path):
+    try:
+        _rmtree(path)
+    except FileNotFoundError:
+        pass
+# fmt: on
+# === /vendored from test.os_support ==========================================