Fix webdav tests (again) (#109)

* Revert "Fix webdav tests (#107)"

This reverts commit 62e1c3d.

* upath.tests: switch root folder for each test for webdav app

* tests: fix webdav tests windows

.flake8

@@ -17,7 +17,6 @@ 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,7 +15,6 @@
 from fsspec.registry import register_implementation
 
 from .utils import posixify
-from .utils import rmtree
 
 
 class DummyTestFS(LocalFileSystem):
@@ -306,21 +305,22 @@ def webdav_server(tmp_path_factory):
     thread.start()
 
     try:
-        yield tempdir, f"webdav+http://{host}:{port}"
+        yield f"webdav+http://{host}:{port}", app
     finally:
         srvr.stop()
-        thread.join()
 
 
 @pytest.fixture
 def webdav_fixture(local_testdir, webdav_server):
-    webdav_path, webdav_url = webdav_server
+    webdav_url, app = webdav_server
+    # switch to new test directory
+    fs_provider = app.provider_map["/"]
+    fs_provider.root_folder_path = os.path.abspath(local_testdir)
     try:
-        shutil.copytree(local_testdir, webdav_path, dirs_exist_ok=True)
         yield webdav_url
     finally:
-        rmtree(webdav_path)
-        os.mkdir(webdav_path, mode=0o700)
+        # clear locks if any are held
+        fs_provider.lock_manager.storage.clear()
 
 
 @pytest.fixture(scope="session")

upath/tests/utils.py

@@ -1,8 +1,4 @@
-import os
-import stat
 import sys
-import time
-import warnings
 
 import pytest
 
@@ -21,128 +17,3 @@ 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 i7@4.3GHz 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 ==========================================