diff --git a/.flake8 b/.flake8 index 7974cc63..5203719f 100644 --- a/.flake8 +++ b/.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, diff --git a/upath/tests/conftest.py b/upath/tests/conftest.py index 8c0e9e5d..692b9e32 100644 --- a/upath/tests/conftest.py +++ b/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") diff --git a/upath/tests/utils.py b/upath/tests/utils.py index 2fe94668..62c9e0c9 100644 --- a/upath/tests/utils.py +++ b/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 ==========================================