Skip to content

Commit

Permalink
Revert "Fix webdav tests (fsspec#107)"
Browse files Browse the repository at this point in the history
This reverts commit 62e1c3d.
  • Loading branch information
ap-- committed Jun 22, 2023
1 parent 62e1c3d commit 7dd1c00
Show file tree
Hide file tree
Showing 3 changed files with 4 additions and 134 deletions.
1 change: 0 additions & 1 deletion .flake8
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
8 changes: 4 additions & 4 deletions upath/tests/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,6 @@
from fsspec.registry import register_implementation

from .utils import posixify
from .utils import rmtree


class DummyTestFS(LocalFileSystem):
Expand Down Expand Up @@ -315,12 +314,13 @@ 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, dirs_exist_ok=True)
shutil.copytree(local_testdir, webdav_path)
yield webdav_url
finally:
rmtree(webdav_path)
os.mkdir(webdav_path, mode=0o700)
shutil.rmtree(webdav_path, ignore_errors=True)


@pytest.fixture(scope="session")
Expand Down
129 changes: 0 additions & 129 deletions upath/tests/utils.py
Original file line number Diff line number Diff line change
@@ -1,8 +1,4 @@
import os
import stat
import sys
import time
import warnings

import pytest

Expand All @@ -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 ==========================================

0 comments on commit 7dd1c00

Please sign in to comment.