tests: introduce dvc.testing

This is a very early version of test utils for dvc plugins. This module
consists of existing tests with minimal changes and is going to evolve
a lot as our plugins are getting filled in and pytest-test-utils matures.
This commit is contained in:
Ruslan Kuprieiev
2021-12-02 09:08:40 +02:00
parent 004e15c2c0
commit 09f48a5aaa
36 changed files with 1338 additions and 1009 deletions
+23 -110
View File
@@ -1,12 +1,9 @@
import subprocess
import pytest
from .azure import Azure, azure, azure_server # noqa: F401
from .azure import Azure, azure, azure_server, make_azure # noqa: F401
from .gdrive import ( # noqa: F401; noqa: F401
TEST_GDRIVE_REPO_BUCKET,
GDrive,
gdrive,
make_gdrive,
)
from .git_server import git_server, git_ssh # noqa: F401
from .gs import ( # noqa: F401; noqa: F401
@@ -14,13 +11,22 @@ from .gs import ( # noqa: F401; noqa: F401
TEST_GCP_CREDS_FILE,
TEST_GCP_REPO_BUCKET,
gs,
make_gs,
)
from .hdfs import HDFS, hadoop, hdfs, hdfs_server, real_hdfs # noqa: F401
from .http import HTTP, http, http_server # noqa: F401
from .hdfs import ( # noqa: F401
HDFS,
hadoop,
hdfs,
hdfs_server,
make_hdfs,
real_hdfs,
)
from .http import HTTP, http, http_server, make_http # noqa: F401
from .local import Local, local_cloud, local_remote, make_local # noqa: F401
from .oss import ( # noqa: F401
OSS,
TEST_OSS_REPO_BUCKET,
make_oss,
oss,
oss_server,
real_oss,
@@ -28,14 +34,21 @@ from .oss import ( # noqa: F401
from .s3 import ( # noqa: F401
S3,
TEST_AWS_REPO_BUCKET,
make_s3,
real_s3,
s3,
s3_fake_creds_file,
s3_server,
)
from .ssh import SSH, ssh, ssh_connection, ssh_server # noqa: F401; noqa: F401
from .webdav import Webdav, webdav, webdav_server # noqa: F401
from .webhdfs import WebHDFS, webhdfs # noqa: F401
from .ssh import ( # noqa: F401; noqa: F401
SSH,
make_ssh,
ssh,
ssh_connection,
ssh_server,
)
from .webdav import Webdav, make_webdav, webdav, webdav_server # noqa: F401
from .webhdfs import WebHDFS, make_webhdfs, webhdfs # noqa: F401
TEST_REMOTE = "upstream"
TEST_CONFIG = {
@@ -43,103 +56,3 @@ TEST_CONFIG = {
"core": {"remote": TEST_REMOTE},
"remote": {TEST_REMOTE: {"url": ""}},
}
@pytest.fixture(scope="session")
def docker():
import os
# See https://travis-ci.community/t/docker-linux-containers-on-windows/301
if os.environ.get("CI") and os.name == "nt":
pytest.skip("disabled for Windows on Github Actions")
try:
subprocess.check_output("docker ps", shell=True)
except (subprocess.CalledProcessError, OSError):
pytest.skip("no docker installed")
@pytest.fixture(scope="session")
def docker_compose(docker):
try:
subprocess.check_output("docker-compose version", shell=True)
except (subprocess.CalledProcessError, OSError):
pytest.skip("no docker-compose installed")
@pytest.fixture(scope="session")
def docker_compose_project_name():
return "pytest-dvc-test"
@pytest.fixture(scope="session")
def docker_services(
docker_compose_file, docker_compose_project_name, tmp_path_factory
):
# overriding `docker_services` fixture from `pytest_docker` plugin to
# only launch docker images once.
from filelock import FileLock
from pytest_docker.plugin import DockerComposeExecutor, Services
executor = DockerComposeExecutor(
docker_compose_file, docker_compose_project_name
)
# making sure we don't accidentally launch docker-compose in parallel,
# as it might result in network conflicts. Inspired by:
# https://github.com/pytest-dev/pytest-xdist#making-session-scoped-fixtures-execute-only-once
lockfile = tmp_path_factory.getbasetemp().parent / "docker-compose.lock"
with FileLock(str(lockfile)): # pylint:disable=abstract-class-instantiated
executor.execute("up --build -d")
return Services(executor)
@pytest.fixture
def make_cloud(request):
def _make_cloud(typ):
return request.getfixturevalue(f"make_{typ}")()
return _make_cloud
@pytest.fixture
def make_remote(tmp_dir, dvc, make_cloud):
def _make_remote(name, typ="local", **kwargs):
cloud = make_cloud(typ)
tmp_dir.add_remote(name=name, config=cloud.config, **kwargs)
return cloud
return _make_remote
@pytest.fixture
def remote(tmp_dir, dvc, request):
cloud = request.param
assert cloud
tmp_dir.add_remote(config=cloud.config)
yield cloud
@pytest.fixture
def workspace(tmp_dir, dvc, request):
from dvc.objects.db import ODBManager
cloud = request.param
assert cloud
tmp_dir.add_remote(name="workspace", config=cloud.config, default=False)
tmp_dir.add_remote(
name="cache", url="remote://workspace/cache", default=False
)
scheme = getattr(cloud, "scheme", "local")
if scheme != "http":
with dvc.config.edit() as conf:
conf["cache"][scheme] = "cache"
dvc.odb = ODBManager(dvc)
return cloud
+40 -14
View File
@@ -6,8 +6,8 @@ import uuid
import pytest
from funcy import cached_property
from .base import Base
from .path_info import CloudURLInfo
from dvc.testing.cloud import Cloud
from dvc.testing.path_info import CloudURLInfo
TEST_AZURE_CONTAINER = "tests"
TEST_AZURE_CONNECTION_STRING = (
@@ -19,11 +19,15 @@ TEST_AZURE_CONNECTION_STRING = (
)
class Azure(Base, CloudURLInfo):
class Azure(Cloud, CloudURLInfo):
IS_OBJECT_STORAGE = True
CONNECTION_STRING = None
def __init__(self, url, **kwargs):
super().__init__(url)
self.opts = kwargs
@cached_property
def service_client(self):
# pylint: disable=no-name-in-module
@@ -62,6 +66,19 @@ class Azure(Base, CloudURLInfo):
path = self.path.lstrip("/")
return f"{bucket}/{path}"
def is_file(self):
raise NotImplementedError
def is_dir(self):
raise NotImplementedError
def exists(self):
raise NotImplementedError
@property
def config(self):
return {"url": self.url, **self.opts}
@pytest.fixture(scope="session")
def azure_server(test_config, docker_compose, docker_services):
@@ -95,17 +112,26 @@ def azure_server(test_config, docker_compose, docker_services):
@pytest.fixture
def azure(azure_server):
from azure.core.exceptions import ResourceNotFoundError
def make_azure(azure_server):
def _make_azure():
from azure.core.exceptions import ResourceNotFoundError
url = f"azure://{TEST_AZURE_CONTAINER}/{uuid.uuid4()}"
ret = Azure(url)
ret.config = {"url": url, "connection_string": azure_server}
url = f"azure://{TEST_AZURE_CONTAINER}/{uuid.uuid4()}"
ret = Azure(url, connection_string=azure_server)
container = ret.service_client.get_container_client(TEST_AZURE_CONTAINER)
try: # verify that container exists
container.get_container_properties()
except ResourceNotFoundError:
container.create_container()
container = ret.service_client.get_container_client(
TEST_AZURE_CONTAINER
)
try: # verify that container exists
container.get_container_properties()
except ResourceNotFoundError:
container.create_container()
return ret
return ret
return _make_azure
@pytest.fixture
def azure(make_azure):
return make_azure()
-76
View File
@@ -1,76 +0,0 @@
import locale
import pathlib
from funcy import cached_property
from .path_info import URLInfo
class Base(URLInfo):
IS_OBJECT_STORAGE = False
def is_file(self):
raise NotImplementedError
def is_dir(self):
raise NotImplementedError
def exists(self):
raise NotImplementedError
def mkdir(self, mode=0o777, parents=False, exist_ok=False):
raise NotImplementedError
def write_text(self, contents, encoding=None, errors=None):
if not encoding:
encoding = locale.getpreferredencoding(False)
assert errors is None
self.write_bytes(contents.encode(encoding))
def write_bytes(self, contents):
raise NotImplementedError
def read_text(self, encoding=None, errors=None):
raise NotImplementedError
def read_bytes(self):
raise NotImplementedError
def _gen(self, struct, prefix=None):
for name, contents in struct.items():
path = (prefix or self) / name
if isinstance(contents, dict):
if not contents:
path.mkdir(parents=True, exist_ok=True)
else:
self._gen(contents, prefix=path)
else:
path.parent.mkdir(parents=True, exist_ok=True)
if isinstance(contents, bytes):
path.write_bytes(contents)
else:
path.write_text(contents, encoding="utf-8")
def gen(self, struct, text=""):
if isinstance(struct, (str, bytes, pathlib.PurePath)):
struct = {struct: text}
self._gen(struct)
return struct.keys()
def close(self):
pass
@staticmethod
def should_test():
return True
@staticmethod
def get_url():
raise NotImplementedError
@cached_property
def config(self):
return {"url": self.url}
+31 -14
View File
@@ -9,11 +9,10 @@ import pytest
from funcy import cached_property, retry
from dvc.fs.gdrive import GDriveFileSystem
from dvc.testing.cloud import Cloud
from dvc.testing.path_info import CloudURLInfo
from dvc.utils import tmp_fname
from .base import Base
from .path_info import CloudURLInfo
TEST_GDRIVE_REPO_BUCKET = "root"
@@ -62,12 +61,12 @@ def _gdrive_retry(func):
)(func)
class GDrive(Base, GDriveURLInfo):
class GDrive(Cloud, GDriveURLInfo):
@staticmethod
def should_test():
return bool(os.getenv(GDriveFileSystem.GDRIVE_CREDENTIALS_DATA))
@cached_property
@property
def config(self):
tmp_path = tmp_fname()
with open(tmp_path, "w", encoding="utf-8") as stream:
@@ -110,6 +109,15 @@ class GDrive(Base, GDriveURLInfo):
service_account=self.config["gdrive_use_service_account"],
)
def is_file(self):
raise NotImplementedError
def is_dir(self):
raise NotImplementedError
def exists(self):
raise NotImplementedError
@_gdrive_retry
def mkdir(self, mode=0o777, parents=False, exist_ok=False):
try:
@@ -135,17 +143,26 @@ class GDrive(Base, GDriveURLInfo):
@pytest.fixture
def gdrive(test_config, make_tmp_dir):
def make_gdrive(test_config, make_tmp_dir):
test_config.requires("gdrive")
if not GDrive.should_test():
pytest.skip("no gdrive")
# NOTE: temporary workaround
tmp_dir = make_tmp_dir("gdrive", dvc=True)
def _make_gdrive():
# NOTE: temporary workaround
tmp_dir = make_tmp_dir("gdrive", dvc=True)
ret = GDrive(GDrive.get_url())
fs = GDriveFileSystem(
gdrive_credentials_tmp_dir=tmp_dir.dvc.tmp_dir, **ret.config
)
fs.fs._gdrive_create_dir("root", fs.url)
yield ret
ret = GDrive(GDrive.get_url())
fs = GDriveFileSystem(
gdrive_credentials_tmp_dir=tmp_dir.dvc.tmp_dir, **ret.config
)
fs.fs._gdrive_create_dir("root", fs.url)
yield ret
return _make_gdrive
@pytest.fixture
def gdrive(make_gdrive):
return make_gdrive()
+10 -7
View File
@@ -3,13 +3,11 @@ import os
import uuid
import pytest
from funcy import cached_property
from dvc.testing.cloud import Cloud
from dvc.testing.path_info import CloudURLInfo
from dvc.utils import env2bool
from .base import Base
from .path_info import CloudURLInfo
TEST_GCP_REPO_BUCKET = os.environ.get("DVC_TEST_GCP_REPO_BUCKET", "dvc-test")
TEST_GCP_CREDS_FILE = os.path.abspath(
@@ -22,7 +20,7 @@ TEST_GCP_CREDS_FILE = os.path.abspath(
os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = TEST_GCP_CREDS_FILE
class GCP(Base, CloudURLInfo):
class GCP(Cloud, CloudURLInfo):
IS_OBJECT_STORAGE = True
@@ -37,7 +35,7 @@ class GCP(Base, CloudURLInfo):
return True
@cached_property
@property
def config(self):
return {"url": self.url, "credentialpath": TEST_GCP_CREDS_FILE}
@@ -95,8 +93,13 @@ class GCP(Base, CloudURLInfo):
@pytest.fixture
def gs(test_config):
def make_gs(test_config):
test_config.requires("gs")
if not GCP.should_test():
pytest.skip("no gs")
yield GCP(GCP.get_url())
@pytest.fixture
def gs(make_gs):
return make_gs()
+18 -6
View File
@@ -10,13 +10,13 @@ from urllib.parse import urlparse
import pytest
from .base import Base
from .path_info import URLInfo
from dvc.testing.cloud import Cloud
from dvc.testing.path_info import URLInfo
_hdfs_root = TemporaryDirectory()
class HDFS(Base, URLInfo): # pylint: disable=abstract-method
class HDFS(Cloud, URLInfo): # pylint: disable=abstract-method
@contextmanager
def _hdfs(self):
import pyarrow.fs
@@ -24,6 +24,10 @@ class HDFS(Base, URLInfo): # pylint: disable=abstract-method
conn = pyarrow.fs.HadoopFileSystem(self.host, self.port)
yield conn
@property
def config(self):
return {"url": self.url}
def is_file(self):
with self._hdfs() as _hdfs:
import pyarrow.fs
@@ -280,7 +284,7 @@ class FakeHadoopFileSystem:
@pytest.fixture
def hdfs(test_config, mocker):
def make_hdfs(test_config, mocker):
# Windows might not have Visual C++ Redistributable for Visual Studio
# 2015 installed, which will result in the following error:
# "The pyarrow installation is not built with support for
@@ -295,5 +299,13 @@ def hdfs(test_config, mocker):
mocker.patch("dvc.fs.hdfs.HDFSFileSystem._checksum", hadoop_fs_checksum)
url = f"hdfs://example.com:12345/{uuid.uuid4()}"
yield HDFS(url)
def _make_hdfs():
url = f"hdfs://example.com:12345/{uuid.uuid4()}"
return HDFS(url)
return _make_hdfs
@pytest.fixture
def hdfs(make_hdfs):
return make_hdfs()
+29 -5
View File
@@ -5,11 +5,11 @@ import uuid
import pytest
import requests
from .base import Base
from .path_info import HTTPURLInfo
from dvc.testing.cloud import Cloud
from dvc.testing.path_info import HTTPURLInfo
class HTTP(Base, HTTPURLInfo):
class HTTP(Cloud, HTTPURLInfo):
@staticmethod
def get_url(port): # pylint: disable=arguments-differ
dname = str(uuid.uuid4())
@@ -24,10 +24,26 @@ class HTTP(Base, HTTPURLInfo):
response = requests.post(self.url, data=contents)
assert response.status_code == 200
@property
def config(self):
return {"url": self.url}
@property
def fs_path(self):
return self.url
def exists(self):
raise NotImplementedError
def is_dir(self):
raise NotImplementedError
def is_file(self):
raise NotImplementedError
def read_bytes(self):
raise NotImplementedError
@pytest.fixture(scope="session")
def http_server(test_config, tmp_path_factory):
@@ -40,5 +56,13 @@ def http_server(test_config, tmp_path_factory):
@pytest.fixture
def http(http_server):
yield HTTP(HTTP.get_url(http_server.server_port))
def make_http(http_server):
def _make_http():
return HTTP(HTTP.get_url(http_server.server_port))
return _make_http
@pytest.fixture
def http(make_http):
return make_http()
+3 -7
View File
@@ -1,12 +1,11 @@
# pylint: disable=cyclic-import,disable=abstract-method
import pytest
from dvc.testing.cloud import Cloud
from tests.basic_env import TestDvc
from .base import Base
class Local(Base):
class Local(Cloud):
@staticmethod
def get_storagepath():
return TestDvc.mkdtemp()
@@ -19,10 +18,7 @@ class Local(Base):
@pytest.fixture
def make_local(make_tmp_dir):
def _make_local():
ret = make_tmp_dir("local-cloud")
ret.url = str(ret)
ret.config = {"url": ret.url}
return ret
return make_tmp_dir("local-cloud")
return _make_local
+57 -26
View File
@@ -4,18 +4,17 @@ import uuid
import pytest
from dvc.testing.cloud import Cloud
from dvc.testing.path_info import CloudURLInfo
from dvc.utils import env2bool
from .base import Base
from .path_info import CloudURLInfo
TEST_OSS_REPO_BUCKET = "dvc-test-github"
EMULATOR_OSS_ENDPOINT = "127.0.0.1:{port}"
EMULATOR_OSS_ACCESS_KEY_ID = "AccessKeyID"
EMULATOR_OSS_ACCESS_KEY_SECRET = "AccessKeySecret"
class OSS(Base, CloudURLInfo):
class OSS(Cloud, CloudURLInfo):
IS_OBJECT_STORAGE = True
@@ -36,6 +35,33 @@ class OSS(Base, CloudURLInfo):
return False
@property
def config(self):
return {
"url": self.url,
"oss_key_id": os.environ.get("OSS_ACCESS_KEY_ID"),
"oss_key_secret": os.environ.get("OSS_ACCESS_KEY_SECRET"),
"oss_endpoint": os.environ.get("OSS_ENDPOINT"),
}
def is_file(self):
raise NotImplementedError
def is_dir(self):
raise NotImplementedError
def exists(self):
raise NotImplementedError
def mkdir(self, mode=0o777, parents=False, exist_ok=False):
raise NotImplementedError
def write_bytes(self, contents):
raise NotImplementedError
def read_bytes(self):
raise NotImplementedError
@pytest.fixture(scope="session")
def oss_server(test_config, docker_compose, docker_services):
@@ -63,24 +89,36 @@ def oss_server(test_config, docker_compose, docker_services):
@pytest.fixture
def oss(real_oss):
import oss2
def make_oss(real_oss):
def _make_oss():
import oss2
ret = real_oss
ret = real_oss
auth = oss2.Auth(ret.config["oss_key_id"], ret.config["oss_key_secret"])
bucket = oss2.Bucket(
auth, ret.config["oss_endpoint"], TEST_OSS_REPO_BUCKET
)
try:
bucket.get_bucket_info()
except oss2.exceptions.NoSuchBucket:
bucket.create_bucket(
oss2.BUCKET_ACL_PUBLIC_READ,
oss2.models.BucketCreateConfig(oss2.BUCKET_STORAGE_CLASS_STANDARD),
auth = oss2.Auth(
ret.config["oss_key_id"], ret.config["oss_key_secret"]
)
bucket = oss2.Bucket(
auth, ret.config["oss_endpoint"], TEST_OSS_REPO_BUCKET
)
try:
bucket.get_bucket_info()
except oss2.exceptions.NoSuchBucket:
bucket.create_bucket(
oss2.BUCKET_ACL_PUBLIC_READ,
oss2.models.BucketCreateConfig(
oss2.BUCKET_STORAGE_CLASS_STANDARD
),
)
return ret
return ret
return _make_oss
@pytest.fixture
def oss(make_oss):
return make_oss()
@pytest.fixture
@@ -90,11 +128,4 @@ def real_oss(test_config):
pytest.skip("no real OSS")
url = OSS.get_url()
ret = OSS(url)
ret.config = {
"url": url,
"oss_key_id": os.environ.get("OSS_ACCESS_KEY_ID"),
"oss_key_secret": os.environ.get("OSS_ACCESS_KEY_SECRET"),
"oss_endpoint": os.environ.get("OSS_ENDPOINT"),
}
return ret
return OSS(url)
-350
View File
@@ -1,350 +0,0 @@
# pylint: disable=protected-access
import os
import pathlib
import posixpath
from typing import Callable
from urllib.parse import urlparse
from funcy import cached_property
from dvc.utils import relpath
class _BasePath:
def overlaps(self, other):
if isinstance(other, (str, bytes)):
other = self.__class__(other)
elif self.__class__ != other.__class__:
return False
return self.isin_or_eq(other) or other.isin(self)
def isin_or_eq(self, other):
return self == other or self.isin(other) # pylint: disable=no-member
class PathInfo(pathlib.PurePath, _BasePath):
# Use __slots__ in PathInfo objects following PurePath implementation.
# This makes objects smaller and speeds up attribute access.
# We don't add any fields so it's empty.
__slots__ = ()
scheme = "local"
def __new__(cls, *args):
# Construct a proper subclass depending on current os
if cls is PathInfo:
cls = ( # pylint: disable=self-cls-assignment
WindowsPathInfo if os.name == "nt" else PosixPathInfo
)
return cls._from_parts(args)
def as_posix(self):
f = self._flavour # pylint: disable=no-member
# Unlike original implementation [1] that uses `str()` we actually need
# to use `fspath`, because we've overridden `__str__` method to return
# relative paths, which will break original `as_posix`.
#
# [1] https://github.com/python/cpython/blob/v3.7.0/Lib/pathlib.py#L692
return self.fspath.replace(f.sep, "/")
def __str__(self):
path = self.__fspath__()
return relpath(path)
def __repr__(self):
return f"{type(self).__name__}: '{self}'"
# This permits passing it to file utils directly in Python 3.6+
def __fspath__(self):
return pathlib.PurePath.__str__(self)
@property
def fspath(self):
return self.__fspath__()
url = fspath
path = fspath
def relpath(self, other):
return self.__class__(relpath(self, other))
def isin(self, other):
if isinstance(other, (str, bytes)):
other = self.__class__(other)
elif self.__class__ != other.__class__:
return False
# Use cached casefolded parts to compare paths
n = len(other._cparts)
return len(self._cparts) > n and self._cparts[:n] == other._cparts
def relative_to(self, other): # pylint: disable=arguments-differ
# pathlib relative_to raises exception when one path is not a direct
# descendant of the other when os.path.relpath would return abspath.
# For DVC PathInfo we only need the relpath behavior.
# See: https://bugs.python.org/issue40358
try:
path = super().relative_to(other)
except ValueError:
path = relpath(self, other)
return self.__class__(path)
class WindowsPathInfo(PathInfo, pathlib.PureWindowsPath):
pass
class PosixPathInfo(PathInfo, pathlib.PurePosixPath):
pass
class _URLPathInfo(PosixPathInfo):
def __str__(self):
return self.__fspath__()
__unicode__ = __str__
class _URLPathParents:
def __init__(self, src):
self.src = src
self._parents = self.src._path.parents
def __len__(self):
return len(self._parents)
def __getitem__(self, idx):
return self.src.replace(path=self._parents[idx])
def __repr__(self):
return f"<{self.src}.parents>"
class URLInfo(_BasePath):
DEFAULT_PORTS = {"http": 80, "https": 443, "ssh": 22, "hdfs": 0}
def __init__(self, url):
p = urlparse(url)
assert not p.query and not p.params and not p.fragment
assert p.password is None
self._fill_parts(p.scheme, p.hostname, p.username, p.port, p.path)
@classmethod
def from_parts(
cls, scheme=None, host=None, user=None, port=None, path="", netloc=None
):
assert bool(host) ^ bool(netloc)
if netloc is not None:
return cls(f"{scheme}://{netloc}{path}")
obj = cls.__new__(cls)
obj._fill_parts(scheme, host, user, port, path)
return obj
def _fill_parts(self, scheme, host, user, port, path):
assert scheme != "remote"
assert isinstance(path, (str, bytes, _URLPathInfo))
self.scheme, self.host, self.user = scheme, host, user
self.port = int(port) if port else self.DEFAULT_PORTS.get(self.scheme)
if isinstance(path, _URLPathInfo):
self._spath = str(path)
self._path = path
else:
if path and path[0] != "/":
path = "/" + path
self._spath = path
@property
def _base_parts(self):
return (self.scheme, self.host, self.user, self.port)
@property
def parts(self):
return self._base_parts + self._path.parts
def replace(self, path=None):
return self.from_parts(*self._base_parts, path=path)
@cached_property
def url(self):
return f"{self.scheme}://{self.netloc}{self._spath}"
def __str__(self):
return self.url
def __repr__(self):
return f"{type(self).__name__}: '{self}'"
def __eq__(self, other):
if isinstance(other, (str, bytes)):
other = self.__class__(other)
return (
self.__class__ == other.__class__
and self._base_parts == other._base_parts
and self._path == other._path
)
def __hash__(self):
return hash(self.parts)
def __div__(self, other):
return self.replace(path=posixpath.join(self._spath, other))
def joinpath(self, *args):
return self.replace(path=posixpath.join(self._spath, *args))
__truediv__ = __div__
@property
def path(self):
return self._spath
@cached_property
def _path(self): # false-positive, pylint: disable=method-hidden
return _URLPathInfo(self._spath)
@property
def name(self):
return self._path.name
@cached_property
def netloc(self):
netloc = self.host
if self.user:
netloc = self.user + "@" + netloc
if self.port and int(self.port) != self.DEFAULT_PORTS.get(self.scheme):
netloc += ":" + str(self.port)
return netloc
@property
def bucket(self):
return self.netloc
@property
def parent(self):
return self.replace(path=self._path.parent)
@property
def parents(self):
return _URLPathParents(self)
def relative_to(self, other):
if isinstance(other, (str, bytes)):
other = self.__class__(other)
if self.__class__ != other.__class__:
msg = f"'{self}' has incompatible class with '{other}'"
raise ValueError(msg)
if self._base_parts != other._base_parts:
msg = f"'{self}' does not start with '{other}'"
raise ValueError(msg)
return self._path.relative_to(other._path)
def isin(self, other):
if isinstance(other, (str, bytes)):
other = self.__class__(other)
elif self.__class__ != other.__class__:
return False
return self._base_parts == other._base_parts and self._path.isin(
other._path
)
class CloudURLInfo(URLInfo):
@property
def path(self):
return self._spath.lstrip("/")
class HTTPURLInfo(URLInfo):
__hash__: Callable[["HTTPURLInfo"], int] = URLInfo.__hash__
def __init__(self, url):
p = urlparse(url)
stripped = p._replace(params=None, query=None, fragment=None)
super().__init__(stripped.geturl())
self.params = p.params
self.query = p.query
self.fragment = p.fragment
def replace(self, path=None):
return self.from_parts(
*self._base_parts,
params=self.params,
query=self.query,
fragment=self.fragment,
path=path,
)
@classmethod
def from_parts(
cls,
scheme=None,
host=None,
user=None,
port=None,
path="",
netloc=None,
params=None,
query=None,
fragment=None,
): # pylint: disable=arguments-differ
assert bool(host) ^ bool(netloc)
if netloc is not None:
return cls(
"{}://{}{}{}{}{}".format(
scheme,
netloc,
path,
(";" + params) if params else "",
("?" + query) if query else "",
("#" + fragment) if fragment else "",
)
)
obj = cls.__new__(cls)
obj._fill_parts(scheme, host, user, port, path)
obj.params = params
obj.query = query
obj.fragment = fragment
return obj
@property
def _extra_parts(self):
return (self.params, self.query, self.fragment)
@property
def parts(self):
return self._base_parts + self._path.parts + self._extra_parts
@cached_property
def url(self):
return "{}://{}{}{}{}{}".format(
self.scheme,
self.netloc,
self._spath,
(";" + self.params) if self.params else "",
("?" + self.query) if self.query else "",
("#" + self.fragment) if self.fragment else "",
)
def __eq__(self, other):
if isinstance(other, (str, bytes)):
other = self.__class__(other)
return (
self.__class__ == other.__class__
and self._base_parts == other._base_parts
and self._path == other._path
and self._extra_parts == other._extra_parts
)
class WebDAVURLInfo(URLInfo):
@cached_property
def url(self):
return "{}://{}{}".format(
self.scheme.replace("webdav", "http"), self.netloc, self._spath
)
+17 -9
View File
@@ -5,21 +5,20 @@ import uuid
import pytest
from funcy import cached_property
from dvc.testing.cloud import Cloud
from dvc.testing.path_info import CloudURLInfo
from dvc.utils import env2bool
from .base import Base
from .path_info import CloudURLInfo
TEST_AWS_REPO_BUCKET = os.environ.get("DVC_TEST_AWS_REPO_BUCKET", "dvc-temp")
TEST_AWS_ENDPOINT_URL = "http://127.0.0.1:{port}/"
class S3(Base, CloudURLInfo):
class S3(Cloud, CloudURLInfo):
IS_OBJECT_STORAGE = True
TEST_AWS_ENDPOINT_URL = None
@cached_property
@property
def config(self):
return {"url": self.url, "endpointurl": self.TEST_AWS_ENDPOINT_URL}
@@ -151,11 +150,20 @@ def s3_server(test_config, docker_compose, docker_services):
@pytest.fixture
def s3(test_config, s3_server, s3_fake_creds_file):
def make_s3(test_config, s3_server, s3_fake_creds_file):
test_config.requires("s3")
workspace = S3(S3.get_url())
workspace._s3.create_bucket(Bucket=TEST_AWS_REPO_BUCKET)
yield workspace
def _make_s3():
cloud = S3(S3.get_url())
cloud._s3.create_bucket(Bucket=TEST_AWS_REPO_BUCKET)
return cloud
return _make_s3
@pytest.fixture
def s3(make_s3):
return make_s3()
@pytest.fixture
+19 -11
View File
@@ -5,8 +5,8 @@ import uuid
import pytest
from funcy import cached_property
from .base import Base
from .path_info import URLInfo
from dvc.testing.cloud import Cloud
from dvc.testing.path_info import URLInfo
TEST_SSH_USER = "user"
TEST_SSH_KEY_PATH = os.path.join(
@@ -14,12 +14,12 @@ TEST_SSH_KEY_PATH = os.path.join(
)
class SSH(Base, URLInfo):
class SSH(Cloud, URLInfo):
@staticmethod
def get_url(host, port): # pylint: disable=arguments-differ
return f"ssh://{host}:{port}/tmp/data/{uuid.uuid4()}"
@cached_property
@property
def config(self):
return {
"url": self.url,
@@ -116,12 +116,20 @@ def ssh_connection(ssh_server):
@pytest.fixture
def ssh(ssh_server, monkeypatch):
from dvc.fs.ssh import SSHFileSystem
def make_ssh(ssh_server, monkeypatch):
def _make_ssh():
from dvc.fs.ssh import SSHFileSystem
# NOTE: see http://github.com/iterative/dvc/pull/3501
monkeypatch.setattr(SSHFileSystem, "CAN_TRAVERSE", False)
# NOTE: see http://github.com/iterative/dvc/pull/3501
monkeypatch.setattr(SSHFileSystem, "CAN_TRAVERSE", False)
url = SSH(SSH.get_url(**ssh_server))
url.mkdir(exist_ok=True, parents=True)
return url
url = SSH(SSH.get_url(**ssh_server))
url.mkdir(exist_ok=True, parents=True)
return url
return _make_ssh
@pytest.fixture
def ssh(make_ssh):
return make_ssh()
+32 -10
View File
@@ -6,19 +6,24 @@ import pytest
from funcy import cached_property, first
from wsgidav.wsgidav_app import WsgiDAVApp
from dvc.testing.cloud import Cloud
from dvc.testing.path_info import WebDAVURLInfo
from tests.utils.httpd import run_server_on_thread
from .base import Base
from .path_info import WebDAVURLInfo
AUTH = {"user1": {"password": "password1"}}
class Webdav(Base, WebDAVURLInfo):
class Webdav(Cloud, WebDAVURLInfo):
@staticmethod
def get_url(port): # pylint: disable=arguments-differ
return f"webdav://localhost:{port}"
@property
def config(self):
user, secrets = first(AUTH.items())
url = f"webdav://{self.netloc}{self._spath}"
return {"url": url, "user": user, **secrets}
@cached_property
def client(self):
from webdav4.client import Client
@@ -45,6 +50,18 @@ class Webdav(Base, WebDAVURLInfo):
def fs_path(self):
return self.path.lstrip("/")
def exists(self):
raise NotImplementedError
def is_dir(self):
raise NotImplementedError
def is_file(self):
raise NotImplementedError
def read_bytes(self):
raise NotImplementedError
@pytest.fixture
def webdav_server(test_config, tmp_path_factory):
@@ -68,9 +85,14 @@ def webdav_server(test_config, tmp_path_factory):
@pytest.fixture
def webdav(webdav_server):
url = Webdav.get_url(webdav_server.server_port)
ret = Webdav(url)
user, secrets = first(AUTH.items())
ret.config = {"url": url, "user": user, **secrets}
yield ret
def make_webdav(webdav_server):
def _make_webdav():
url = Webdav.get_url(webdav_server.server_port)
return Webdav(url)
return _make_webdav
@pytest.fixture
def webdav(make_webdav):
return make_webdav()
+19 -7
View File
@@ -4,11 +4,11 @@ from contextlib import contextmanager
import pytest
from .base import Base
from .path_info import CloudURLInfo
from dvc.testing.cloud import Cloud
from dvc.testing.path_info import CloudURLInfo
class WebHDFS(Base, CloudURLInfo): # pylint: disable=abstract-method
class WebHDFS(Cloud, CloudURLInfo): # pylint: disable=abstract-method
@contextmanager
def _webhdfs(self):
from hdfs import InsecureClient
@@ -54,13 +54,25 @@ class WebHDFS(Base, CloudURLInfo): # pylint: disable=abstract-method
assert errors is None
return self.read_bytes().decode(encoding)
@property
def config(self):
return {"url": self.url}
@property
def fs_path(self):
return "/" + self.path.lstrip("/")
@pytest.fixture
def webhdfs(hdfs_server):
port = hdfs_server["webhdfs"]
url = f"webhdfs://127.0.0.1:{port}/{uuid.uuid4()}"
yield WebHDFS(url)
def make_webhdfs(hdfs_server):
def _make_webhdfs():
port = hdfs_server["webhdfs"]
url = f"webhdfs://127.0.0.1:{port}/{uuid.uuid4()}"
yield WebHDFS(url)
return _make_webhdfs
@pytest.fixture
def webhdfs(make_webhdfs):
return make_webhdfs()