Stone_Red/dvc
1
0
Fork 0
mirror of https://github.com/Stone-Red-Code/dvc.git synced 2026-11-10 23:10:42 +01:00
Code Issues Packages Projects Releases Wiki Activity

remote: continue remote/cache/tree refactoring (#4041)

Browse Source 
* remote: move constants into trees

* remote: replace Remote() Cache() helper functions

* output: update for remote changes

* cache/data_cloud: update for remote changes

* dependency: update for remote changes

* stage: update for remote changes

* remote: move remaining non-index methods into tree

* tests: update remote unit tests

* remote: LocalCache needs to lock remote index, not cache index

* fix bugs from moved PARAM_CHECKSUM/PARAM_RELPATH

* tests: repo tree open needs state context

* remote: make Remote.gc a class method

* operates on obj.tree, so that GC can be done with both remote and
  cache instances

* bug fixes

* tests: update test_data_cloud

* bug fixes

* tests: update func tests
This commit is contained in:
Peter Rowlands (변기호)
2020-06-15 16:22:55 +03:00
committed by GitHub
parent a59f90fac9
commit a0c229cf9b
54 changed files with 925 additions and 900 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dvc/cache.py
+8 -4
View File
@@ -28,13 +28,15 @@ def _make_remote_property(name):
"""
def getter(self):
from dvc.remote import Cache as CloudCache
from dvc.remote import get_cloud_tree
from dvc.remote.base import CloudCache
remote = self.config.get(name)
if not remote:
return None
return CloudCache(self.repo, name=remote)
tree = get_cloud_tree(self.repo, name=remote)
return CloudCache(tree)
getter.__name__ = name
return cached_property(getter)
@@ -50,7 +52,8 @@ class Cache:
CACHE_DIR = "cache"
def __init__(self, repo):
from dvc.remote import Cache as CloudCache
from dvc.remote import get_cloud_tree
from dvc.remote.local import LocalCache
self.repo = repo
self.config = config = repo.config["cache"]
@@ -62,7 +65,8 @@ class Cache:
else:
settings = {**config, "url": config["dir"]}
self.local = CloudCache(repo, **settings)
tree = get_cloud_tree(repo, **settings)
self.local = LocalCache(tree)
s3 = _make_remote_property("s3")
gs = _make_remote_property("gs")
dvc/command/version.py
+4 -4
View File
@@ -122,12 +122,12 @@ class CmdVersion(CmdBaseNoRepo):
@staticmethod
def get_supported_remotes():
from dvc.remote import REMOTES
from dvc.remote import TREES
supported_remotes = []
for remote in REMOTES:
if not remote.get_missing_deps():
supported_remotes.append(remote.scheme)
for tree_cls in TREES:
if not tree_cls.get_missing_deps():
supported_remotes.append(tree_cls.scheme)
return ", ".join(supported_remotes)
dvc/data_cloud.py
+3 -3
View File
@@ -3,7 +3,7 @@
import logging
from dvc.config import NoRemoteError
from dvc.remote import Remote
from dvc.remote import get_remote
logger = logging.getLogger(__name__)
@@ -45,7 +45,7 @@ class DataCloud:
raise NoRemoteError(error_msg)
def _init_remote(self, name):
return Remote(self.repo, name=name)
return get_remote(self.repo, name=name)
def push(
self, cache, jobs=None, remote=None, show_checksums=False,
@@ -85,7 +85,7 @@ class DataCloud:
cache, jobs=jobs, remote=remote, show_checksums=show_checksums
)
if not remote.verify:
if not remote.tree.verify:
self._save_pulled_checksums(cache)
return downloaded_items_num
dvc/dependency/__init__.py
+2 -2
View File
@@ -12,7 +12,7 @@ from dvc.dependency.param import ParamsDependency
from dvc.dependency.s3 import S3Dependency
from dvc.dependency.ssh import SSHDependency
from dvc.output.base import BaseOutput
from dvc.remote import Remote
from dvc.remote import get_remote
from dvc.scheme import Schemes
from .repo import RepoDependency
@@ -54,7 +54,7 @@ SCHEMA.update(ParamsDependency.PARAM_SCHEMA)
def _get(stage, p, info):
parsed = urlparse(p) if p else None
if parsed and parsed.scheme == "remote":
remote = Remote(stage.repo, name=parsed.netloc)
remote = get_remote(stage.repo, name=parsed.netloc)
return DEP_MAP[remote.scheme](stage, p, info, remote=remote)
if info and info.get(RepoDependency.PARAM_REPO):
dvc/dependency/azure.py
+2 -2
View File
@@ -1,7 +1,7 @@
from dvc.dependency.base import BaseDependency
from dvc.output.base import BaseOutput
from dvc.remote.azure import AzureRemote
from dvc.remote.azure import AzureRemoteTree
class AzureDependency(BaseDependency, BaseOutput):
REMOTE = AzureRemote
TREE_CLS = AzureRemoteTree
dvc/dependency/http.py
+2 -2
View File
@@ -1,7 +1,7 @@
from dvc.dependency.base import BaseDependency
from dvc.output.base import BaseOutput
from dvc.remote.http import HTTPRemote
from dvc.remote.http import HTTPRemoteTree
class HTTPDependency(BaseDependency, BaseOutput):
REMOTE = HTTPRemote
TREE_CLS = HTTPRemoteTree
dvc/dependency/https.py
+2 -2
View File
@@ -1,7 +1,7 @@
from dvc.remote.https import HTTPSRemote
from dvc.remote.https import HTTPSRemoteTree
from .http import HTTPDependency
class HTTPSDependency(HTTPDependency):
REMOTE = HTTPSRemote
TREE_CLS = HTTPSRemoteTree
dvc/output/__init__.py
+8 -8
View File
@@ -10,10 +10,10 @@ from dvc.output.hdfs import HDFSOutput
from dvc.output.local import LocalOutput
from dvc.output.s3 import S3Output
from dvc.output.ssh import SSHOutput
from dvc.remote import Remote
from dvc.remote.hdfs import HDFSRemote
from dvc.remote.local import LocalRemote
from dvc.remote.s3 import S3Remote
from dvc.remote import get_remote
from dvc.remote.hdfs import HDFSRemoteTree
from dvc.remote.local import LocalRemoteTree
from dvc.remote.s3 import S3RemoteTree
from dvc.scheme import Schemes
OUTS = [
@@ -47,9 +47,9 @@ CHECKSUM_SCHEMA = Any(
# so when a few types of outputs share the same name, we only need
# specify it once.
CHECKSUMS_SCHEMA = {
LocalRemote.PARAM_CHECKSUM: CHECKSUM_SCHEMA,
S3Remote.PARAM_CHECKSUM: CHECKSUM_SCHEMA,
HDFSRemote.PARAM_CHECKSUM: CHECKSUM_SCHEMA,
LocalRemoteTree.PARAM_CHECKSUM: CHECKSUM_SCHEMA,
S3RemoteTree.PARAM_CHECKSUM: CHECKSUM_SCHEMA,
HDFSRemoteTree.PARAM_CHECKSUM: CHECKSUM_SCHEMA,
}
SCHEMA = CHECKSUMS_SCHEMA.copy()
@@ -66,7 +66,7 @@ def _get(
parsed = urlparse(p)
if parsed.scheme == "remote":
remote = Remote(stage.repo, name=parsed.netloc)
remote = get_remote(stage.repo, name=parsed.netloc)
return OUTS_MAP[remote.scheme](
stage,
p,
dvc/output/base.py
+17 -12
View File
@@ -12,7 +12,7 @@ from dvc.exceptions import (
DvcException,
RemoteCacheRequiredError,
)
from dvc.remote.base import BaseRemote
from dvc.remote.base import BaseRemoteTree, Remote
logger = logging.getLogger(__name__)
@@ -47,7 +47,8 @@ class OutputIsStageFileError(DvcException):
class BaseOutput:
IS_DEPENDENCY = False
REMOTE = BaseRemote
REMOTE_CLS = Remote
TREE_CLS = BaseRemoteTree
PARAM_PATH = "path"
PARAM_CACHE = "cache"
@@ -105,7 +106,11 @@ class BaseOutput:
self.repo = stage.repo if stage else None
self.def_path = path
self.info = info
self.remote = remote or self.REMOTE(self.repo, {})
if remote:
self.remote = remote
else:
tree = self.TREE_CLS(self.repo, {})
self.remote = self.REMOTE_CLS(tree)
self.use_cache = False if self.IS_DEPENDENCY else cache
self.metric = False if self.IS_DEPENDENCY else metric
self.plot = False if self.IS_DEPENDENCY else plot
@@ -119,7 +124,7 @@ class BaseOutput:
if remote:
parsed = urlparse(path)
return remote.path_info / parsed.path.lstrip("/")
return self.REMOTE.TREE_CLS.PATH_CLS(path)
return self.TREE_CLS.PATH_CLS(path)
def __repr__(self):
return "{class_name}: '{def_path}'".format(
@@ -131,7 +136,7 @@ class BaseOutput:
@property
def scheme(self):
return self.REMOTE.scheme
return self.TREE_CLS.scheme
@property
def is_in_repo(self):
@@ -154,7 +159,7 @@ class BaseOutput:
@classmethod
def supported(cls, url):
return cls.REMOTE.supported(url)
return cls.TREE_CLS.supported(url)
@property
def cache_path(self):
@@ -162,15 +167,15 @@ class BaseOutput:
@property
def checksum_type(self):
return self.remote.PARAM_CHECKSUM
return self.remote.tree.PARAM_CHECKSUM
@property
def checksum(self):
return self.info.get(self.remote.PARAM_CHECKSUM)
return self.info.get(self.remote.tree.PARAM_CHECKSUM)
@checksum.setter
def checksum(self, checksum):
self.info[self.remote.PARAM_CHECKSUM] = checksum
self.info[self.remote.tree.PARAM_CHECKSUM] = checksum
def get_checksum(self):
return self.remote.get_checksum(self.path_info)
@@ -357,7 +362,7 @@ class BaseOutput:
def unprotect(self):
if self.exists:
self.remote.unprotect(self.path_info)
self.remote.tree.unprotect(self.path_info)
def get_dir_cache(self, **kwargs):
if not self.is_dir_checksum:
@@ -417,8 +422,8 @@ class BaseOutput:
filter_path = str(filter_info) if filter_info else None
is_win = os.name == "nt"
for entry in self.dir_cache:
checksum = entry[self.remote.PARAM_CHECKSUM]
entry_relpath = entry[self.remote.PARAM_RELPATH]
checksum = entry[self.remote.tree.PARAM_CHECKSUM]
entry_relpath = entry[self.remote.tree.PARAM_RELPATH]
if is_win:
entry_relpath = entry_relpath.replace("/", os.sep)
entry_path = os.path.join(path, entry_relpath)
dvc/output/gs.py
+2 -2
View File
@@ -1,6 +1,6 @@
from dvc.output.s3 import S3Output
from dvc.remote.gs import GSRemote
from dvc.remote.gs import GSRemoteTree
class GSOutput(S3Output):
REMOTE = GSRemote
TREE_CLS = GSRemoteTree
dvc/output/hdfs.py
+2 -2
View File
@@ -1,6 +1,6 @@
from dvc.output.base import BaseOutput
from dvc.remote.hdfs import HDFSRemote
from dvc.remote.hdfs import HDFSRemoteTree
class HDFSOutput(BaseOutput):
REMOTE = HDFSRemote
TREE_CLS = HDFSRemoteTree
dvc/output/local.py
+5 -4
View File
@@ -5,7 +5,7 @@ from urllib.parse import urlparse
from dvc.exceptions import DvcException
from dvc.istextfile import istextfile
from dvc.output.base import BaseOutput
from dvc.remote.local import LocalRemote
from dvc.remote.local import LocalRemote, LocalRemoteTree
from dvc.utils import relpath
from dvc.utils.fs import path_isin
@@ -13,7 +13,8 @@ logger = logging.getLogger(__name__)
class LocalOutput(BaseOutput):
REMOTE = LocalRemote
REMOTE_CLS = LocalRemote
TREE_CLS = LocalRemoteTree
sep = os.sep
def __init__(self, stage, path, *args, **kwargs):
@@ -33,12 +34,12 @@ class LocalOutput(BaseOutput):
#
# FIXME: if we have Windows path containing / or posix one with \
# then we have #2059 bug and can't really handle that.
p = self.REMOTE.TREE_CLS.PATH_CLS(path)
p = self.TREE_CLS.PATH_CLS(path)
if not p.is_absolute():
p = self.stage.wdir / p
abs_p = os.path.abspath(os.path.normpath(p))
return self.REMOTE.TREE_CLS.PATH_CLS(abs_p)
return self.TREE_CLS.PATH_CLS(abs_p)
def __str__(self):
if not self.is_in_repo:
dvc/output/s3.py
+2 -2
View File
@@ -1,6 +1,6 @@
from dvc.output.base import BaseOutput
from dvc.remote.s3 import S3Remote
from dvc.remote.s3 import S3RemoteTree
class S3Output(BaseOutput):
REMOTE = S3Remote
TREE_CLS = S3RemoteTree
dvc/output/ssh.py
+3 -2
View File
@@ -1,6 +1,7 @@
from dvc.output.base import BaseOutput
from dvc.remote.ssh import SSHRemote
from dvc.remote.ssh import SSHRemote, SSHRemoteTree
class SSHOutput(BaseOutput):
REMOTE = SSHRemote
REMOTE_CLS = SSHRemote
TREE_CLS = SSHRemoteTree
dvc/remote/__init__.py
+41 -45
View File
@@ -1,45 +1,37 @@
import posixpath
from urllib.parse import urlparse
from dvc.remote.azure import AzureCache, AzureRemote
from dvc.remote.gdrive import GDriveRemote
from dvc.remote.gs import GSCache, GSRemote
from dvc.remote.hdfs import HDFSCache, HDFSRemote
from dvc.remote.http import HTTPRemote
from dvc.remote.https import HTTPSRemote
from dvc.remote.local import LocalCache, LocalRemote
from dvc.remote.oss import OSSRemote
from dvc.remote.s3 import S3Cache, S3Remote
from dvc.remote.ssh import SSHCache, SSHRemote
from dvc.remote.azure import AzureRemoteTree
from dvc.remote.base import Remote
from dvc.remote.gdrive import GDriveRemoteTree
from dvc.remote.gs import GSRemoteTree
from dvc.remote.hdfs import HDFSRemoteTree
from dvc.remote.http import HTTPRemoteTree
from dvc.remote.https import HTTPSRemoteTree
from dvc.remote.local import LocalRemote, LocalRemoteTree
from dvc.remote.oss import OSSRemoteTree
from dvc.remote.s3 import S3RemoteTree
from dvc.remote.ssh import SSHRemote, SSHRemoteTree
CACHES = [
AzureCache,
GSCache,
HDFSCache,
S3Cache,
SSHCache,
# LocalCache is the default
]
REMOTES = [
AzureRemote,
GDriveRemote,
GSRemote,
HDFSRemote,
HTTPRemote,
HTTPSRemote,
S3Remote,
SSHRemote,
OSSRemote,
# NOTE: LocalRemote is the default
TREES = [
AzureRemoteTree,
GDriveRemoteTree,
GSRemoteTree,
HDFSRemoteTree,
HTTPRemoteTree,
HTTPSRemoteTree,
S3RemoteTree,
SSHRemoteTree,
OSSRemoteTree,
# NOTE: LocalRemoteTree is the default
]
def _get(remote_conf, remotes, default):
for remote in remotes:
if remote.supported(remote_conf):
return remote
return default
def _get_tree(remote_conf):
for tree_cls in TREES:
if tree_cls.supported(remote_conf):
return tree_cls
return LocalRemoteTree
def _get_conf(repo, **kwargs):
@@ -51,16 +43,6 @@ def _get_conf(repo, **kwargs):
return _resolve_remote_refs(repo.config, remote_conf)
def Remote(repo, **kwargs):
remote_conf = _get_conf(repo, **kwargs)
return _get(remote_conf, REMOTES, LocalRemote)(repo, remote_conf)
def Cache(repo, **kwargs):
remote_conf = _get_conf(repo, **kwargs)
return _get(remote_conf, CACHES, LocalCache)(repo, remote_conf)
def _resolve_remote_refs(config, remote_conf):
# Support for cross referenced remotes.
# This will merge the settings, shadowing base ref with remote_conf.
@@ -90,3 +72,17 @@ def _resolve_remote_refs(config, remote_conf):
base = config["remote"][parsed.netloc]
url = posixpath.join(base["url"], parsed.path.lstrip("/"))
return {**base, **remote_conf, "url": url}
def get_cloud_tree(repo, **kwargs):
remote_conf = _get_conf(repo, **kwargs)
return _get_tree(remote_conf)(repo, remote_conf)
def get_remote(repo, **kwargs):
tree = get_cloud_tree(repo, **kwargs)
if tree.scheme == "local":
return LocalRemote(tree)
if tree.scheme == "ssh":
return SSHRemote(tree)
return Remote(tree)
dvc/remote/azure.py
+8 -16
View File
@@ -7,17 +7,22 @@ from funcy import cached_property, wrap_prop
from dvc.path_info import CloudURLInfo
from dvc.progress import Tqdm
from dvc.remote.base import BaseRemote, BaseRemoteTree, CacheMixin
from dvc.remote.base import BaseRemoteTree
from dvc.scheme import Schemes
logger = logging.getLogger(__name__)
class AzureRemoteTree(BaseRemoteTree):
scheme = Schemes.AZURE
PATH_CLS = CloudURLInfo
REQUIRES = {"azure-storage-blob": "azure.storage.blob"}
PARAM_CHECKSUM = "etag"
COPY_POLL_SECONDS = 5
LIST_OBJECT_PAGE_SIZE = 5000
def __init__(self, remote, config):
super().__init__(remote, config)
def __init__(self, repo, config):
super().__init__(repo, config)
url = config.get("url", "azure://")
self.path_info = self.PATH_CLS(url)
@@ -132,16 +137,3 @@ class AzureRemoteTree(BaseRemoteTree):
to_file,
progress_callback=pbar.update_to,
)
class AzureRemote(BaseRemote):
scheme = Schemes.AZURE
REQUIRES = {"azure-storage-blob": "azure.storage.blob"}
TREE_CLS = AzureRemoteTree
PARAM_CHECKSUM = "etag"
COPY_POLL_SECONDS = 5
LIST_OBJECT_PAGE_SIZE = 5000
class AzureCache(AzureRemote, CacheMixin):
pass
dvc/remote/base.py
+384 -331
View File
@@ -73,24 +73,103 @@ class DirCacheError(DvcException):
def index_locked(f):
@wraps(f)
def wrapper(remote_obj, *args, **kwargs):
remote = kwargs.get("remote", remote_obj)
with remote.index:
return f(remote_obj, *args, **kwargs)
def wrapper(obj, named_cache, remote, *args, **kwargs):
if hasattr(remote, "index"):
with remote.index:
return f(obj, named_cache, remote, *args, **kwargs)
return f(obj, named_cache, remote, *args, **kwargs)
return wrapper
class BaseRemoteTree:
SHARED_MODE_MAP = {None: (None, None), "group": (None, None)}
scheme = "base"
REQUIRES = {}
PATH_CLS = URLInfo
JOBS = 4 * cpu_count()
PARAM_RELPATH = "relpath"
CHECKSUM_DIR_SUFFIX = ".dir"
CHECKSUM_JOBS = max(1, min(4, cpu_count() // 2))
DEFAULT_VERIFY = False
LIST_OBJECT_PAGE_SIZE = 1000
TRAVERSE_WEIGHT_MULTIPLIER = 5
TRAVERSE_PREFIX_LEN = 3
TRAVERSE_THRESHOLD_SIZE = 500000
CAN_TRAVERSE = True
CACHE_MODE = None
SHARED_MODE_MAP = {None: (None, None), "group": (None, None)}
CHECKSUM_DIR_SUFFIX = ".dir"
def __init__(self, remote, config):
self.remote = remote
state = StateNoop()
def __init__(self, repo, config):
self.repo = repo
self.config = config
self._check_requires(config)
shared = config.get("shared")
self._file_mode, self._dir_mode = self.SHARED_MODE_MAP[shared]
self.checksum_jobs = (
config.get("checksum_jobs")
or (self.repo and self.repo.config["core"].get("checksum_jobs"))
or self.CHECKSUM_JOBS
)
self.verify = config.get("verify", self.DEFAULT_VERIFY)
@classmethod
def get_missing_deps(cls):
import importlib
missing = []
for package, module in cls.REQUIRES.items():
try:
importlib.import_module(module)
except ImportError:
missing.append(package)
return missing
def _check_requires(self, config):
missing = self.get_missing_deps()
if not missing:
return
url = config.get("url", f"{self.scheme}://")
msg = (
"URL '{}' is supported but requires these missing "
"dependencies: {}. If you have installed dvc using pip, "
"choose one of these options to proceed: \n"
"\n"
" 1) Install specific missing dependencies:\n"
" pip install {}\n"
" 2) Install dvc package that includes those missing "
"dependencies: \n"
" pip install 'dvc[{}]'\n"
" 3) Install dvc package with all possible "
"dependencies included: \n"
" pip install 'dvc[all]'\n"
"\n"
"If you have installed dvc from a binary package and you "
"are still seeing this message, please report it to us "
"using https://github.com/iterative/dvc/issues. Thank you!"
).format(url, missing, " ".join(missing), self.scheme)
raise RemoteMissingDepsError(msg)
@classmethod
def supported(cls, config):
if isinstance(config, (str, bytes)):
url = config
else:
url = config["url"]
# NOTE: silently skipping remote, calling code should handle that
parsed = urlparse(url)
return parsed.scheme == cls.scheme
@property
def file_mode(self):
return self._file_mode
@@ -99,17 +178,9 @@ class BaseRemoteTree:
def dir_mode(self):
return self._dir_mode
@property
def scheme(self):
return self.remote.scheme
@property
def state(self):
return self.remote.state
@property
def cache(self):
return self.remote.cache
return getattr(self.repo.cache, self.scheme)
def open(self, path_info, mode="r", encoding=None):
if hasattr(self, "_generate_download_url"):
@@ -172,6 +243,17 @@ class BaseRemoteTree:
def reflink(self, from_info, to_info):
raise RemoteActionNotImplemented("reflink", self.scheme)
@staticmethod
def protect(path_info):
pass
def is_protected(self, path_info):
return False
@staticmethod
def unprotect(path_info):
pass
@classmethod
def is_dir_checksum(cls, checksum):
if not checksum:
@@ -225,6 +307,24 @@ class BaseRemoteTree:
dir_info = self._collect_dir(path_info, tree, **kwargs)
return self._save_dir_info(dir_info, path_info)
def checksum_to_path_info(self, checksum):
return self.path_info / checksum[0:2] / checksum[2:]
def path_to_checksum(self, path):
parts = self.PATH_CLS(path).parts[-2:]
if not (len(parts) == 2 and parts[0] and len(parts[0]) == 2):
raise ValueError(f"Bad cache file path '{path}'")
return "".join(parts)
def save_info(self, path_info, tree=None, **kwargs):
return {
self.PARAM_CHECKSUM: self.get_checksum(
path_info, tree=tree, **kwargs
)
}
def _calculate_checksums(self, file_infos, tree):
file_infos = list(file_infos)
with Tqdm(
@@ -234,7 +334,7 @@ class BaseRemoteTree:
) as pbar:
worker = pbar.wrap_fn(tree.get_file_checksum)
with ThreadPoolExecutor(
max_workers=self.remote.checksum_jobs
max_workers=self.checksum_jobs
) as executor:
tasks = executor.map(worker, file_infos)
checksums = dict(zip(file_infos, tasks))
@@ -259,7 +359,7 @@ class BaseRemoteTree:
result = [
{
self.remote.PARAM_CHECKSUM: checksums[fi],
self.PARAM_CHECKSUM: checksums[fi],
# NOTE: this is lossy transformation:
# "hey\there" -> "hey/there"
# "hey/there" -> "hey/there"
@@ -268,15 +368,13 @@ class BaseRemoteTree:
#
# Yes, this is a BUG, as long as we permit "/" in
# filenames on Windows and "\" on Unix
self.remote.PARAM_RELPATH: fi.relative_to(
path_info
).as_posix(),
self.PARAM_RELPATH: fi.relative_to(path_info).as_posix(),
}
for fi in file_infos
]
# Sorting the list by path to ensure reproducibility
return sorted(result, key=itemgetter(self.remote.PARAM_RELPATH))
return sorted(result, key=itemgetter(self.PARAM_RELPATH))
def _save_dir_info(self, dir_info, path_info):
checksum, tmp_info = self._get_dir_info_checksum(dir_info)
@@ -284,7 +382,7 @@ class BaseRemoteTree:
if self.cache.changed_cache_file(checksum):
self.cache.tree.makedirs(new_info.parent)
self.cache.tree.move(
tmp_info, new_info, mode=self.remote.CACHE_MODE
tmp_info, new_info, mode=self.cache.CACHE_MODE
)
if self.exists(path_info):
@@ -380,7 +478,7 @@ class BaseRemoteTree:
dir_mode=dir_mode,
)
)
with ThreadPoolExecutor(max_workers=self.remote.JOBS) as executor:
with ThreadPoolExecutor(max_workers=self.JOBS) as executor:
futures = [
executor.submit(download_files, from_info, to_info)
for from_info, to_info in zip(from_infos, to_infos)
@@ -417,161 +515,6 @@ class BaseRemoteTree:
move(tmp_file, to_info, mode=file_mode)
class BaseRemote:
"""Base cloud remote class."""
scheme = "base"
REQUIRES = {}
JOBS = 4 * cpu_count()
INDEX_CLS = RemoteIndex
TREE_CLS = BaseRemoteTree
PARAM_RELPATH = "relpath"
CHECKSUM_DIR_SUFFIX = ".dir"
CHECKSUM_JOBS = max(1, min(4, cpu_count() // 2))
DEFAULT_CACHE_TYPES = ["copy"]
DEFAULT_VERIFY = False
LIST_OBJECT_PAGE_SIZE = 1000
TRAVERSE_WEIGHT_MULTIPLIER = 5
TRAVERSE_PREFIX_LEN = 3
TRAVERSE_THRESHOLD_SIZE = 500000
CAN_TRAVERSE = True
CACHE_MODE = None
state = StateNoop()
def __init__(self, repo, config):
self.repo = repo
self._check_requires(config)
self.checksum_jobs = (
config.get("checksum_jobs")
or (self.repo and self.repo.config["core"].get("checksum_jobs"))
or self.CHECKSUM_JOBS
)
self.verify = config.get("verify", self.DEFAULT_VERIFY)
self._dir_info = {}
self.cache_types = config.get("type") or copy(self.DEFAULT_CACHE_TYPES)
self.cache_type_confirmed = False
url = config.get("url")
if url:
index_name = hashlib.sha256(url.encode("utf-8")).hexdigest()
self.index = self.INDEX_CLS(
self.repo, index_name, dir_suffix=self.CHECKSUM_DIR_SUFFIX
)
else:
self.index = RemoteIndexNoop()
self.tree = self.TREE_CLS(self, config)
@property
def path_info(self):
return self.tree.path_info
@classmethod
def get_missing_deps(cls):
import importlib
missing = []
for package, module in cls.REQUIRES.items():
try:
importlib.import_module(module)
except ImportError:
missing.append(package)
return missing
def _check_requires(self, config):
missing = self.get_missing_deps()
if not missing:
return
url = config.get("url", f"{self.scheme}://")
msg = (
"URL '{}' is supported but requires these missing "
"dependencies: {}. If you have installed dvc using pip, "
"choose one of these options to proceed: \n"
"\n"
" 1) Install specific missing dependencies:\n"
" pip install {}\n"
" 2) Install dvc package that includes those missing "
"dependencies: \n"
" pip install 'dvc[{}]'\n"
" 3) Install dvc package with all possible "
"dependencies included: \n"
" pip install 'dvc[all]'\n"
"\n"
"If you have installed dvc from a binary package and you "
"are still seeing this message, please report it to us "
"using https://github.com/iterative/dvc/issues. Thank you!"
).format(url, missing, " ".join(missing), self.scheme)
raise RemoteMissingDepsError(msg)
def __repr__(self):
return "{class_name}: '{path_info}'".format(
class_name=type(self).__name__,
path_info=self.path_info or "No path",
)
@classmethod
def supported(cls, config):
if isinstance(config, (str, bytes)):
url = config
else:
url = config["url"]
# NOTE: silently skipping remote, calling code should handle that
parsed = urlparse(url)
return parsed.scheme == cls.scheme
@property
def cache(self):
return getattr(self.repo.cache, self.scheme)
@classmethod
def is_dir_checksum(cls, checksum):
return cls.TREE_CLS.is_dir_checksum(checksum)
def get_checksum(self, path_info, **kwargs):
return self.tree.get_checksum(path_info, **kwargs)
def checksum_to_path_info(self, checksum):
return self.path_info / checksum[0:2] / checksum[2:]
def path_to_checksum(self, path):
parts = self.tree.PATH_CLS(path).parts[-2:]
if not (len(parts) == 2 and parts[0] and len(parts[0]) == 2):
raise ValueError(f"Bad cache file path '{path}'")
return "".join(parts)
def save_info(self, path_info, tree=None, **kwargs):
return {
self.PARAM_CHECKSUM: self.tree.get_checksum(
path_info, tree=tree, **kwargs
)
}
def open(self, *args, **kwargs):
return self.tree.open(*args, **kwargs)
@staticmethod
def protect(path_info):
pass
def is_protected(self, path_info):
return False
@staticmethod
def unprotect(path_info):
pass
def list_paths(self, prefix=None, progress_callback=None):
if prefix:
if len(prefix) > 2:
@@ -581,14 +524,14 @@ class BaseRemote:
else:
path_info = self.path_info
if progress_callback:
for file_info in self.tree.walk_files(path_info):
for file_info in self.walk_files(path_info):
progress_callback()
yield file_info.path
else:
yield from self.tree.walk_files(path_info)
yield from self.walk_files(path_info)
def list_checksums(self, prefix=None, progress_callback=None):
"""Iterate over remote checksums.
"""Iterate over checksums in this tree.
If `prefix` is specified, only checksums which begin with `prefix`
will be returned.
@@ -602,7 +545,7 @@ class BaseRemote:
)
def all(self, jobs=None, name=None):
"""Iterate over all checksums in the remote.
"""Iterate over all checksums in this tree.
Checksums will be fetched in parallel threads according to prefix
(except for small remotes) and a progress bar will be displayed.
@@ -616,104 +559,11 @@ class BaseRemote:
if not self.CAN_TRAVERSE:
return self.list_checksums()
remote_size, remote_checksums = self._estimate_remote_size(name=name)
return self._list_checksums_traverse(
remote_size, remote_checksums = self.estimate_remote_size(name=name)
return self.list_checksums_traverse(
remote_size, remote_checksums, jobs, name
)
def checksums_exist(self, checksums, jobs=None, name=None):
"""Check if the given checksums are stored in the remote.
There are two ways of performing this check:
- Traverse method: Get a list of all the files in the remote
(traversing the cache directory) and compare it with
the given checksums. Cache entries will be retrieved in parallel
threads according to prefix (i.e. entries starting with, "00...",
"01...", and so on) and a progress bar will be displayed.
- Exists method: For each given checksum, run the `exists`
method and filter the checksums that aren't on the remote.
This is done in parallel threads.
It also shows a progress bar when performing the check.
The reason for such an odd logic is that most of the remotes
take much shorter time to just retrieve everything they have under
a certain prefix (e.g. s3, gs, ssh, hdfs). Other remotes that can
check if particular file exists much quicker, use their own
implementation of checksums_exist (see ssh, local).
Which method to use will be automatically determined after estimating
the size of the remote cache, and comparing the estimated size with
len(checksums). To estimate the size of the remote cache, we fetch
a small subset of cache entries (i.e. entries starting with "00...").
Based on the number of entries in that subset, the size of the full
cache can be estimated, since the cache is evenly distributed according
to checksum.
Returns:
A list with checksums that were found in the remote
"""
# Remotes which do not use traverse prefix should override
# checksums_exist() (see ssh, local)
assert self.TRAVERSE_PREFIX_LEN >= 2
checksums = set(checksums)
indexed_checksums = set(self.index.intersection(checksums))
checksums -= indexed_checksums
logger.debug(
"Matched '{}' indexed checksums".format(len(indexed_checksums))
)
if not checksums:
return indexed_checksums
if len(checksums) == 1 or not self.CAN_TRAVERSE:
remote_checksums = self._list_checksums_exists(
checksums, jobs, name
)
return list(indexed_checksums) + remote_checksums
# Max remote size allowed for us to use traverse method
remote_size, remote_checksums = self._estimate_remote_size(
checksums, name
)
traverse_pages = remote_size / self.LIST_OBJECT_PAGE_SIZE
# For sufficiently large remotes, traverse must be weighted to account
# for performance overhead from large lists/sets.
# From testing with S3, for remotes with 1M+ files, object_exists is
# faster until len(checksums) is at least 10k~100k
if remote_size > self.TRAVERSE_THRESHOLD_SIZE:
traverse_weight = traverse_pages * self.TRAVERSE_WEIGHT_MULTIPLIER
else:
traverse_weight = traverse_pages
if len(checksums) < traverse_weight:
logger.debug(
"Large remote ('{}' checksums < '{}' traverse weight), "
"using object_exists for remaining checksums".format(
len(checksums), traverse_weight
)
)
return (
list(indexed_checksums)
+ list(checksums & remote_checksums)
+ self._list_checksums_exists(
checksums - remote_checksums, jobs, name
)
)
logger.debug(
"Querying '{}' checksums via traverse".format(len(checksums))
)
remote_checksums = set(
self._list_checksums_traverse(
remote_size, remote_checksums, jobs, name
)
)
return list(indexed_checksums) + list(
checksums & set(remote_checksums)
)
def _checksums_with_limit(
self, limit, prefix=None, progress_callback=None
):
@@ -737,8 +587,8 @@ class BaseRemote:
* self.LIST_OBJECT_PAGE_SIZE,
)
def _estimate_remote_size(self, checksums=None, name=None):
"""Estimate remote cache size based on number of entries beginning with
def estimate_remote_size(self, checksums=None, name=None):
"""Estimate tree size based on number of entries beginning with
"00..." prefix.
"""
prefix = "0" * self.TRAVERSE_PREFIX_LEN
@@ -772,10 +622,10 @@ class BaseRemote:
logger.debug(f"Estimated remote size: {remote_size} files")
return remote_size, remote_checksums
def _list_checksums_traverse(
def list_checksums_traverse(
self, remote_size, remote_checksums, jobs=None, name=None
):
"""Iterate over all checksums in the remote cache.
"""Iterate over all checksums found in this tree.
Checksums are fetched in parallel according to prefix, except in
cases where the remote size is very small.
@@ -827,7 +677,10 @@ class BaseRemote:
in_remote = executor.map(list_with_update, traverse_prefixes,)
yield from itertools.chain.from_iterable(in_remote)
def _list_checksums_exists(self, checksums, jobs=None, name=None):
def list_checksums_exists(self, checksums, jobs=None, name=None):
"""Return list of the specified checksums which exist in this tree.
Checksums will be queried individually.
"""
logger.debug(
"Querying {} checksums via object_exists".format(len(checksums))
)
@@ -839,7 +692,7 @@ class BaseRemote:
) as pbar:
def exists_with_progress(path_info):
ret = self.tree.exists(path_info)
ret = self.exists(path_info)
pbar.update_msg(str(path_info))
return ret
@@ -849,44 +702,243 @@ class BaseRemote:
ret = list(itertools.compress(checksums, in_remote))
return ret
@index_locked
def gc(self, named_cache, jobs=None):
used = set(named_cache.scheme_keys("local"))
if self.scheme != "":
used.update(named_cache.scheme_keys(self.scheme))
removed = False
# checksums must be sorted to ensure we always remove .dir files first
for checksum in sorted(
self.all(jobs, str(self.path_info)),
key=self.is_dir_checksum,
reverse=True,
):
if checksum in used:
continue
path_info = self.checksum_to_path_info(checksum)
if self.is_dir_checksum(checksum):
# backward compatibility
self._remove_unpacked_dir(checksum)
self.tree.remove(path_info)
removed = True
if removed:
self.index.clear()
return removed
def _remove_unpacked_dir(self, checksum):
pass
class CacheMixin:
"""BaseRemote extensions for cache link/checkout operations."""
class Remote:
"""Cloud remote class.
Provides methods for indexing and garbage collecting trees which contain
DVC remotes.
"""
INDEX_CLS = RemoteIndex
def __init__(self, tree):
self.tree = tree
self.repo = tree.repo
config = tree.config
url = config.get("url")
if url:
index_name = hashlib.sha256(url.encode("utf-8")).hexdigest()
self.index = self.INDEX_CLS(
self.repo, index_name, dir_suffix=self.tree.CHECKSUM_DIR_SUFFIX
)
else:
self.index = RemoteIndexNoop()
@property
def path_info(self):
return self.tree.path_info
def __repr__(self):
return "{class_name}: '{path_info}'".format(
class_name=type(self).__name__,
path_info=self.path_info or "No path",
)
@property
def cache(self):
return getattr(self.repo.cache, self.scheme)
@property
def scheme(self):
return self.tree.scheme
def is_dir_checksum(self, checksum):
return self.tree.is_dir_checksum(checksum)
def get_checksum(self, path_info, **kwargs):
return self.tree.get_checksum(path_info, **kwargs)
def checksum_to_path_info(self, checksum):
return self.tree.checksum_to_path_info(checksum)
def path_to_checksum(self, path):
return self.tree.path_to_checksum(path)
def save_info(self, path_info, **kwargs):
return self.tree.save_info(path_info, **kwargs)
def open(self, *args, **kwargs):
return self.tree.open(*args, **kwargs)
def checksums_exist(self, checksums, jobs=None, name=None):
"""Check if the given checksums are stored in the remote.
There are two ways of performing this check:
- Traverse method: Get a list of all the files in the remote
(traversing the cache directory) and compare it with
the given checksums. Cache entries will be retrieved in parallel
threads according to prefix (i.e. entries starting with, "00...",
"01...", and so on) and a progress bar will be displayed.
- Exists method: For each given checksum, run the `exists`
method and filter the checksums that aren't on the remote.
This is done in parallel threads.
It also shows a progress bar when performing the check.
The reason for such an odd logic is that most of the remotes
take much shorter time to just retrieve everything they have under
a certain prefix (e.g. s3, gs, ssh, hdfs). Other remotes that can
check if particular file exists much quicker, use their own
implementation of checksums_exist (see ssh, local).
Which method to use will be automatically determined after estimating
the size of the remote cache, and comparing the estimated size with
len(checksums). To estimate the size of the remote cache, we fetch
a small subset of cache entries (i.e. entries starting with "00...").
Based on the number of entries in that subset, the size of the full
cache can be estimated, since the cache is evenly distributed according
to checksum.
Returns:
A list with checksums that were found in the remote
"""
# Remotes which do not use traverse prefix should override
# checksums_exist() (see ssh, local)
assert self.tree.TRAVERSE_PREFIX_LEN >= 2
checksums = set(checksums)
indexed_checksums = set(self.index.intersection(checksums))
checksums -= indexed_checksums
logger.debug(
"Matched '{}' indexed checksums".format(len(indexed_checksums))
)
if not checksums:
return indexed_checksums
if len(checksums) == 1 or not self.tree.CAN_TRAVERSE:
remote_checksums = self.tree.list_checksums_exists(
checksums, jobs, name
)
return list(indexed_checksums) + remote_checksums
# Max remote size allowed for us to use traverse method
remote_size, remote_checksums = self.tree.estimate_remote_size(
checksums, name
)
traverse_pages = remote_size / self.tree.LIST_OBJECT_PAGE_SIZE
# For sufficiently large remotes, traverse must be weighted to account
# for performance overhead from large lists/sets.
# From testing with S3, for remotes with 1M+ files, object_exists is
# faster until len(checksums) is at least 10k~100k
if remote_size > self.tree.TRAVERSE_THRESHOLD_SIZE:
traverse_weight = (
traverse_pages * self.tree.TRAVERSE_WEIGHT_MULTIPLIER
)
else:
traverse_weight = traverse_pages
if len(checksums) < traverse_weight:
logger.debug(
"Large remote ('{}' checksums < '{}' traverse weight), "
"using object_exists for remaining checksums".format(
len(checksums), traverse_weight
)
)
return (
list(indexed_checksums)
+ list(checksums & remote_checksums)
+ self.tree.list_checksums_exists(
checksums - remote_checksums, jobs, name
)
)
logger.debug(
"Querying '{}' checksums via traverse".format(len(checksums))
)
remote_checksums = set(
self.tree.list_checksums_traverse(
remote_size, remote_checksums, jobs, name
)
)
return list(indexed_checksums) + list(
checksums & set(remote_checksums)
)
@classmethod
@index_locked
def gc(cls, named_cache, remote, jobs=None):
tree = remote.tree
used = set(named_cache.scheme_keys("local"))
if tree.scheme != "":
used.update(named_cache.scheme_keys(tree.scheme))
removed = False
# checksums must be sorted to ensure we always remove .dir files first
for checksum in sorted(
tree.all(jobs, str(tree.path_info)),
key=tree.is_dir_checksum,
reverse=True,
):
if checksum in used:
continue
path_info = tree.checksum_to_path_info(checksum)
if tree.is_dir_checksum(checksum):
# backward compatibility
tree._remove_unpacked_dir(checksum)
tree.remove(path_info)
removed = True
if removed and hasattr(remote, "index"):
remote.index.clear()
return removed
class CloudCache:
"""Cloud cache class."""
DEFAULT_CACHE_TYPES = ["copy"]
CACHE_MODE = BaseRemoteTree.CACHE_MODE
def __init__(self, tree):
self.tree = tree
self.repo = tree.repo
self.cache_types = tree.config.get("type") or copy(
self.DEFAULT_CACHE_TYPES
)
self.cache_type_confirmed = False
self._dir_info = {}
@property
def path_info(self):
return self.tree.path_info
@property
def cache(self):
return getattr(self.repo.cache, self.scheme)
@property
def scheme(self):
return self.tree.scheme
@property
def state(self):
return self.tree.state
def open(self, *args, **kwargs):
return self.tree.open(*args, **kwargs)
def is_dir_checksum(self, checksum):
return self.tree.is_dir_checksum(checksum)
def get_checksum(self, path_info, **kwargs):
return self.tree.get_checksum(path_info, **kwargs)
# Override to return path as a string instead of PathInfo for clouds
# which support string paths (see local)
def checksum_to_path(self, checksum):
return self.checksum_to_path_info(checksum)
def checksum_to_path_info(self, checksum):
return self.tree.checksum_to_path_info(checksum)
def get_dir_cache(self, checksum):
assert checksum
@@ -922,7 +974,8 @@ class CacheMixin:
# only need to convert it for Windows
for info in d:
# NOTE: here is a BUG, see comment to .as_posix() below
info[self.PARAM_RELPATH] = info[self.PARAM_RELPATH].replace(
relpath = info[self.tree.PARAM_RELPATH]
info[self.tree.PARAM_RELPATH] = relpath.replace(
"/", self.tree.PATH_CLS.sep
)
@@ -953,7 +1006,7 @@ class CacheMixin:
logger.debug("'%s' doesn't exist.", path_info)
return True
checksum = checksum_info.get(self.PARAM_CHECKSUM)
checksum = checksum_info.get(self.tree.PARAM_CHECKSUM)
if checksum is None:
logger.debug("hash value for '%s' is missing.", path_info)
return True
@@ -1037,7 +1090,7 @@ class CacheMixin:
path_info
):
# Default relink procedure involves unneeded copy
self.unprotect(path_info)
self.tree.unprotect(path_info)
else:
self.tree.remove(path_info)
self.link(cache_info, path_info)
@@ -1062,7 +1115,7 @@ class CacheMixin:
callback(1)
self.state.save(cache_info, checksum)
return {self.PARAM_CHECKSUM: checksum}
return {self.tree.PARAM_CHECKSUM: checksum}
def _cache_is_copy(self, path_info):
"""Checks whether cache uses copies."""
@@ -1091,8 +1144,8 @@ class CacheMixin:
for entry in Tqdm(
dir_info, desc="Saving " + path_info.name, unit="file"
):
entry_info = path_info / entry[self.PARAM_RELPATH]
entry_checksum = entry[self.PARAM_CHECKSUM]
entry_info = path_info / entry[self.tree.PARAM_RELPATH]
entry_checksum = entry[self.tree.PARAM_CHECKSUM]
self._save_file(
entry_info, tree, entry_checksum, save_link=False, **kwargs
)
@@ -1104,7 +1157,7 @@ class CacheMixin:
cache_info = self.checksum_to_path_info(checksum)
self.state.save(cache_info, checksum)
return {self.PARAM_CHECKSUM: checksum}
return {self.tree.PARAM_CHECKSUM: checksum}
def save(self, path_info, tree, checksum_info, save_link=True, **kwargs):
if path_info.scheme != self.scheme:
@@ -1113,8 +1166,8 @@ class CacheMixin:
)
if not checksum_info:
checksum_info = self.save_info(path_info, tree=tree, **kwargs)
checksum = checksum_info[self.PARAM_CHECKSUM]
checksum_info = self.tree.save_info(path_info, tree=tree, **kwargs)
checksum = checksum_info[self.tree.PARAM_CHECKSUM]
return self._save(path_info, tree, checksum, save_link, **kwargs)
def _save(self, path_info, tree, checksum, save_link=True, **kwargs):
@@ -1141,7 +1194,7 @@ class CacheMixin:
"""
# Prefer string path over PathInfo when possible due to performance
cache_info = self.checksum_to_path(checksum)
if self.is_protected(cache_info):
if self.tree.is_protected(cache_info):
logger.debug(
"Assuming '%s' is unchanged since it is read-only", cache_info
)
@@ -1162,7 +1215,7 @@ class CacheMixin:
if actual.split(".")[0] == checksum.split(".")[0]:
# making cache file read-only so we don't need to check it
# next time
self.protect(cache_info)
self.tree.protect(cache_info)
return False
if self.tree.exists(cache_info):
@@ -1176,10 +1229,10 @@ class CacheMixin:
return True
for entry in self.get_dir_cache(checksum):
entry_checksum = entry[self.PARAM_CHECKSUM]
entry_checksum = entry[self.tree.PARAM_CHECKSUM]
if path_info and filter_info:
entry_info = path_info / entry[self.PARAM_RELPATH]
entry_info = path_info / entry[self.tree.PARAM_RELPATH]
if not entry_info.isin_or_eq(filter_info):
continue
@@ -1258,15 +1311,15 @@ class CacheMixin:
logger.debug("Linking directory '%s'.", path_info)
for entry in dir_info:
relative_path = entry[self.PARAM_RELPATH]
entry_checksum = entry[self.PARAM_CHECKSUM]
relative_path = entry[self.tree.PARAM_RELPATH]
entry_checksum = entry[self.tree.PARAM_CHECKSUM]
entry_cache_info = self.checksum_to_path_info(entry_checksum)
entry_info = path_info / relative_path
if filter_info and not entry_info.isin_or_eq(filter_info):
continue
entry_checksum_info = {self.PARAM_CHECKSUM: entry_checksum}
entry_checksum_info = {self.tree.PARAM_CHECKSUM: entry_checksum}
if relink or self.changed(entry_info, entry_checksum_info):
modified = True
self.safe_remove(entry_info, force=force)
@@ -1290,7 +1343,7 @@ class CacheMixin:
existing_files = set(self.tree.walk_files(path_info))
needed_files = {
path_info / entry[self.PARAM_RELPATH] for entry in dir_info
path_info / entry[self.tree.PARAM_RELPATH] for entry in dir_info
}
redundant_files = existing_files - needed_files
for path in redundant_files:
@@ -1310,7 +1363,7 @@ class CacheMixin:
if path_info.scheme not in ["local", self.scheme]:
raise NotImplementedError
checksum = checksum_info.get(self.PARAM_CHECKSUM)
checksum = checksum_info.get(self.tree.PARAM_CHECKSUM)
failed = None
skip = False
if not checksum:
@@ -1385,6 +1438,6 @@ class CacheMixin:
return len(self.get_dir_cache(checksum))
return ilen(
filter_info.isin_or_eq(path_info / entry[self.PARAM_CHECKSUM])
filter_info.isin_or_eq(path_info / entry[self.tree.PARAM_CHECKSUM])
for entry in self.get_dir_cache(checksum)
)
dvc/remote/gdrive.py
+11 -16
View File
@@ -14,7 +14,7 @@ from funcy.py3 import cat
from dvc.exceptions import DvcException, FileMissingError
from dvc.path_info import CloudURLInfo
from dvc.progress import Tqdm
from dvc.remote.base import BaseRemote, BaseRemoteTree
from dvc.remote.base import BaseRemoteTree
from dvc.scheme import Schemes
from dvc.utils import format_link, tmp_fname
from dvc.utils.stream import IterStream
@@ -85,7 +85,13 @@ class GDriveURLInfo(CloudURLInfo):
class GDriveRemoteTree(BaseRemoteTree):
scheme = Schemes.GDRIVE
PATH_CLS = GDriveURLInfo
REQUIRES = {"pydrive2": "pydrive2"}
DEFAULT_VERIFY = True
# Always prefer traverse for GDrive since API usage quotas are a concern.
TRAVERSE_WEIGHT_MULTIPLIER = 1
TRAVERSE_PREFIX_LEN = 2
GDRIVE_CREDENTIALS_DATA = "GDRIVE_CREDENTIALS_DATA"
DEFAULT_USER_CREDENTIALS_FILE = "gdrive-user-credentials.json"
@@ -93,8 +99,8 @@ class GDriveRemoteTree(BaseRemoteTree):
DEFAULT_GDRIVE_CLIENT_ID = "710796635688-iivsgbgsb6uv1fap6635dhvuei09o66c.apps.googleusercontent.com" # noqa: E501
DEFAULT_GDRIVE_CLIENT_SECRET = "a1Fz59uTpVNeG_VGuSKDLJXv"
def __init__(self, remote, config):
super().__init__(remote, config)
def __init__(self, repo, config):
super().__init__(repo, config)
self.path_info = self.PATH_CLS(config["url"])
@@ -122,13 +128,12 @@ class GDriveRemoteTree(BaseRemoteTree):
self._client_secret = config.get("gdrive_client_secret")
self._validate_config()
self._gdrive_user_credentials_path = (
tmp_fname(os.path.join(self.remote.repo.tmp_dir, ""))
tmp_fname(os.path.join(self.repo.tmp_dir, ""))
if os.getenv(GDriveRemoteTree.GDRIVE_CREDENTIALS_DATA)
else config.get(
"gdrive_user_credentials_file",
os.path.join(
self.remote.repo.tmp_dir,
self.DEFAULT_USER_CREDENTIALS_FILE,
self.repo.tmp_dir, self.DEFAULT_USER_CREDENTIALS_FILE,
),
)
)
@@ -574,13 +579,3 @@ class GDriveRemoteTree(BaseRemoteTree):
def _download(self, from_info, to_file, name=None, no_progress_bar=False):
item_id = self._get_item_id(from_info)
self._gdrive_download_file(item_id, to_file, name, no_progress_bar)
class GDriveRemote(BaseRemote):
scheme = Schemes.GDRIVE
REQUIRES = {"pydrive2": "pydrive2"}
TREE_CLS = GDriveRemoteTree
DEFAULT_VERIFY = True
# Always prefer traverse for GDrive since API usage quotas are a concern.
TRAVERSE_WEIGHT_MULTIPLIER = 1
TRAVERSE_PREFIX_LEN = 2
dvc/remote/gs.py
+6 -14
View File
@@ -9,7 +9,7 @@ from funcy import cached_property, wrap_prop
from dvc.exceptions import DvcException
from dvc.path_info import CloudURLInfo
from dvc.progress import Tqdm
from dvc.remote.base import BaseRemote, BaseRemoteTree, CacheMixin
from dvc.remote.base import BaseRemoteTree
from dvc.scheme import Schemes
logger = logging.getLogger(__name__)
@@ -65,10 +65,13 @@ def _upload_to_bucket(
class GSRemoteTree(BaseRemoteTree):
scheme = Schemes.GS
PATH_CLS = CloudURLInfo
REQUIRES = {"google-cloud-storage": "google.cloud.storage"}
PARAM_CHECKSUM = "md5"
def __init__(self, remote, config):
super().__init__(remote, config)
def __init__(self, repo, config):
super().__init__(repo, config)
url = config.get("url", "gs:///")
self.path_info = self.PATH_CLS(url)
@@ -193,14 +196,3 @@ class GSRemoteTree(BaseRemoteTree):
disable=no_progress_bar,
) as wrapped:
blob.download_to_file(wrapped)
class GSRemote(BaseRemote):
scheme = Schemes.GS
REQUIRES = {"google-cloud-storage": "google.cloud.storage"}
TREE_CLS = GSRemoteTree
PARAM_CHECKSUM = "md5"
class GSCache(GSRemote, CacheMixin):
pass
dvc/remote/hdfs.py
+9 -16
View File
@@ -11,15 +11,21 @@ from urllib.parse import urlparse
from dvc.scheme import Schemes
from dvc.utils import fix_env, tmp_fname
from .base import BaseRemote, BaseRemoteTree, CacheMixin, RemoteCmdError
from .base import BaseRemoteTree, RemoteCmdError
from .pool import get_connection
logger = logging.getLogger(__name__)
class HDFSRemoteTree(BaseRemoteTree):
def __init__(self, remote, config):
super().__init__(remote, config)
scheme = Schemes.HDFS
REQUIRES = {"pyarrow": "pyarrow"}
REGEX = r"^hdfs://((?P<user>.*)@)?.*$"
PARAM_CHECKSUM = "checksum"
TRAVERSE_PREFIX_LEN = 2
def __init__(self, repo, config):
super().__init__(repo, config)
self.path_info = None
url = config.get("url")
@@ -173,16 +179,3 @@ class HDFSRemoteTree(BaseRemoteTree):
with self.hdfs(from_info) as hdfs:
with open(to_file, "wb+") as fobj:
hdfs.download(from_info.path, fobj)
class HDFSRemote(BaseRemote):
scheme = Schemes.HDFS
REGEX = r"^hdfs://((?P<user>.*)@)?.*$"
PARAM_CHECKSUM = "checksum"
REQUIRES = {"pyarrow": "pyarrow"}
TREE_CLS = HDFSRemoteTree
TRAVERSE_PREFIX_LEN = 2
class HDFSCache(HDFSRemote, CacheMixin):
pass
dvc/remote/http.py
+6 -16
View File
@@ -8,7 +8,7 @@ import dvc.prompt as prompt
from dvc.exceptions import DvcException, HTTPError
from dvc.path_info import HTTPURLInfo
from dvc.progress import Tqdm
from dvc.remote.base import BaseRemote, BaseRemoteTree
from dvc.remote.base import BaseRemoteTree
from dvc.scheme import Schemes
logger = logging.getLogger(__name__)
@@ -24,15 +24,18 @@ def ask_password(host, user):
class HTTPRemoteTree(BaseRemoteTree):
scheme = Schemes.HTTP
PATH_CLS = HTTPURLInfo
PARAM_CHECKSUM = "etag"
CAN_TRAVERSE = False
SESSION_RETRIES = 5
SESSION_BACKOFF_FACTOR = 0.1
REQUEST_TIMEOUT = 10
CHUNK_SIZE = 2 ** 16
def __init__(self, remote, config):
super().__init__(remote, config)
def __init__(self, repo, config):
super().__init__(repo, config)
url = config.get("url")
if url:
@@ -185,16 +188,3 @@ class HTTPRemoteTree(BaseRemoteTree):
def _content_length(response):
res = response.headers.get("Content-Length")
return int(res) if res else None
class HTTPRemote(BaseRemote):
scheme = Schemes.HTTP
PARAM_CHECKSUM = "etag"
CAN_TRAVERSE = False
TREE_CLS = HTTPRemoteTree
def list_paths(self, prefix=None, progress_callback=None):
raise NotImplementedError
def gc(self):
raise NotImplementedError
dvc/remote/https.py
+2 -2
View File
@@ -1,7 +1,7 @@
from dvc.scheme import Schemes
from .http import HTTPRemote
from .http import HTTPRemoteTree
class HTTPSRemote(HTTPRemote):
class HTTPSRemoteTree(HTTPRemoteTree):
scheme = Schemes.HTTPS
dvc/remote/local.py
+83 -89
View File
@@ -16,9 +16,9 @@ from dvc.remote.base import (
STATUS_MAP,
STATUS_MISSING,
STATUS_NEW,
BaseRemote,
BaseRemoteTree,
CacheMixin,
CloudCache,
Remote,
index_locked,
)
from dvc.remote.index import RemoteIndexNoop
@@ -39,17 +39,24 @@ logger = logging.getLogger(__name__)
class LocalRemoteTree(BaseRemoteTree):
SHARED_MODE_MAP = {None: (0o644, 0o755), "group": (0o664, 0o775)}
scheme = Schemes.LOCAL
PATH_CLS = PathInfo
PARAM_CHECKSUM = "md5"
PARAM_PATH = "path"
TRAVERSE_PREFIX_LEN = 2
UNPACKED_DIR_SUFFIX = ".unpacked"
def __init__(self, remote, config):
super().__init__(remote, config)
CACHE_MODE = 0o444
SHARED_MODE_MAP = {None: (0o644, 0o755), "group": (0o664, 0o775)}
def __init__(self, repo, config):
super().__init__(repo, config)
url = config.get("url")
self.path_info = self.PATH_CLS(url) if url else None
@property
def repo(self):
return self.remote.repo
def state(self):
return self.repo.state
@cached_property
def work_tree(self):
@@ -195,78 +202,6 @@ class LocalRemoteTree(BaseRemoteTree):
os.chmod(tmp_info, self.file_mode)
os.rename(tmp_info, to_info)
def get_file_checksum(self, path_info):
return file_md5(path_info)[0]
@staticmethod
def getsize(path_info):
return os.path.getsize(path_info)
def _upload(
self, from_file, to_info, name=None, no_progress_bar=False, **_kwargs
):
makedirs(to_info.parent, exist_ok=True)
tmp_file = tmp_fname(to_info)
copyfile(
from_file, tmp_file, name=name, no_progress_bar=no_progress_bar
)
self.remote.protect(tmp_file)
os.rename(tmp_file, to_info)
@staticmethod
def _download(
from_info, to_file, name=None, no_progress_bar=False, **_kwargs
):
copyfile(
from_info, to_file, no_progress_bar=no_progress_bar, name=name
)
def _log_exceptions(func, operation):
@wraps(func)
def wrapper(from_info, to_info, *args, **kwargs):
try:
func(from_info, to_info, *args, **kwargs)
return 0
except Exception as exc:
# NOTE: this means we ran out of file descriptors and there is no
# reason to try to proceed, as we will hit this error anyways.
if isinstance(exc, OSError) and exc.errno == errno.EMFILE:
raise
logger.exception(
"failed to %s '%s' to '%s'", operation, from_info, to_info
)
return 1
return wrapper
class LocalRemote(BaseRemote):
scheme = Schemes.LOCAL
INDEX_CLS = RemoteIndexNoop
TREE_CLS = LocalRemoteTree
PARAM_CHECKSUM = "md5"
PARAM_PATH = "path"
DEFAULT_CACHE_TYPES = ["reflink", "copy"]
TRAVERSE_PREFIX_LEN = 2
UNPACKED_DIR_SUFFIX = ".unpacked"
CACHE_MODE = 0o444
@property
def state(self):
return self.repo.state
def get(self, md5):
if not md5:
return None
return self.checksum_to_path_info(md5).url
def _unprotect_file(self, path):
if System.is_symlink(path) or System.is_hardlink(path):
logger.debug(f"Unprotecting '{path}'")
@@ -287,7 +222,7 @@ class LocalRemote(BaseRemote):
"a symlink or a hardlink.".format(path)
)
os.chmod(path, self.tree.file_mode)
os.chmod(path, self.file_mode)
def _unprotect_dir(self, path):
assert is_working_tree(self.repo.tree)
@@ -333,14 +268,44 @@ class LocalRemote(BaseRemote):
return stat.S_IMODE(mode) == self.CACHE_MODE
def get_file_checksum(self, path_info):
return file_md5(path_info)[0]
@staticmethod
def getsize(path_info):
return os.path.getsize(path_info)
def _upload(
self, from_file, to_info, name=None, no_progress_bar=False, **_kwargs
):
makedirs(to_info.parent, exist_ok=True)
tmp_file = tmp_fname(to_info)
copyfile(
from_file, tmp_file, name=name, no_progress_bar=no_progress_bar
)
self.protect(tmp_file)
os.rename(tmp_file, to_info)
@staticmethod
def _download(
from_info, to_file, name=None, no_progress_bar=False, **_kwargs
):
copyfile(
from_info, to_file, no_progress_bar=no_progress_bar, name=name
)
def list_paths(self, prefix=None, progress_callback=None):
assert self.path_info is not None
if prefix:
path_info = self.path_info / prefix[:2]
if not self.tree.exists(path_info):
if not self.exists(path_info):
return
else:
path_info = self.path_info
# NOTE: use utils.fs walk_files since tree.walk_files will not follow
# symlinks
if progress_callback:
for path in walk_files(path_info):
progress_callback()
@@ -351,13 +316,40 @@ class LocalRemote(BaseRemote):
def _remove_unpacked_dir(self, checksum):
info = self.checksum_to_path_info(checksum)
path_info = info.with_name(info.name + self.UNPACKED_DIR_SUFFIX)
self.tree.remove(path_info)
self.remove(path_info)
class LocalCache(LocalRemote, CacheMixin):
def __init__(self, repo, config):
super().__init__(repo, config)
self.cache_dir = config.get("url")
def _log_exceptions(func, operation):
@wraps(func)
def wrapper(from_info, to_info, *args, **kwargs):
try:
func(from_info, to_info, *args, **kwargs)
return 0
except Exception as exc:
# NOTE: this means we ran out of file descriptors and there is no
# reason to try to proceed, as we will hit this error anyways.
if isinstance(exc, OSError) and exc.errno == errno.EMFILE:
raise
logger.exception(
"failed to %s '%s' to '%s'", operation, from_info, to_info
)
return 1
return wrapper
class LocalRemote(Remote):
INDEX_CLS = RemoteIndexNoop
class LocalCache(CloudCache):
DEFAULT_CACHE_TYPES = ["reflink", "copy"]
CACHE_MODE = LocalRemoteTree.CACHE_MODE
def __init__(self, tree):
super().__init__(tree)
self.cache_dir = tree.config.get("url")
@property
def cache_dir(self):
@@ -513,7 +505,7 @@ class LocalCache(LocalRemote, CacheMixin):
indexed_dir_exists = set()
if indexed_dirs:
indexed_dir_exists.update(
remote._list_checksums_exists(indexed_dirs)
remote.tree.list_checksums_exists(indexed_dirs)
)
missing_dirs = indexed_dirs.difference(indexed_dir_exists)
if missing_dirs:
@@ -525,7 +517,9 @@ class LocalCache(LocalRemote, CacheMixin):
# Check if non-indexed (new) dir checksums exist on remote
dir_exists = dir_md5s.intersection(indexed_dir_exists)
dir_exists.update(remote._list_checksums_exists(dir_md5s - dir_exists))
dir_exists.update(
remote.tree.list_checksums_exists(dir_md5s - dir_exists)
)
# If .dir checksum exists on the remote, assume directory contents
# still exists on the remote
@@ -605,7 +599,7 @@ class LocalCache(LocalRemote, CacheMixin):
desc = "Uploading"
if jobs is None:
jobs = remote.JOBS
jobs = remote.tree.JOBS
dir_status, file_status, dir_contents = self._status(
named_cache,
dvc/remote/oss.py
+24 -28
View File
@@ -6,14 +6,37 @@ from funcy import cached_property, wrap_prop
from dvc.path_info import CloudURLInfo
from dvc.progress import Tqdm
from dvc.remote.base import BaseRemote, BaseRemoteTree
from dvc.remote.base import BaseRemoteTree
from dvc.scheme import Schemes
logger = logging.getLogger(__name__)
class OSSRemoteTree(BaseRemoteTree):
"""
oss2 document:
https://www.alibabacloud.com/help/doc-detail/32026.htm
Examples
----------
$ dvc remote add myremote oss://my-bucket/path
Set key id, key secret and endpoint using modify command
$ dvc remote modify myremote oss_key_id my-key-id
$ dvc remote modify myremote oss_key_secret my-key-secret
$ dvc remote modify myremote oss_endpoint endpoint
or environment variables
$ export OSS_ACCESS_KEY_ID="my-key-id"
$ export OSS_ACCESS_KEY_SECRET="my-key-secret"
$ export OSS_ENDPOINT="endpoint"
"""
scheme = Schemes.OSS
PATH_CLS = CloudURLInfo
REQUIRES = {"oss2": "oss2"}
PARAM_CHECKSUM = "etag"
COPY_POLL_SECONDS = 5
LIST_OBJECT_PAGE_SIZE = 100
def __init__(self, repo, config):
super().__init__(repo, config)
@@ -105,30 +128,3 @@ class OSSRemoteTree(BaseRemoteTree):
self.oss_service.get_object_to_file(
from_info.path, to_file, progress_callback=pbar.update_to
)
class OSSRemote(BaseRemote):
"""
oss2 document:
https://www.alibabacloud.com/help/doc-detail/32026.htm
Examples
----------
$ dvc remote add myremote oss://my-bucket/path
Set key id, key secret and endpoint using modify command
$ dvc remote modify myremote oss_key_id my-key-id
$ dvc remote modify myremote oss_key_secret my-key-secret
$ dvc remote modify myremote oss_endpoint endpoint
or environment variables
$ export OSS_ACCESS_KEY_ID="my-key-id"
$ export OSS_ACCESS_KEY_SECRET="my-key-secret"
$ export OSS_ENDPOINT="endpoint"
"""
scheme = Schemes.OSS
REQUIRES = {"oss2": "oss2"}
PARAM_CHECKSUM = "etag"
COPY_POLL_SECONDS = 5
LIST_OBJECT_PAGE_SIZE = 100
TREE_CLS = OSSRemoteTree
dvc/remote/s3.py
+4 -12
View File
@@ -8,14 +8,17 @@ from dvc.config import ConfigError
from dvc.exceptions import DvcException, ETagMismatchError
from dvc.path_info import CloudURLInfo
from dvc.progress import Tqdm
from dvc.remote.base import BaseRemote, BaseRemoteTree, CacheMixin
from dvc.remote.base import BaseRemoteTree
from dvc.scheme import Schemes
logger = logging.getLogger(__name__)
class S3RemoteTree(BaseRemoteTree):
scheme = Schemes.S3
PATH_CLS = CloudURLInfo
REQUIRES = {"boto3": "boto3"}
PARAM_CHECKSUM = "etag"
def __init__(self, repo, config):
super().__init__(repo, config)
@@ -334,14 +337,3 @@ class S3RemoteTree(BaseRemoteTree):
self.s3.download_file(
from_info.bucket, from_info.path, to_file, Callback=pbar.update
)
class S3Remote(BaseRemote):
scheme = Schemes.S3
REQUIRES = {"boto3": "boto3"}
PARAM_CHECKSUM = "etag"
TREE_CLS = S3RemoteTree
class S3Cache(S3Remote, CacheMixin):
pass
dvc/remote/ssh/__init__.py
+22 -25
View File
@@ -14,7 +14,7 @@ from funcy import first, memoize, silent, wrap_with
import dvc.prompt as prompt
from dvc.progress import Tqdm
from dvc.remote.base import BaseRemote, BaseRemoteTree, CacheMixin
from dvc.remote.base import BaseRemoteTree, Remote
from dvc.remote.pool import get_connection
from dvc.scheme import Schemes
from dvc.utils import to_chunks
@@ -34,6 +34,18 @@ def ask_password(host, user, port):
class SSHRemoteTree(BaseRemoteTree):
scheme = Schemes.SSH
REQUIRES = {"paramiko": "paramiko"}
JOBS = 4
PARAM_CHECKSUM = "md5"
# At any given time some of the connections will go over network and
# paramiko stuff, so we would ideally have it double of server processors.
# We use conservative setting of 4 instead to not exhaust max sessions.
CHECKSUM_JOBS = 4
DEFAULT_CACHE_TYPES = ["copy"]
TRAVERSE_PREFIX_LEN = 2
DEFAULT_PORT = 22
TIMEOUT = 1800
@@ -256,27 +268,12 @@ class SSHRemoteTree(BaseRemoteTree):
no_progress_bar=no_progress_bar,
)
class SSHRemote(BaseRemote):
scheme = Schemes.SSH
REQUIRES = {"paramiko": "paramiko"}
JOBS = 4
TREE_CLS = SSHRemoteTree
PARAM_CHECKSUM = "md5"
# At any given time some of the connections will go over network and
# paramiko stuff, so we would ideally have it double of server processors.
# We use conservative setting of 4 instead to not exhaust max sessions.
CHECKSUM_JOBS = 4
DEFAULT_CACHE_TYPES = ["copy"]
TRAVERSE_PREFIX_LEN = 2
def list_paths(self, prefix=None, progress_callback=None):
if prefix:
root = posixpath.join(self.path_info.path, prefix[:2])
else:
root = self.path_info.path
with self.tree.ssh(self.path_info) as ssh:
with self.ssh(self.path_info) as ssh:
if prefix and not ssh.exists(root):
return
# If we simply return an iterator then with above closes instantly
@@ -287,6 +284,8 @@ class SSHRemote(BaseRemote):
else:
yield from ssh.walk_files(root)
class SSHRemote(Remote):
def batch_exists(self, path_infos, callback):
def _exists(chunk_and_channel):
chunk, channel = chunk_and_channel
@@ -321,8 +320,8 @@ class SSHRemote(BaseRemote):
faster than current approach (relying on exists(path_info)) applied in
remote/base.
"""
if not self.CAN_TRAVERSE:
return list(set(checksums) & set(self.all()))
if not self.tree.CAN_TRAVERSE:
return list(set(checksums) & set(self.tree.all()))
# possibly prompt for credentials before "Querying" progress output
self.tree.ensure_credentials()
@@ -337,14 +336,12 @@ class SSHRemote(BaseRemote):
def exists_with_progress(chunks):
return self.batch_exists(chunks, callback=pbar.update_msg)
with ThreadPoolExecutor(max_workers=jobs or self.JOBS) as executor:
with ThreadPoolExecutor(
max_workers=jobs or self.tree.JOBS
) as executor:
path_infos = [self.checksum_to_path_info(x) for x in checksums]
chunks = to_chunks(path_infos, num_chunks=self.JOBS)
chunks = to_chunks(path_infos, num_chunks=self.tree.JOBS)
results = executor.map(exists_with_progress, chunks)
in_remote = itertools.chain.from_iterable(results)
ret = list(itertools.compress(checksums, in_remote))
return ret
class SSHCache(SSHRemote, CacheMixin):
pass
dvc/repo/__init__.py
+1 -1
View File
@@ -187,7 +187,7 @@ class Repo:
return Repo(root_dir)
def unprotect(self, target):
return self.cache.local.unprotect(PathInfo(target))
return self.cache.local.tree.unprotect(PathInfo(target))
def _ignore(self):
flist = [self.config.files["local"], self.tmp_dir]
dvc/repo/gc.py
+12 -9
View File
@@ -8,8 +8,10 @@ from . import locked
logger = logging.getLogger(__name__)
def _do_gc(typ, func, clist, jobs=None):
removed = func(clist, jobs=jobs)
def _do_gc(typ, remote, clist, jobs=None):
from dvc.remote.base import Remote
removed = Remote.gc(clist, remote, jobs=jobs)
if not removed:
logger.info(f"No unused '{typ}' cache to remove.")
@@ -74,22 +76,23 @@ def gc(
)
)
_do_gc("local", self.cache.local.gc, used, jobs)
# treat caches as remotes for garbage collection
_do_gc("local", self.cache.local, used, jobs)
if self.cache.s3:
_do_gc("s3", self.cache.s3.gc, used, jobs)
_do_gc("s3", self.cache.s3, used, jobs)
if self.cache.gs:
_do_gc("gs", self.cache.gs.gc, used, jobs)
_do_gc("gs", self.cache.gs, used, jobs)
if self.cache.ssh:
_do_gc("ssh", self.cache.ssh.gc, used, jobs)
_do_gc("ssh", self.cache.ssh, used, jobs)
if self.cache.hdfs:
_do_gc("hdfs", self.cache.hdfs.gc, used, jobs)
_do_gc("hdfs", self.cache.hdfs, used, jobs)
if self.cache.azure:
_do_gc("azure", self.cache.azure.gc, used, jobs)
_do_gc("azure", self.cache.azure, used, jobs)
if cloud:
_do_gc("remote", self.cloud.get_remote(remote, "gc -c").gc, used, jobs)
_do_gc("remote", self.cloud.get_remote(remote, "gc -c"), used, jobs)
dvc/repo/tree.py
+3 -3
View File
@@ -48,11 +48,11 @@ class DvcTree(BaseTree):
raise FileNotFoundError
dir_cache = out.get_dir_cache(remote=remote)
for entry in dir_cache:
entry_relpath = entry[out.remote.PARAM_RELPATH]
entry_relpath = entry[out.remote.tree.PARAM_RELPATH]
if os.name == "nt":
entry_relpath = entry_relpath.replace("/", os.sep)
if path == out.path_info / entry_relpath:
return entry[out.remote.PARAM_CHECKSUM]
return entry[out.remote.tree.PARAM_CHECKSUM]
raise FileNotFoundError
def open(self, path, mode="r", encoding="utf-8", remote=None):
@@ -194,7 +194,7 @@ class DvcTree(BaseTree):
download_callback(downloaded)
for entry in dir_cache:
entry_relpath = entry[out.remote.PARAM_RELPATH]
entry_relpath = entry[out.remote.tree.PARAM_RELPATH]
if os.name == "nt":
entry_relpath = entry_relpath.replace("/", os.sep)
path_info = out.path_info / entry_relpath
dvc/stage/utils.py
+4 -3
View File
@@ -8,7 +8,8 @@ from dvc import dependency, output
from dvc.utils.fs import path_isin
from ..dependency import ParamsDependency
from ..remote import LocalRemote, S3Remote
from ..remote.local import LocalRemoteTree
from ..remote.s3 import S3RemoteTree
from ..utils import dict_md5, format_link, relpath
from .exceptions import (
MissingDataSource,
@@ -132,8 +133,8 @@ def stage_dump_eq(stage_cls, old_d, new_d):
new_d.pop(stage_cls.PARAM_MD5, None)
outs = old_d.get(stage_cls.PARAM_OUTS, [])
for out in outs:
out.pop(LocalRemote.PARAM_CHECKSUM, None)
out.pop(S3Remote.PARAM_CHECKSUM, None)
out.pop(LocalRemoteTree.PARAM_CHECKSUM, None)
out.pop(S3RemoteTree.PARAM_CHECKSUM, None)
# outs and deps are lists of dicts. To check equality, we need to make
# them independent of the order, so, we convert them to dicts.
tests/func/remote/test_index.py
+3 -3
View File
@@ -3,7 +3,7 @@ import os
import pytest
from dvc.exceptions import DownloadError, UploadError
from dvc.remote.base import BaseRemote
from dvc.remote.base import Remote
from dvc.remote.index import RemoteIndex
from dvc.remote.local import LocalRemote, LocalRemoteTree
from dvc.utils.fs import remove
@@ -16,9 +16,9 @@ def remote(tmp_dir, dvc, tmp_path_factory, mocker):
dvc.config["core"]["remote"] = "upstream"
# patch checksums_exist since the LocalRemote normally overrides
# BaseRemote.checksums_exist.
# BaseRemoteTree.checksums_exist.
def checksums_exist(self, *args, **kwargs):
return BaseRemote.checksums_exist(self, *args, **kwargs)
return Remote.checksums_exist(self, *args, **kwargs)
mocker.patch.object(LocalRemote, "checksums_exist", checksums_exist)
tests/func/test_add.py
+2 -2
View File
@@ -21,7 +21,7 @@ from dvc.exceptions import (
)
from dvc.main import main
from dvc.output.base import OutputAlreadyTrackedError, OutputIsStageFileError
from dvc.remote.local import LocalRemote, LocalRemoteTree
from dvc.remote.local import LocalRemoteTree
from dvc.repo import Repo as DvcRepo
from dvc.stage import Stage
from dvc.system import System
@@ -583,7 +583,7 @@ def test_readding_dir_should_not_unprotect_all(tmp_dir, dvc, mocker):
dvc.add("dir")
tmp_dir.gen("dir/new_file", "new_file_content")
unprotect_spy = mocker.spy(LocalRemote, "unprotect")
unprotect_spy = mocker.spy(LocalRemoteTree, "unprotect")
dvc.add("dir")
assert not unprotect_spy.mock.called
tests/func/test_cache.py
+5 -5
View File
@@ -30,14 +30,14 @@ class TestCache(TestDvc):
self.create(self.cache2, "2")
def test_all(self):
md5_list = list(Cache(self.dvc).local.all())
md5_list = list(Cache(self.dvc).local.tree.all())
self.assertEqual(len(md5_list), 2)
self.assertIn(self.cache1_md5, md5_list)
self.assertIn(self.cache2_md5, md5_list)
def test_get(self):
cache = Cache(self.dvc).local.get(self.cache1_md5)
self.assertEqual(cache, self.cache1)
cache = Cache(self.dvc).local.checksum_to_path_info(self.cache1_md5)
self.assertEqual(os.fspath(cache), self.cache1)
class TestCacheLoadBadDirCache(TestDvc):
@@ -47,13 +47,13 @@ class TestCacheLoadBadDirCache(TestDvc):
def test(self):
checksum = "123.dir"
fname = self.dvc.cache.local.get(checksum)
fname = os.fspath(self.dvc.cache.local.checksum_to_path_info(checksum))
self.create(fname, "<clearly>not,json")
with pytest.raises(DirCacheError):
self.dvc.cache.local.load_dir_cache(checksum)
checksum = "234.dir"
fname = self.dvc.cache.local.get(checksum)
fname = os.fspath(self.dvc.cache.local.checksum_to_path_info(checksum))
self.create(fname, '{"a": "b"}')
self._do_test(self.dvc.cache.local.load_dir_cache(checksum))
tests/func/test_checkout.py
+9 -8
View File
@@ -16,8 +16,9 @@ from dvc.exceptions import (
NoOutputOrStageError,
)
from dvc.main import main
from dvc.remote import S3Cache, S3Remote
from dvc.remote.local import LocalRemote
from dvc.remote.base import CloudCache, Remote
from dvc.remote.local import LocalRemoteTree
from dvc.remote.s3 import S3RemoteTree
from dvc.repo import Repo as DvcRepo
from dvc.stage import Stage
from dvc.stage.exceptions import StageFileDoesNotExistError
@@ -99,8 +100,8 @@ class TestCheckoutCorruptedCacheDir(TestDvc):
# NOTE: modifying cache file for one of the files inside the directory
# to check if dvc will detect that the cache is corrupted.
entry = self.dvc.cache.local.load_dir_cache(out.checksum)[0]
checksum = entry[self.dvc.cache.local.PARAM_CHECKSUM]
cache = self.dvc.cache.local.get(checksum)
checksum = entry[self.dvc.cache.local.tree.PARAM_CHECKSUM]
cache = os.fspath(self.dvc.cache.local.checksum_to_path_info(checksum))
os.chmod(cache, 0o644)
with open(cache, "w+") as fobj:
@@ -305,8 +306,8 @@ class TestGitIgnoreWhenCheckout(CheckoutBase):
class TestCheckoutMissingMd5InStageFile(TestRepro):
def test(self):
d = load_yaml(self.file1_stage)
del d[Stage.PARAM_OUTS][0][LocalRemote.PARAM_CHECKSUM]
del d[Stage.PARAM_DEPS][0][LocalRemote.PARAM_CHECKSUM]
del d[Stage.PARAM_OUTS][0][LocalRemoteTree.PARAM_CHECKSUM]
del d[Stage.PARAM_DEPS][0][LocalRemoteTree.PARAM_CHECKSUM]
dump_yaml(self.file1_stage, d)
with pytest.raises(CheckoutError):
@@ -755,9 +756,9 @@ def test_checkout_recursive(tmp_dir, dvc):
not S3.should_test(), reason="Only run with S3 credentials"
)
def test_checkout_for_external_outputs(tmp_dir, dvc):
dvc.cache.s3 = S3Cache(dvc, {"url": S3.get_url()})
dvc.cache.s3 = CloudCache(S3RemoteTree(dvc, {"url": S3.get_url()}))
remote = S3Remote(dvc, {"url": S3.get_url()})
remote = Remote(S3RemoteTree(dvc, {"url": S3.get_url()}))
file_path = remote.path_info / "foo"
remote.tree.s3.put_object(
Bucket=remote.path_info.bucket, Key=file_path.path, Body="foo"
tests/func/test_data_cloud.py
+43 -37
View File
@@ -12,19 +12,16 @@ from dvc.cache import NamedCache
from dvc.data_cloud import DataCloud
from dvc.external_repo import clean_repos
from dvc.main import main
from dvc.remote import (
AzureRemote,
GDriveRemote,
GSRemote,
HDFSRemote,
HTTPRemote,
LocalRemote,
OSSRemote,
S3Remote,
SSHRemote,
)
from dvc.remote.azure import AzureRemoteTree
from dvc.remote.base import STATUS_DELETED, STATUS_NEW, STATUS_OK
from dvc.remote.gdrive import GDriveRemoteTree
from dvc.remote.gs import GSRemoteTree
from dvc.remote.hdfs import HDFSRemoteTree
from dvc.remote.http import HTTPRemoteTree
from dvc.remote.local import LocalRemoteTree
from dvc.remote.oss import OSSRemoteTree
from dvc.remote.s3 import S3RemoteTree
from dvc.remote.ssh import SSHRemoteTree
from dvc.stage.exceptions import StageNotFound
from dvc.utils import file_md5
from dvc.utils.fs import remove
@@ -50,19 +47,19 @@ class TestDataCloud(TestDvc):
def _test_cloud(self, config, cl):
self.dvc.config = config
cloud = DataCloud(self.dvc)
self.assertIsInstance(cloud.get_remote(), cl)
self.assertIsInstance(cloud.get_remote().tree, cl)
def test(self):
config = copy.deepcopy(TEST_CONFIG)
clist = [
("s3://mybucket/", S3Remote),
("gs://mybucket/", GSRemote),
("ssh://user@localhost:/", SSHRemote),
("http://localhost:8000/", HTTPRemote),
("azure://ContainerName=mybucket;conn_string;", AzureRemote),
("oss://mybucket/", OSSRemote),
(TestDvc.mkdtemp(), LocalRemote),
("s3://mybucket/", S3RemoteTree),
("gs://mybucket/", GSRemoteTree),
("ssh://user@localhost:/", SSHRemoteTree),
("http://localhost:8000/", HTTPRemoteTree),
("azure://ContainerName=mybucket;conn_string;", AzureRemoteTree),
("oss://mybucket/", OSSRemoteTree),
(TestDvc.mkdtemp(), LocalRemoteTree),
]
for scheme, cl in clist:
@@ -101,7 +98,9 @@ class TestDataCloudBase(TestDvc):
self.dvc.config = config
self.cloud = DataCloud(self.dvc)
self.assertIsInstance(self.cloud.get_remote(), self._get_cloud_class())
self.assertIsInstance(
self.cloud.get_remote().tree, self._get_cloud_class()
)
def _test_cloud(self):
self._setup_cloud()
@@ -187,7 +186,7 @@ class TestDataCloudBase(TestDvc):
class TestS3Remote(S3, TestDataCloudBase):
def _get_cloud_class(self):
return S3Remote
return S3RemoteTree
class TestGDriveRemote(GDrive, TestDataCloudBase):
@@ -208,10 +207,10 @@ class TestGDriveRemote(GDrive, TestDataCloudBase):
self.dvc.config = config
self.cloud = DataCloud(self.dvc)
remote = self.cloud.get_remote()
self.assertIsInstance(remote, self._get_cloud_class())
self.assertIsInstance(remote.tree, self._get_cloud_class())
def _get_cloud_class(self):
return GDriveRemote
return GDriveRemoteTree
class TestGSRemote(GCP, TestDataCloudBase):
@@ -228,25 +227,27 @@ class TestGSRemote(GCP, TestDataCloudBase):
self.dvc.config = config
self.cloud = DataCloud(self.dvc)
self.assertIsInstance(self.cloud.get_remote(), self._get_cloud_class())
self.assertIsInstance(
self.cloud.get_remote().tree, self._get_cloud_class()
)
def _get_cloud_class(self):
return GSRemote
return GSRemoteTree
class TestAzureRemote(Azure, TestDataCloudBase):
def _get_cloud_class(self):
return AzureRemote
return AzureRemoteTree
class TestOSSRemote(OSS, TestDataCloudBase):
def _get_cloud_class(self):
return OSSRemote
return OSSRemoteTree
class TestLocalRemote(Local, TestDataCloudBase):
def _get_cloud_class(self):
return LocalRemote
return LocalRemoteTree
@pytest.mark.usefixtures("ssh_server")
@@ -271,7 +272,9 @@ class TestSSHRemoteMocked(SSHMocked, TestDataCloudBase):
self.dvc.config = config
self.cloud = DataCloud(self.dvc)
self.assertIsInstance(self.cloud.get_remote(), self._get_cloud_class())
self.assertIsInstance(
self.cloud.get_remote().tree, self._get_cloud_class()
)
def get_url(self):
user = self.ssh_server.test_creds["username"]
@@ -281,12 +284,12 @@ class TestSSHRemoteMocked(SSHMocked, TestDataCloudBase):
return self.ssh_server.test_creds["key_filename"]
def _get_cloud_class(self):
return SSHRemote
return SSHRemoteTree
class TestHDFSRemote(HDFS, TestDataCloudBase):
def _get_cloud_class(self):
return HDFSRemote
return HDFSRemoteTree
@pytest.mark.usefixtures("http_server")
@@ -300,7 +303,7 @@ class TestHTTPRemote(HTTP, TestDataCloudBase):
return super().get_url(self.http_server.server_port)
def _get_cloud_class(self):
return HTTPRemote
return HTTPRemoteTree
class TestDataCloudCLIBase(TestDvc):
@@ -544,7 +547,7 @@ class TestRecursiveSyncOperations(Local, TestDataCloudBase):
self.assertEqual(ret, 0)
def _get_cloud_class(self):
return LocalRemote
return LocalRemoteTree
def _prepare_repo(self):
remote = self.cloud.get_remote()
@@ -577,8 +580,11 @@ class TestRecursiveSyncOperations(Local, TestDataCloudBase):
def _test_recursive_fetch(self, data_md5, data_sub_md5):
self._clear_local_cache()
local_cache_data_path = self.dvc.cache.local.get(data_md5)
local_cache_data_sub_path = self.dvc.cache.local.get(data_sub_md5)
local_cache = self.dvc.cache.local
local_cache_data_path = local_cache.checksum_to_path_info(data_md5)
local_cache_data_sub_path = local_cache.checksum_to_path_info(
data_sub_md5
)
self.assertFalse(os.path.exists(local_cache_data_path))
self.assertFalse(os.path.exists(local_cache_data_sub_path))
@@ -590,8 +596,8 @@ class TestRecursiveSyncOperations(Local, TestDataCloudBase):
def _test_recursive_push(self, data_md5, data_sub_md5):
remote = self.cloud.get_remote()
cloud_data_path = remote.get(data_md5)
cloud_data_sub_path = remote.get(data_sub_md5)
cloud_data_path = remote.checksum_to_path_info(data_md5)
cloud_data_sub_path = remote.checksum_to_path_info(data_sub_md5)
self.assertFalse(os.path.exists(cloud_data_path))
self.assertFalse(os.path.exists(cloud_data_sub_path))
tests/func/test_gc.py
+4 -3
View File
@@ -8,7 +8,7 @@ from git import Repo
from dvc.exceptions import CollectCacheError
from dvc.main import main
from dvc.remote.local import LocalRemote, LocalRemoteTree
from dvc.remote.local import LocalRemoteTree
from dvc.repo import Repo as DvcRepo
from dvc.utils.fs import remove
from tests.basic_env import TestDir, TestDvcGit
@@ -21,7 +21,8 @@ class TestGC(TestDvcGit):
self.dvc.add(self.FOO)
self.dvc.add(self.DATA_DIR)
self.good_cache = [
self.dvc.cache.local.get(md5) for md5 in self.dvc.cache.local.all()
self.dvc.cache.local.checksum_to_path_info(md5)
for md5 in self.dvc.cache.local.tree.all()
]
self.bad_cache = []
@@ -216,7 +217,7 @@ def test_gc_no_unpacked_dir(tmp_dir, dvc):
os.remove("dir.dvc")
unpackeddir = (
dir_stages[0].outs[0].cache_path + LocalRemote.UNPACKED_DIR_SUFFIX
dir_stages[0].outs[0].cache_path + LocalRemoteTree.UNPACKED_DIR_SUFFIX
)
# older (pre 1.0) versions of dvc used to generate this dir
tests/func/test_repro.py
+3 -3
View File
@@ -26,7 +26,7 @@ from dvc.exceptions import (
from dvc.main import main
from dvc.output.base import BaseOutput
from dvc.path_info import URLInfo
from dvc.remote.local import LocalRemote, LocalRemoteTree
from dvc.remote.local import LocalRemoteTree
from dvc.repo import Repo as DvcRepo
from dvc.stage import Stage
from dvc.stage.exceptions import StageFileDoesNotExistError
@@ -782,8 +782,8 @@ class TestReproChangedDirData(SingleStageRun, TestDvc):
class TestReproMissingMd5InStageFile(TestRepro):
def test(self):
d = load_yaml(self.file1_stage)
del d[Stage.PARAM_OUTS][0][LocalRemote.PARAM_CHECKSUM]
del d[Stage.PARAM_DEPS][0][LocalRemote.PARAM_CHECKSUM]
del d[Stage.PARAM_OUTS][0][LocalRemoteTree.PARAM_CHECKSUM]
del d[Stage.PARAM_DEPS][0][LocalRemoteTree.PARAM_CHECKSUM]
dump_yaml(self.file1_stage, d)
stages = self.dvc.reproduce(self.file1_stage)
tests/func/test_s3.py
+6 -4
View File
@@ -5,7 +5,8 @@ import moto.s3.models as s3model
import pytest
from moto import mock_s3
from dvc.remote.s3 import S3Cache, S3Remote, S3RemoteTree
from dvc.remote.base import CloudCache
from dvc.remote.s3 import S3RemoteTree
from tests.remotes import S3
# from https://github.com/spulec/moto/blob/v1.3.5/tests/test_s3/test_s3.py#L40
@@ -54,7 +55,8 @@ def test_copy_singlepart_preserve_etag():
],
)
def test_link_created_on_non_nested_path(base_info, tmp_dir, dvc, scm):
cache = S3Cache(dvc, {"url": str(base_info.parent)})
tree = S3RemoteTree(dvc, {"url": str(base_info.parent)})
cache = CloudCache(tree)
s3 = cache.tree.s3
s3.create_bucket(Bucket=base_info.bucket)
s3.put_object(
@@ -69,8 +71,8 @@ def test_link_created_on_non_nested_path(base_info, tmp_dir, dvc, scm):
@mock_s3
def test_makedirs_doesnot_try_on_top_level_paths(tmp_dir, dvc, scm):
base_info = S3RemoteTree.PATH_CLS("s3://bucket/")
remote = S3Remote(dvc, {"url": str(base_info)})
remote.tree.makedirs(base_info)
tree = S3RemoteTree(dvc, {"url": str(base_info)})
tree.makedirs(base_info)
def _upload_multipart(s3, Bucket, Key):
tests/func/test_stage.py
+3 -3
View File
@@ -6,7 +6,7 @@ import pytest
from dvc.dvcfile import SingleStageFile
from dvc.main import main
from dvc.output.local import LocalOutput
from dvc.remote.local import LocalRemote
from dvc.remote.local import LocalRemoteTree
from dvc.repo import Repo
from dvc.stage import PipelineStage, Stage
from dvc.stage.exceptions import StageFileFormatError
@@ -54,8 +54,8 @@ def test_empty_list():
def test_list():
lst = [
{LocalOutput.PARAM_PATH: "foo", LocalRemote.PARAM_CHECKSUM: "123"},
{LocalOutput.PARAM_PATH: "bar", LocalRemote.PARAM_CHECKSUM: None},
{LocalOutput.PARAM_PATH: "foo", LocalRemoteTree.PARAM_CHECKSUM: "123"},
{LocalOutput.PARAM_PATH: "bar", LocalRemoteTree.PARAM_CHECKSUM: None},
{LocalOutput.PARAM_PATH: "baz"},
]
d = {Stage.PARAM_DEPS: lst}
tests/func/test_tree.py
+1 -1
View File
@@ -192,7 +192,7 @@ def test_repotree_walk_fetch(tmp_dir, dvc, scm, setup_remote):
assert os.path.exists(out.cache_path)
for entry in out.dir_cache:
checksum = entry[out.remote.PARAM_CHECKSUM]
checksum = entry[out.remote.tree.PARAM_CHECKSUM]
assert os.path.exists(dvc.cache.local.checksum_to_path_info(checksum))
tests/remotes.py
+8 -7
View File
@@ -7,9 +7,10 @@ from subprocess import CalledProcessError, Popen, check_output
from moto.s3 import mock_s3
from dvc.remote.gdrive import GDriveRemote, GDriveRemoteTree
from dvc.remote.gs import GSRemote
from dvc.remote.s3 import S3Remote
from dvc.remote.base import Remote
from dvc.remote.gdrive import GDriveRemoteTree
from dvc.remote.gs import GSRemoteTree
from dvc.remote.s3 import S3RemoteTree
from dvc.utils import env2bool
from tests.basic_env import TestDvc
@@ -79,7 +80,7 @@ class S3Mocked(S3):
@contextmanager
def remote(cls, repo):
with mock_s3():
yield S3Remote(repo, {"url": cls.get_url()})
yield Remote(S3RemoteTree(repo, {"url": cls.get_url()}))
@staticmethod
def put_objects(remote, objects):
@@ -127,7 +128,7 @@ class GCP:
@classmethod
@contextmanager
def remote(cls, repo):
yield GSRemote(repo, {"url": cls.get_url()})
yield Remote(GSRemoteTree(repo, {"url": cls.get_url()}))
@staticmethod
def put_objects(remote, objects):
@@ -150,8 +151,8 @@ class GDrive:
"gdrive_service_account_p12_file_path": "test.p12",
"gdrive_use_service_account": True,
}
remote = GDriveRemote(dvc, config)
remote.tree._gdrive_create_dir("root", remote.path_info.path)
tree = GDriveRemoteTree(dvc, config)
tree._gdrive_create_dir("root", tree.path_info.path)
@staticmethod
def get_storagepath():
tests/unit/remote/ssh/test_ssh.py
+22 -22
View File
@@ -5,7 +5,7 @@ import sys
import pytest
from mock import mock_open, patch
from dvc.remote.ssh import SSHRemote, SSHRemoteTree
from dvc.remote.ssh import SSHRemoteTree
from dvc.system import System
from tests.remotes import SSHMocked
@@ -20,21 +20,21 @@ def test_url(dvc):
url = f"ssh://{user}@{host}:{port}{path}"
config = {"url": url}
remote = SSHRemote(dvc, config)
assert remote.path_info == url
tree = SSHRemoteTree(dvc, config)
assert tree.path_info == url
# SCP-like URL ssh://[user@]host.xz:/absolute/path
url = f"ssh://{user}@{host}:{path}"
config = {"url": url}
remote = SSHRemote(dvc, config)
assert remote.tree.path_info == url
tree = SSHRemoteTree(dvc, config)
assert tree.path_info == url
def test_no_path(dvc):
config = {"url": "ssh://127.0.0.1"}
remote = SSHRemote(dvc, config)
assert remote.tree.path_info.path == ""
tree = SSHRemoteTree(dvc, config)
assert tree.path_info.path == ""
mock_ssh_config = """
@@ -67,11 +67,11 @@ else:
def test_ssh_host_override_from_config(
mock_file, mock_exists, dvc, config, expected_host
):
remote = SSHRemote(dvc, config)
tree = SSHRemoteTree(dvc, config)
mock_exists.assert_called_with(SSHRemoteTree.ssh_config_filename())
mock_file.assert_called_with(SSHRemoteTree.ssh_config_filename())
assert remote.tree.path_info.host == expected_host
assert tree.path_info.host == expected_host
@pytest.mark.parametrize(
@@ -95,11 +95,11 @@ def test_ssh_host_override_from_config(
read_data=mock_ssh_config,
)
def test_ssh_user(mock_file, mock_exists, dvc, config, expected_user):
remote = SSHRemote(dvc, config)
tree = SSHRemoteTree(dvc, config)
mock_exists.assert_called_with(SSHRemoteTree.ssh_config_filename())
mock_file.assert_called_with(SSHRemoteTree.ssh_config_filename())
assert remote.tree.path_info.user == expected_user
assert tree.path_info.user == expected_user
@pytest.mark.parametrize(
@@ -120,11 +120,11 @@ def test_ssh_user(mock_file, mock_exists, dvc, config, expected_user):
read_data=mock_ssh_config,
)
def test_ssh_port(mock_file, mock_exists, dvc, config, expected_port):
remote = SSHRemote(dvc, config)
tree = SSHRemoteTree(dvc, config)
mock_exists.assert_called_with(SSHRemoteTree.ssh_config_filename())
mock_file.assert_called_with(SSHRemoteTree.ssh_config_filename())
assert remote.path_info.port == expected_port
assert tree.path_info.port == expected_port
@pytest.mark.parametrize(
@@ -155,11 +155,11 @@ def test_ssh_port(mock_file, mock_exists, dvc, config, expected_port):
read_data=mock_ssh_config,
)
def test_ssh_keyfile(mock_file, mock_exists, dvc, config, expected_keyfile):
remote = SSHRemote(dvc, config)
tree = SSHRemoteTree(dvc, config)
mock_exists.assert_called_with(SSHRemoteTree.ssh_config_filename())
mock_file.assert_called_with(SSHRemoteTree.ssh_config_filename())
assert remote.tree.keyfile == expected_keyfile
assert tree.keyfile == expected_keyfile
@pytest.mark.parametrize(
@@ -177,11 +177,11 @@ def test_ssh_keyfile(mock_file, mock_exists, dvc, config, expected_keyfile):
read_data=mock_ssh_config,
)
def test_ssh_gss_auth(mock_file, mock_exists, dvc, config, expected_gss_auth):
remote = SSHRemote(dvc, config)
tree = SSHRemoteTree(dvc, config)
mock_exists.assert_called_with(SSHRemoteTree.ssh_config_filename())
mock_file.assert_called_with(SSHRemoteTree.ssh_config_filename())
assert remote.tree.gss_auth == expected_gss_auth
assert tree.gss_auth == expected_gss_auth
def test_hardlink_optimization(dvc, tmp_dir, ssh_server):
@@ -194,12 +194,12 @@ def test_hardlink_optimization(dvc, tmp_dir, ssh_server):
"user": user,
"keyfile": ssh_server.test_creds["key_filename"],
}
remote = SSHRemote(dvc, config)
tree = SSHRemoteTree(dvc, config)
from_info = remote.path_info / "empty"
to_info = remote.path_info / "link"
from_info = tree.path_info / "empty"
to_info = tree.path_info / "link"
with remote.open(from_info, "wb"):
with tree.open(from_info, "wb"):
pass
if os.name == "nt":
@@ -207,5 +207,5 @@ def test_hardlink_optimization(dvc, tmp_dir, ssh_server):
else:
link_path = to_info.path
remote.tree.hardlink(from_info, to_info)
tree.hardlink(from_info, to_info)
assert not System.is_hardlink(link_path)
tests/unit/remote/test_azure.py
+12 -12
View File
@@ -1,7 +1,7 @@
import pytest
from dvc.path_info import PathInfo
from dvc.remote.azure import AzureRemote
from dvc.remote.azure import AzureRemoteTree
from tests.remotes import Azure
container_name = "container-name"
@@ -18,18 +18,18 @@ def test_init_env_var(monkeypatch, dvc):
monkeypatch.setenv("AZURE_STORAGE_CONNECTION_STRING", connection_string)
config = {"url": "azure://"}
remote = AzureRemote(dvc, config)
assert remote.tree.path_info == "azure://" + container_name
assert remote.tree.connection_string == connection_string
tree = AzureRemoteTree(dvc, config)
assert tree.path_info == "azure://" + container_name
assert tree.connection_string == connection_string
def test_init(dvc):
prefix = "some/prefix"
url = f"azure://{container_name}/{prefix}"
config = {"url": url, "connection_string": connection_string}
remote = AzureRemote(dvc, config)
assert remote.tree.path_info == url
assert remote.tree.connection_string == connection_string
tree = AzureRemoteTree(dvc, config)
assert tree.path_info == url
assert tree.connection_string == connection_string
def test_get_file_checksum(tmp_dir):
@@ -38,11 +38,11 @@ def test_get_file_checksum(tmp_dir):
tmp_dir.gen("foo", "foo")
remote = AzureRemote(None, {})
to_info = remote.tree.PATH_CLS(Azure.get_url())
remote.tree.upload(PathInfo("foo"), to_info)
assert remote.tree.exists(to_info)
checksum = remote.tree.get_file_checksum(to_info)
tree = AzureRemoteTree(None, {})
to_info = tree.PATH_CLS(Azure.get_url())
tree.upload(PathInfo("foo"), to_info)
assert tree.exists(to_info)
checksum = tree.get_file_checksum(to_info)
assert checksum
assert isinstance(checksum, str)
assert checksum.strip("'").strip('"') == checksum
tests/unit/remote/test_base.py
+34 -30
View File
@@ -4,7 +4,12 @@ import mock
import pytest
from dvc.path_info import PathInfo
from dvc.remote.base import BaseRemote, RemoteCmdError, RemoteMissingDepsError
from dvc.remote.base import (
BaseRemoteTree,
Remote,
RemoteCmdError,
RemoteMissingDepsError,
)
class _CallableOrNone:
@@ -15,14 +20,13 @@ class _CallableOrNone:
CallableOrNone = _CallableOrNone()
REMOTE_CLS = BaseRemote
def test_missing_deps(dvc):
requires = {"missing": "missing"}
with mock.patch.object(REMOTE_CLS, "REQUIRES", requires):
with mock.patch.object(BaseRemoteTree, "REQUIRES", requires):
with pytest.raises(RemoteMissingDepsError):
REMOTE_CLS(dvc, {})
BaseRemoteTree(dvc, {})
def test_cmd_error(dvc):
@@ -33,44 +37,44 @@ def test_cmd_error(dvc):
err = "sed: expression #1, char 2: extra characters after command"
with mock.patch.object(
REMOTE_CLS.TREE_CLS,
BaseRemoteTree,
"remove",
side_effect=RemoteCmdError("base", cmd, ret, err),
):
with pytest.raises(RemoteCmdError):
REMOTE_CLS(dvc, config).tree.remove("file")
BaseRemoteTree(dvc, config).remove("file")
@mock.patch.object(BaseRemote, "_list_checksums_traverse")
@mock.patch.object(BaseRemote, "_list_checksums_exists")
@mock.patch.object(BaseRemoteTree, "list_checksums_traverse")
@mock.patch.object(BaseRemoteTree, "list_checksums_exists")
def test_checksums_exist(object_exists, traverse, dvc):
remote = BaseRemote(dvc, {})
remote = Remote(BaseRemoteTree(dvc, {}))
# remote does not support traverse
remote.CAN_TRAVERSE = False
remote.tree.CAN_TRAVERSE = False
with mock.patch.object(
remote, "list_checksums", return_value=list(range(256))
remote.tree, "list_checksums", return_value=list(range(256))
):
checksums = set(range(1000))
remote.checksums_exist(checksums)
object_exists.assert_called_with(checksums, None, None)
traverse.assert_not_called()
remote.CAN_TRAVERSE = True
remote.tree.CAN_TRAVERSE = True
# large remote, small local
object_exists.reset_mock()
traverse.reset_mock()
with mock.patch.object(
remote, "list_checksums", return_value=list(range(256))
remote.tree, "list_checksums", return_value=list(range(256))
):
checksums = list(range(1000))
remote.checksums_exist(checksums)
# verify that _cache_paths_with_max() short circuits
# before returning all 256 remote checksums
max_checksums = math.ceil(
remote._max_estimation_size(checksums)
/ pow(16, remote.TRAVERSE_PREFIX_LEN)
remote.tree._max_estimation_size(checksums)
/ pow(16, remote.tree.TRAVERSE_PREFIX_LEN)
)
assert max_checksums < 256
object_exists.assert_called_with(
@@ -81,15 +85,15 @@ def test_checksums_exist(object_exists, traverse, dvc):
# large remote, large local
object_exists.reset_mock()
traverse.reset_mock()
remote.JOBS = 16
remote.tree.JOBS = 16
with mock.patch.object(
remote, "list_checksums", return_value=list(range(256))
remote.tree, "list_checksums", return_value=list(range(256))
):
checksums = list(range(1000000))
remote.checksums_exist(checksums)
object_exists.assert_not_called()
traverse.assert_called_with(
256 * pow(16, remote.TRAVERSE_PREFIX_LEN),
256 * pow(16, remote.tree.TRAVERSE_PREFIX_LEN),
set(range(256)),
None,
None,
@@ -97,18 +101,18 @@ def test_checksums_exist(object_exists, traverse, dvc):
@mock.patch.object(
BaseRemote, "list_checksums", return_value=[],
BaseRemoteTree, "list_checksums", return_value=[],
)
@mock.patch.object(
BaseRemote, "path_to_checksum", side_effect=lambda x: x,
BaseRemoteTree, "path_to_checksum", side_effect=lambda x: x,
)
def test_list_checksums_traverse(path_to_checksum, list_checksums, dvc):
remote = BaseRemote(dvc, {})
remote.tree.path_info = PathInfo("foo")
tree = BaseRemoteTree(dvc, {})
tree.path_info = PathInfo("foo")
# parallel traverse
size = 256 / remote.JOBS * remote.LIST_OBJECT_PAGE_SIZE
list(remote._list_checksums_traverse(size, {0}))
size = 256 / tree.JOBS * tree.LIST_OBJECT_PAGE_SIZE
list(tree.list_checksums_traverse(size, {0}))
for i in range(1, 16):
list_checksums.assert_any_call(
prefix=f"{i:03x}", progress_callback=CallableOrNone
@@ -121,20 +125,20 @@ def test_list_checksums_traverse(path_to_checksum, list_checksums, dvc):
# default traverse (small remote)
size -= 1
list_checksums.reset_mock()
list(remote._list_checksums_traverse(size - 1, {0}))
list(tree.list_checksums_traverse(size - 1, {0}))
list_checksums.assert_called_with(
prefix=None, progress_callback=CallableOrNone
)
def test_list_checksums(dvc):
remote = BaseRemote(dvc, {})
remote.tree.path_info = PathInfo("foo")
tree = BaseRemoteTree(dvc, {})
tree.path_info = PathInfo("foo")
with mock.patch.object(
remote, "list_paths", return_value=["12/3456", "bar"]
tree, "list_paths", return_value=["12/3456", "bar"]
):
checksums = list(remote.list_checksums())
checksums = list(tree.list_checksums())
assert checksums == ["123456"]
@@ -143,4 +147,4 @@ def test_list_checksums(dvc):
[(None, False), ("", False), ("3456.dir", True), ("3456", False)],
)
def test_is_dir_checksum(checksum, result):
assert BaseRemote.is_dir_checksum(checksum) == result
assert BaseRemoteTree.is_dir_checksum(checksum) == result
tests/unit/remote/test_gdrive.py
+7 -7
View File
@@ -2,7 +2,7 @@ import os
import pytest
from dvc.remote.gdrive import GDriveAuthError, GDriveRemote, GDriveRemoteTree
from dvc.remote.gdrive import GDriveAuthError, GDriveRemoteTree
USER_CREDS_TOKEN_REFRESH_ERROR = '{"access_token": "", "client_id": "", "client_secret": "", "refresh_token": "", "token_expiry": "", "token_uri": "https://oauth2.googleapis.com/token", "user_agent": null, "revoke_uri": "https://oauth2.googleapis.com/revoke", "id_token": null, "id_token_jwt": null, "token_response": {"access_token": "", "expires_in": 3600, "scope": "https://www.googleapis.com/auth/drive.appdata https://www.googleapis.com/auth/drive", "token_type": "Bearer"}, "scopes": ["https://www.googleapis.com/auth/drive", "https://www.googleapis.com/auth/drive.appdata"], "token_info_uri": "https://oauth2.googleapis.com/tokeninfo", "invalid": true, "_class": "OAuth2Credentials", "_module": "oauth2client.client"}' # noqa: E501
@@ -17,21 +17,21 @@ class TestRemoteGDrive:
}
def test_init(self, dvc):
remote = GDriveRemote(dvc, self.CONFIG)
assert str(remote.tree.path_info) == self.CONFIG["url"]
tree = GDriveRemoteTree(dvc, self.CONFIG)
assert str(tree.path_info) == self.CONFIG["url"]
def test_drive(self, dvc):
remote = GDriveRemote(dvc, self.CONFIG)
tree = GDriveRemoteTree(dvc, self.CONFIG)
os.environ[
GDriveRemoteTree.GDRIVE_CREDENTIALS_DATA
] = USER_CREDS_TOKEN_REFRESH_ERROR
with pytest.raises(GDriveAuthError):
remote.tree._drive
tree._drive
os.environ[GDriveRemoteTree.GDRIVE_CREDENTIALS_DATA] = ""
remote = GDriveRemote(dvc, self.CONFIG)
tree = GDriveRemoteTree(dvc, self.CONFIG)
os.environ[
GDriveRemoteTree.GDRIVE_CREDENTIALS_DATA
] = USER_CREDS_MISSED_KEY_ERROR
with pytest.raises(GDriveAuthError):
remote.tree._drive
tree._drive
tests/unit/remote/test_gs.py
+10 -10
View File
@@ -2,7 +2,7 @@ import mock
import pytest
import requests
from dvc.remote.gs import GSRemote, dynamic_chunk_size
from dvc.remote.gs import GSRemoteTree, dynamic_chunk_size
BUCKET = "bucket"
PREFIX = "prefix"
@@ -17,17 +17,17 @@ CONFIG = {
def test_init(dvc):
remote = GSRemote(dvc, CONFIG)
assert remote.tree.path_info == URL
assert remote.tree.projectname == PROJECT
assert remote.tree.credentialpath == CREDENTIALPATH
tree = GSRemoteTree(dvc, CONFIG)
assert tree.path_info == URL
assert tree.projectname == PROJECT
assert tree.credentialpath == CREDENTIALPATH
@mock.patch("google.cloud.storage.Client.from_service_account_json")
def test_gs(mock_client, dvc):
remote = GSRemote(dvc, CONFIG)
assert remote.tree.credentialpath
remote.tree.gs()
tree = GSRemoteTree(dvc, CONFIG)
assert tree.credentialpath
tree.gs()
mock_client.assert_called_once_with(CREDENTIALPATH)
@@ -35,8 +35,8 @@ def test_gs(mock_client, dvc):
def test_gs_no_credspath(mock_client, dvc):
config = CONFIG.copy()
del config["credentialpath"]
remote = GSRemote(dvc, config)
remote.tree.gs()
tree = GSRemoteTree(dvc, config)
tree.gs()
mock_client.assert_called_with(PROJECT)
tests/unit/remote/test_http.py
+15 -15
View File
@@ -2,7 +2,7 @@ import pytest
from dvc.exceptions import HTTPError
from dvc.path_info import URLInfo
from dvc.remote.http import HTTPRemote
from dvc.remote.http import HTTPRemoteTree
from tests.utils.httpd import StaticFileServer
@@ -11,10 +11,10 @@ def test_download_fails_on_error_code(dvc):
url = f"http://localhost:{httpd.server_port}/"
config = {"url": url}
remote = HTTPRemote(dvc, config)
tree = HTTPRemoteTree(dvc, config)
with pytest.raises(HTTPError):
remote.tree._download(URLInfo(url) / "missing.txt", "missing.txt")
tree._download(URLInfo(url) / "missing.txt", "missing.txt")
def test_public_auth_method(dvc):
@@ -25,9 +25,9 @@ def test_public_auth_method(dvc):
"password": "",
}
remote = HTTPRemote(dvc, config)
tree = HTTPRemoteTree(dvc, config)
assert remote.tree._auth_method() is None
assert tree._auth_method() is None
def test_basic_auth_method(dvc):
@@ -44,10 +44,10 @@ def test_basic_auth_method(dvc):
"password": password,
}
remote = HTTPRemote(dvc, config)
tree = HTTPRemoteTree(dvc, config)
assert remote.tree._auth_method() == auth
assert isinstance(remote.tree._auth_method(), HTTPBasicAuth)
assert tree._auth_method() == auth
assert isinstance(tree._auth_method(), HTTPBasicAuth)
def test_digest_auth_method(dvc):
@@ -64,10 +64,10 @@ def test_digest_auth_method(dvc):
"password": password,
}
remote = HTTPRemote(dvc, config)
tree = HTTPRemoteTree(dvc, config)
assert remote.tree._auth_method() == auth
assert isinstance(remote.tree._auth_method(), HTTPDigestAuth)
assert tree._auth_method() == auth
assert isinstance(tree._auth_method(), HTTPDigestAuth)
def test_custom_auth_method(dvc):
@@ -81,8 +81,8 @@ def test_custom_auth_method(dvc):
"password": password,
}
remote = HTTPRemote(dvc, config)
tree = HTTPRemoteTree(dvc, config)
assert remote.tree._auth_method() is None
assert header in remote.tree.headers
assert remote.tree.headers[header] == password
assert tree._auth_method() is None
assert header in tree.headers
assert tree.headers[header] == password
tests/unit/remote/test_local.py
+20 -18
View File
@@ -5,7 +5,8 @@ import pytest
from dvc.cache import NamedCache
from dvc.path_info import PathInfo
from dvc.remote.local import LocalCache
from dvc.remote.index import RemoteIndexNoop
from dvc.remote.local import LocalCache, LocalRemoteTree
def test_status_download_optimization(mocker, dvc):
@@ -13,7 +14,7 @@ def test_status_download_optimization(mocker, dvc):
And the desired files to fetch are already on the local cache,
Don't check the existence of the desired files on the remote cache
"""
cache = LocalCache(dvc, {})
cache = LocalCache(LocalRemoteTree(dvc, {}))
infos = NamedCache()
infos.add("local", "acbd18db4cc2f85cedef654fccc4a4d8", "foo")
@@ -25,6 +26,7 @@ def test_status_download_optimization(mocker, dvc):
other_remote = mocker.Mock()
other_remote.url = "other_remote"
other_remote.checksums_exist.return_value = []
other_remote.index = RemoteIndexNoop()
cache.status(infos, other_remote, download=True)
@@ -33,8 +35,8 @@ def test_status_download_optimization(mocker, dvc):
@pytest.mark.parametrize("link_name", ["hardlink", "symlink"])
def test_is_protected(tmp_dir, dvc, link_name):
cache = LocalCache(dvc, {})
link_method = getattr(cache.tree, link_name)
tree = LocalRemoteTree(dvc, {})
link_method = getattr(tree, link_name)
(tmp_dir / "foo").write_text("foo")
@@ -43,47 +45,47 @@ def test_is_protected(tmp_dir, dvc, link_name):
link_method(foo, link)
assert not cache.is_protected(foo)
assert not cache.is_protected(link)
assert not tree.is_protected(foo)
assert not tree.is_protected(link)
cache.protect(foo)
tree.protect(foo)
assert cache.is_protected(foo)
assert cache.is_protected(link)
assert tree.is_protected(foo)
assert tree.is_protected(link)
cache.unprotect(link)
tree.unprotect(link)
assert not cache.is_protected(link)
assert not tree.is_protected(link)
if os.name == "nt" and link_name == "hardlink":
# NOTE: NTFS doesn't allow deleting read-only files, which forces us to
# set write perms on the link, which propagates to the source.
assert not cache.is_protected(foo)
assert not tree.is_protected(foo)
else:
assert cache.is_protected(foo)
assert tree.is_protected(foo)
@pytest.mark.parametrize("err", [errno.EPERM, errno.EACCES])
def test_protect_ignore_errors(tmp_dir, mocker, err):
tmp_dir.gen("foo", "foo")
foo = PathInfo("foo")
cache = LocalCache(None, {})
tree = LocalRemoteTree(None, {})
cache.protect(foo)
tree.protect(foo)
mock_chmod = mocker.patch(
"os.chmod", side_effect=OSError(err, "something")
)
cache.protect(foo)
tree.protect(foo)
assert mock_chmod.called
def test_protect_ignore_erofs(tmp_dir, mocker):
tmp_dir.gen("foo", "foo")
foo = PathInfo("foo")
cache = LocalCache(None, {})
tree = LocalRemoteTree(None, {})
mock_chmod = mocker.patch(
"os.chmod", side_effect=OSError(errno.EROFS, "read-only fs")
)
cache.protect(foo)
tree.protect(foo)
assert mock_chmod.called
tests/unit/remote/test_oss.py
+6 -6
View File
@@ -1,4 +1,4 @@
from dvc.remote.oss import OSSRemote
from dvc.remote.oss import OSSRemoteTree
bucket_name = "bucket-name"
endpoint = "endpoint"
@@ -15,8 +15,8 @@ def test_init(dvc):
"oss_key_secret": key_secret,
"oss_endpoint": endpoint,
}
remote = OSSRemote(dvc, config)
assert remote.tree.path_info == url
assert remote.tree.endpoint == endpoint
assert remote.tree.key_id == key_id
assert remote.tree.key_secret == key_secret
tree = OSSRemoteTree(dvc, config)
assert tree.path_info == url
assert tree.endpoint == endpoint
assert tree.key_id == key_id
assert tree.key_secret == key_secret
tests/unit/remote/test_remote.py
+15 -13
View File
@@ -1,6 +1,8 @@
import pytest
from dvc.remote import GSRemote, Remote, S3Remote
from dvc.remote import get_cloud_tree
from dvc.remote.gs import GSRemoteTree
from dvc.remote.s3 import S3RemoteTree
def test_remote_with_checksum_jobs(dvc):
@@ -10,32 +12,32 @@ def test_remote_with_checksum_jobs(dvc):
}
dvc.config["core"]["checksum_jobs"] = 200
remote = Remote(dvc, name="with_checksum_jobs")
assert remote.checksum_jobs == 100
tree = get_cloud_tree(dvc, name="with_checksum_jobs")
assert tree.checksum_jobs == 100
def test_remote_without_checksum_jobs(dvc):
dvc.config["remote"]["without_checksum_jobs"] = {"url": "s3://bucket/name"}
dvc.config["core"]["checksum_jobs"] = 200
remote = Remote(dvc, name="without_checksum_jobs")
assert remote.checksum_jobs == 200
tree = get_cloud_tree(dvc, name="without_checksum_jobs")
assert tree.checksum_jobs == 200
def test_remote_without_checksum_jobs_default(dvc):
dvc.config["remote"]["without_checksum_jobs"] = {"url": "s3://bucket/name"}
remote = Remote(dvc, name="without_checksum_jobs")
assert remote.checksum_jobs == remote.CHECKSUM_JOBS
tree = get_cloud_tree(dvc, name="without_checksum_jobs")
assert tree.checksum_jobs == tree.CHECKSUM_JOBS
@pytest.mark.parametrize("remote_cls", [GSRemote, S3Remote])
def test_makedirs_not_create_for_top_level_path(remote_cls, dvc, mocker):
url = f"{remote_cls.scheme}://bucket/"
remote = remote_cls(dvc, {"url": url})
@pytest.mark.parametrize("tree_cls", [GSRemoteTree, S3RemoteTree])
def test_makedirs_not_create_for_top_level_path(tree_cls, dvc, mocker):
url = f"{tree_cls.scheme}://bucket/"
tree = tree_cls(dvc, {"url": url})
mocked_client = mocker.PropertyMock()
# we use remote clients with same name as scheme to interact with remote
mocker.patch.object(remote_cls.TREE_CLS, remote.scheme, mocked_client)
mocker.patch.object(tree_cls, tree.scheme, mocked_client)
remote.tree.makedirs(remote.path_info)
tree.makedirs(tree.path_info)
assert not mocked_client.called
tests/unit/remote/test_remote_tree.py
+2 -2
View File
@@ -3,7 +3,7 @@ import os
import pytest
from dvc.path_info import PathInfo
from dvc.remote.s3 import S3Remote, S3RemoteTree
from dvc.remote.s3 import S3RemoteTree
from dvc.utils.fs import walk_files
from tests.remotes import GCP, S3Mocked
@@ -91,7 +91,7 @@ def test_copy_preserve_etag_across_buckets(remote, dvc):
s3 = remote.tree.s3
s3.create_bucket(Bucket="another")
another = S3Remote(dvc, {"url": "s3://another", "region": "us-east-1"})
another = S3RemoteTree(dvc, {"url": "s3://another", "region": "us-east-1"})
from_info = remote.path_info / "foo"
to_info = another.path_info / "foo"
tests/unit/remote/test_s3.py
+7 -8
View File
@@ -1,7 +1,7 @@
import pytest
from dvc.config import ConfigError
from dvc.remote.s3 import S3Remote
from dvc.remote.s3 import S3RemoteTree
bucket_name = "bucket-name"
prefix = "some/prefix"
@@ -20,9 +20,9 @@ def grants():
def test_init(dvc):
config = {"url": url}
remote = S3Remote(dvc, config)
tree = S3RemoteTree(dvc, config)
assert remote.tree.path_info == url
assert tree.path_info == url
def test_grants(dvc):
@@ -33,8 +33,7 @@ def test_grants(dvc):
"grant_write_acp": "id=write-acp-permission-id",
"grant_full_control": "id=full-control-permission-id",
}
remote = S3Remote(dvc, config)
tree = remote.tree
tree = S3RemoteTree(dvc, config)
assert (
tree.extra_args["GrantRead"]
@@ -52,9 +51,9 @@ def test_grants_mutually_exclusive_acl_error(dvc, grants):
config = {"url": url, "acl": "public-read", grant_option: grant_value}
with pytest.raises(ConfigError):
S3Remote(dvc, config)
S3RemoteTree(dvc, config)
def test_sse_kms_key_id(dvc):
remote = S3Remote(dvc, {"url": url, "sse_kms_key_id": "key"})
assert remote.tree.extra_args["SSEKMSKeyId"] == "key"
tree = S3RemoteTree(dvc, {"url": url, "sse_kms_key_id": "key"})
assert tree.extra_args["SSEKMSKeyId"] == "key"
tests/unit/repo/test_repo_tree.py
+3 -2
View File
@@ -22,8 +22,9 @@ def test_open(tmp_dir, dvc):
(tmp_dir / "foo").unlink()
tree = RepoTree(dvc)
with tree.open("foo", "r") as fobj:
assert fobj.read() == "foo"
with dvc.state:
with tree.open("foo", "r") as fobj:
assert fobj.read() == "foo"
def test_open_dirty_hash(tmp_dir, dvc):
tests/unit/repo/test_tree.py
+3 -2
View File
@@ -22,8 +22,9 @@ def test_open(tmp_dir, dvc):
(tmp_dir / "foo").unlink()
tree = DvcTree(dvc)
with tree.open("foo", "r") as fobj:
assert fobj.read() == "foo"
with dvc.state:
with tree.open("foo", "r") as fobj:
assert fobj.read() == "foo"
def test_open_dirty_hash(tmp_dir, dvc):