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

Merge pull request #1892 from efiop/1654

Browse Source 
support external ssh directories as dependencies and outputs
This commit is contained in:
Ruslan Kuprieiev
2019-04-24 05:57:06 +03:00
committed by GitHub
parent ee8d9a1312 37ee55b1f8
commit b8018c16a7
33 changed files with 863 additions and 747 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dvc/analytics.py
+3 -2
View File
@@ -96,11 +96,12 @@ class Analytics(object):
with open(self.user_id_file, "r") as fobj:
try:
info = json.load(fobj)
return info[self.PARAM_USER_ID]
except json.JSONDecodeError as exc:
except ValueError as exc:
logger.debug("Failed to load user_id: {}".format(exc))
return None
return info[self.PARAM_USER_ID]
def _get_user_id(self):
from dvc.lock import LockError
dvc/dependency/http.py
+2 -1
View File
@@ -16,4 +16,5 @@ class DependencyHTTP(DependencyBase, OutputBase):
if path.startswith("remote"):
path = urljoin(self.remote.cache_dir, urlparse(path).path)
self.path_info = {"scheme": urlparse(path).scheme, "path": path}
self.path_info["scheme"] = urlparse(path).scheme
self.path_info["path"] = path
dvc/output/base.py
+79 -16
View File
@@ -1,6 +1,7 @@
from __future__ import unicode_literals
import re
import logging
from schema import Or, Optional
from dvc.exceptions import DvcException
@@ -8,6 +9,9 @@ from dvc.utils.compat import str
from dvc.remote.base import RemoteBase
logger = logging.getLogger(__name__)
class OutputDoesNotExistError(DvcException):
def __init__(self, path):
msg = "output '{}' does not exist".format(path)
@@ -73,16 +77,15 @@ class OutputBase(object):
self.persist = persist
self.tags = None if self.IS_DEPENDENCY else (tags or {})
if (
self.use_cache
and getattr(self.repo.cache, self.REMOTE.scheme) is None
):
if self.use_cache and self.cache is None:
raise DvcException(
"no cache location setup for '{}' outputs.".format(
self.REMOTE.scheme
)
)
self.path_info = {"scheme": self.scheme, "url": self.url}
def __repr__(self):
return "{class_name}: '{url}'".format(
class_name=type(self).__name__, url=(self.url or "No url")
@@ -95,6 +98,14 @@ class OutputBase(object):
def is_in_repo(self):
return False
@property
def cache(self):
return getattr(self.repo.cache, self.scheme)
@property
def dir_cache(self):
return self.cache.get_dir_cache(self.checksum)
def assign_to_stage_file(self, target_repo):
raise DvcException(
"change repo is not supported for {}".format(self.scheme)
@@ -122,6 +133,10 @@ class OutputBase(object):
def path(self):
return self.path_info["path"]
@property
def cache_path(self):
return self.cache.checksum_to_path(self.checksum)
@property
def sep(self):
return "/"
@@ -130,6 +145,10 @@ class OutputBase(object):
def checksum(self):
return self.info.get(self.remote.PARAM_CHECKSUM)
@property
def is_dir_checksum(self):
return self.remote.is_dir_checksum(self.checksum)
@property
def exists(self):
return self.remote.exists(self.path_info)
@@ -146,9 +165,7 @@ class OutputBase(object):
if not self.use_cache or not self.checksum:
return True
cache = self.repo.cache.__getattribute__(self.scheme)
return cache.changed_cache(self.checksum)
return self.cache.changed_cache(self.checksum)
def status(self):
if self.checksum and self.use_cache and self.changed_cache():
@@ -166,19 +183,62 @@ class OutputBase(object):
return {}
def changed(self):
return bool(self.status())
status = self.status()
logger.debug(str(status))
return bool(status)
@property
def is_empty(self):
return self.remote.is_empty(self.path_info)
def isdir(self):
return self.remote.isdir(self.path_info)
def isfile(self):
return self.remote.isfile(self.path_info)
def save(self):
if not self.exists:
raise self.DoesNotExistError(self)
if not self.isfile and not self.isdir:
raise self.IsNotFileOrDirError(self)
if self.is_empty:
logger.warning("'{}' is empty.".format(self))
if not self.use_cache:
self.info = self.remote.save_info(self.path_info)
if self.metric:
self.verify_metric()
if not self.IS_DEPENDENCY:
logger.info(
"Output '{}' doesn't use cache. Skipping saving.".format(
self
)
)
return
assert not self.IS_DEPENDENCY
if not self.changed():
logger.info(
"Output '{}' didn't change. Skipping saving.".format(self)
)
return
if self.is_in_repo:
if self.repo.scm.is_tracked(self.path):
raise OutputAlreadyTrackedError(self)
if self.use_cache:
self.repo.scm.ignore(self.path)
self.info = self.remote.save_info(self.path_info)
def commit(self):
if self.use_cache:
getattr(self.repo.cache, self.scheme).save(
self.path_info, self.info
)
self.cache.save(self.path_info, self.info)
def dumpd(self):
ret = self.info.copy()
@@ -221,7 +281,7 @@ class OutputBase(object):
else:
info = self.info
getattr(self.repo.cache, self.scheme).checkout(
self.cache.checkout(
self.path_info,
info,
force=force,
@@ -242,7 +302,7 @@ class OutputBase(object):
self.remote.move(self.path_info, out.path_info)
self.url = out.url
self.path_info = out.path_info
self.path_info = out.path_info.copy()
self.save()
self.commit()
@@ -250,10 +310,13 @@ class OutputBase(object):
self.repo.scm.ignore(self.path)
def get_files_number(self):
if self.use_cache and self.checksum:
return 1
if not self.use_cache or not self.checksum:
return 0
return 0
if self.is_dir_checksum:
return len(self.dir_cache)
return 1
def unprotect(self):
if self.exists:
dvc/output/hdfs.py
+2 -1
View File
@@ -34,4 +34,5 @@ class OutputHDFS(OutputBase):
if remote:
path = posixpath.join(remote.url, urlparse(path).path.lstrip("/"))
user = remote.user if remote else self.group("user")
self.path_info = {"scheme": "hdfs", "user": user, "path": path}
self.path_info["user"] = user
self.path_info["path"] = path
dvc/output/local.py
+3 -62
View File
@@ -7,7 +7,7 @@ from dvc.utils.compat import urlparse
from dvc.istextfile import istextfile
from dvc.exceptions import DvcException
from dvc.remote.local import RemoteLOCAL
from dvc.output.base import OutputBase, OutputAlreadyTrackedError
from dvc.output.base import OutputBase
logger = logging.getLogger(__name__)
@@ -49,9 +49,7 @@ class OutputLOCAL(OutputBase):
p = os.path.join(stage.wdir, p)
p = os.path.abspath(os.path.normpath(p))
self.path_info = {"scheme": "local", "path": p}
self._dir_cache = {}
self.path_info["path"] = p
def __str__(self):
return self.rel_path
@@ -85,18 +83,10 @@ class OutputLOCAL(OutputBase):
def stage_path(self):
return os.path.relpath(self.path, self.stage.wdir)
@property
def cache(self):
return self.repo.cache.local.get(self.checksum)
@property
def is_dir_cache(self):
return self.repo.cache.local.is_dir_cache(self.checksum)
def dumpd(self):
ret = super(OutputLOCAL, self).dumpd()
if self.is_in_repo:
path = self.remote.unixpath(
path = self.remote.to_posixpath(
os.path.relpath(self.path, self.stage.wdir)
)
else:
@@ -120,52 +110,3 @@ class OutputLOCAL(OutputBase):
if not istextfile(self.path):
msg = "binary file '{}' cannot be used as metrics."
raise DvcException(msg.format(self.rel_path))
def save(self):
if not os.path.exists(self.path):
raise self.DoesNotExistError(self.rel_path)
if not os.path.isfile(self.path) and not os.path.isdir(self.path):
raise self.IsNotFileOrDirError(self.rel_path)
if (os.path.isfile(self.path) and os.path.getsize(self.path) == 0) or (
os.path.isdir(self.path) and len(os.listdir(self.path)) == 0
):
msg = "file/directory '{}' is empty.".format(self.rel_path)
logger.warning(msg)
if not self.use_cache:
self.info = self.remote.save_info(self.path_info)
self.verify_metric()
if not self.IS_DEPENDENCY:
msg = "Output '{}' doesn't use cache. Skipping saving."
logger.info(msg.format(self.rel_path))
return
assert not self.IS_DEPENDENCY
if not self.changed():
msg = "Output '{}' didn't change. Skipping saving."
logger.info(msg.format(self.rel_path))
return
if self.is_in_repo:
if self.repo.scm.is_tracked(self.path):
raise OutputAlreadyTrackedError(self.rel_path)
if self.use_cache:
self.repo.scm.ignore(self.path)
self.info = self.remote.save_info(self.path_info)
@property
def dir_cache(self):
return self.repo.cache.local.load_dir_cache(self.checksum)
def get_files_number(self):
if self.cache is None:
return 0
if self.is_dir_cache:
return len(self.dir_cache)
return 1
dvc/output/s3.py
+3 -5
View File
@@ -35,8 +35,6 @@ class OutputS3(OutputBase):
path = urlparse(path).path.lstrip("/")
if remote:
path = posixpath.join(remote.prefix, path)
self.path_info = {
"scheme": self.scheme,
"bucket": bucket,
"path": path,
}
self.path_info["bucket"] = bucket
self.path_info["path"] = path
dvc/output/ssh.py
+4 -7
View File
@@ -48,10 +48,7 @@ class OutputSSH(OutputBase):
else:
path = parsed.path
self.path_info = {
"scheme": "ssh",
"host": host,
"port": port,
"user": user,
"path": path,
}
self.path_info["host"] = host
self.path_info["port"] = port
self.path_info["user"] = user
self.path_info["path"] = path
dvc/remote/azure.py
+2 -2
View File
@@ -120,7 +120,7 @@ class RemoteAzure(RemoteBase):
def list_cache_paths(self):
return self._list_paths(self.bucket, self.prefix)
def upload(self, from_infos, to_infos, names=None):
def upload(self, from_infos, to_infos, names=None, no_progress_bar=False):
names = self._verify_path_args(to_infos, from_infos, names)
for from_info, to_info, name in zip(from_infos, to_infos, names):
@@ -142,7 +142,7 @@ class RemoteAzure(RemoteBase):
if not name:
name = os.path.basename(from_info["path"])
cb = Callback(name)
cb = None if no_progress_bar else Callback(name)
try:
self.blob_service.create_blob_from_path(
dvc/remote/base.py
+332 -32
View File
@@ -2,14 +2,21 @@ from __future__ import unicode_literals
from dvc.utils.compat import str
import os
import re
import posixpath
import json
import ntpath
import logging
import tempfile
import posixpath
from operator import itemgetter
from multiprocessing import cpu_count
import dvc.prompt as prompt
from dvc.config import Config
from dvc.exceptions import DvcException, ConfirmRemoveError
from dvc.utils import tmp_fname
from dvc.state import StateBase
logger = logging.getLogger(__name__)
@@ -61,6 +68,9 @@ class RemoteBase(object):
REQUIRES = {}
JOBS = 4 * cpu_count()
PARAM_RELPATH = "relpath"
CHECKSUM_DIR_SUFFIX = ".dir"
def __init__(self, repo, config):
self.repo = repo
deps_ok = all(self.REQUIRES.values())
@@ -89,6 +99,11 @@ class RemoteBase(object):
).format(url, missing, " ".join(missing), self.scheme)
raise RemoteMissingDepsError(msg)
self.protected = False
self.state = StateBase()
self._dir_info = {}
def __repr__(self):
return "{class_name}: '{url}'".format(
class_name=type(self).__name__, url=(self.url or "No url")
@@ -102,6 +117,27 @@ class RemoteBase(object):
url = config[Config.SECTION_REMOTE_URL]
return cls.match(url) is not None
@staticmethod
def _replace_sep(path, from_sep, to_sep):
assert not ntpath.isabs(path)
assert not posixpath.isabs(path)
return path.replace(from_sep, to_sep)
@classmethod
def to_posixpath(cls, path):
return cls._replace_sep(path, "\\", "/")
@classmethod
def to_ntpath(cls, path):
return cls._replace_sep(path, "/", "\\")
def to_ospath(self, path):
if self.ospath == ntpath:
return self.to_ntpath(path)
elif self.ospath == posixpath:
return self.to_posixpath(path)
assert False
@classmethod
def match(cls, url):
return re.match(cls.REGEX, url)
@@ -112,9 +148,125 @@ class RemoteBase(object):
return None
return m.group(name)
def save_info(self, path_info):
@property
def cache(self):
return getattr(self.repo.cache, self.scheme)
def get_file_checksum(self, path_info):
raise NotImplementedError
def _collect_dir(self, path_info):
dir_info = []
p_info = path_info.copy()
dpath = p_info["path"]
for root, dirs, files in self.walk(path_info):
for fname in files:
path = self.ospath.join(root, fname)
p_info["path"] = path
relpath = self.to_posixpath(self.ospath.relpath(path, dpath))
checksum = self.get_file_checksum(p_info)
dir_info.append(
{
self.PARAM_RELPATH: relpath,
self.PARAM_CHECKSUM: checksum,
}
)
# NOTE: sorting the list by path to ensure reproducibility
return sorted(dir_info, key=itemgetter(self.PARAM_RELPATH))
def get_dir_checksum(self, path_info):
dir_info = self._collect_dir(path_info)
to_info = path_info.copy()
to_info["path"] = self.cache.ospath.join(
self.cache.prefix, tmp_fname("")
)
tmp = tempfile.NamedTemporaryFile(delete=False).name
with open(tmp, "w+") as fobj:
json.dump(dir_info, fobj)
from_info = {"scheme": "local", "path": tmp}
self.cache.upload([from_info], [to_info], no_progress_bar=True)
checksum = self.get_file_checksum(to_info) + self.CHECKSUM_DIR_SUFFIX
from_info = to_info.copy()
to_info["path"] = self.cache.checksum_to_path(checksum)
if self.cache.changed_cache_file(checksum):
self.cache.move(from_info, to_info)
self.state.save(path_info, checksum)
self.state.save(to_info, checksum)
return checksum
def get_dir_cache(self, checksum):
assert checksum
dir_info = self._dir_info.get(checksum)
if dir_info:
return dir_info
dir_info = self.load_dir_cache(checksum)
self._dir_info[checksum] = dir_info
return dir_info
def load_dir_cache(self, checksum):
path_info = self.checksum_to_path_info(checksum)
fobj = tempfile.NamedTemporaryFile(delete=False)
path = fobj.name
to_info = {"scheme": "local", "path": path}
self.cache.download([path_info], [to_info], no_progress_bar=True)
try:
with open(path, "r") as fobj:
d = json.load(fobj)
except ValueError:
logger.exception("Failed to load dir cache '{}'".format(path_info))
return []
finally:
os.unlink(path)
if not isinstance(d, list):
msg = "dir cache file format error '{}' [skipping the file]"
logger.error(msg.format(os.path.relpath(path)))
return []
for info in d:
info[self.PARAM_RELPATH] = self.to_ospath(info[self.PARAM_RELPATH])
return d
@classmethod
def is_dir_checksum(cls, checksum):
return checksum.endswith(cls.CHECKSUM_DIR_SUFFIX)
def get_checksum(self, path_info):
if not self.exists(path_info):
return None
checksum = self.state.get(path_info)
if checksum:
return checksum
if self.isdir(path_info):
checksum = self.get_dir_checksum(path_info)
else:
checksum = self.get_file_checksum(path_info)
if checksum:
self.state.save(path_info, checksum)
return checksum
def save_info(self, path_info):
assert path_info["scheme"] == self.scheme
return {self.PARAM_CHECKSUM: self.get_checksum(path_info)}
def changed(self, path_info, checksum_info):
"""Checks if data has changed.
@@ -165,6 +317,61 @@ class RemoteBase(object):
logger.debug("'{}' hasn't changed.".format(path_info))
return False
def link(self, from_info, to_info):
self.copy(from_info, to_info)
def _save_file(self, path_info, checksum, save_link=True):
assert checksum
cache_info = self.checksum_to_path_info(checksum)
if self.changed_cache(checksum):
self.move(path_info, cache_info)
else:
self.remove(path_info)
self.link(cache_info, path_info)
if save_link:
self.state.save_link(path_info)
# we need to update path and cache, since in case of reflink,
# or copy cache type moving original file results in updates on
# next executed command, which causes md5 recalculation
self.state.save(path_info, checksum)
self.state.save(cache_info, checksum)
def _save_dir(self, path_info, checksum):
cache_info = self.checksum_to_path_info(checksum)
dir_info = self.get_dir_cache(checksum)
entry_info = path_info.copy()
for entry in dir_info:
entry_checksum = entry[self.PARAM_CHECKSUM]
entry_info["path"] = self.ospath.join(
path_info["path"], entry[self.PARAM_RELPATH]
)
self._save_file(entry_info, entry_checksum, save_link=False)
self.state.save_link(path_info)
self.state.save(cache_info, checksum)
self.state.save(path_info, checksum)
def is_empty(self, path_info):
return False
def isfile(self, path_info):
raise NotImplementedError
def isdir(self, path_info):
return False
def walk(self, path_info):
raise NotImplementedError
def protect(self, path_info):
pass
def save(self, path_info, checksum_info):
if path_info["scheme"] != self.scheme:
raise RemoteActionNotImplemented(
@@ -174,13 +381,18 @@ class RemoteBase(object):
checksum = checksum_info[self.PARAM_CHECKSUM]
if not self.changed_cache(checksum):
self._checkout(path_info, checksum)
return
self._save(path_info, checksum)
def _save(self, path_info, checksum):
to_info = self.checksum_to_path_info(checksum)
logger.info("Saving '{}' to '{}'.".format(path_info, to_info))
self.copy(path_info, to_info)
if self.isdir(path_info):
self._save_dir(path_info, checksum)
return
self._save_file(path_info, checksum)
def download(
self,
@@ -192,14 +404,15 @@ class RemoteBase(object):
):
raise RemoteActionNotImplemented("download", self.scheme)
def upload(self, from_infos, to_infos, names=None):
def upload(self, from_infos, to_infos, names=None, no_progress_bar=False):
raise RemoteActionNotImplemented("upload", self.scheme)
def remove(self, path_info):
raise RemoteActionNotImplemented("remove", self.scheme)
def move(self, path_info):
raise RemoteActionNotImplemented("move", self.scheme)
def move(self, from_info, to_info):
self.copy(from_info, to_info)
self.remove(from_info)
def copy(self, from_info, to_info):
raise RemoteActionNotImplemented("copy", self.scheme)
@@ -226,6 +439,7 @@ class RemoteBase(object):
return posixpath
def checksum_to_path(self, checksum):
assert checksum
return self.ospath.join(self.prefix, checksum[0:2], checksum[2:])
def path_to_checksum(self, path):
@@ -269,28 +483,51 @@ class RemoteBase(object):
removed = True
return removed
def changed_cache(self, checksum):
cache = self.checksum_to_path_info(checksum)
expected = {self.PARAM_CHECKSUM: checksum}
if not self.exists(cache):
return True
actual = self.save_info(cache)
def changed_cache_file(self, checksum):
cache_info = self.checksum_to_path_info(checksum)
actual = self.get_checksum(cache_info)
logger.debug(
"Cache '{}' actual '{}'.".format(str(expected), str(actual))
"cache '{}' expected '{}' actual '{}'".format(
str(cache_info), checksum, actual
)
)
if expected != actual:
if self.exists(cache):
msg = "corrupted cache file {}"
logger.warning(msg.format(str(cache)))
self.remove(cache)
if not checksum or not actual:
return True
if actual.split(".")[0] == checksum.split(".")[0]:
return False
if self.exists(cache_info):
logger.warning("corrupted cache file '{}'.".format(cache_info))
self.remove(cache_info)
return True
def _changed_cache_dir(self):
# NOTE: only implemented for RemoteLOCAL
return True
def _changed_dir_cache(self, checksum):
if not self._changed_cache_dir():
return False
if self.changed_cache_file(checksum):
return True
for entry in self.get_dir_cache(checksum):
checksum = entry[self.PARAM_CHECKSUM]
if self.changed_cache_file(checksum):
return True
return False
def changed_cache(self, checksum):
if self.is_dir_checksum(checksum):
return self._changed_dir_cache(checksum)
return self.changed_cache_file(checksum)
def cache_exists(self, checksums):
# NOTE: The reason for such an odd logic is that most of the remotes
# take much shorter time to just retrieve everything they have under
@@ -303,7 +540,7 @@ class RemoteBase(object):
return list(filter(lambda checksum: checksum in checksums, self.all()))
def already_cached(self, path_info):
current = self.save_info(path_info)[self.PARAM_CHECKSUM]
current = self.get_checksum(path_info)
if not current:
return False
@@ -325,16 +562,73 @@ class RemoteBase(object):
self.remove(path_info)
def do_checkout(
self, path_info, checksum, force=False, progress_callback=None
def _checkout_file(
self, path_info, checksum, force, progress_callback=None
):
cache_info = self.checksum_to_path_info(checksum)
if self.exists(path_info):
msg = "data '{}' exists. Removing before checkout."
logger.warning(msg.format(str(path_info)))
self.safe_remove(path_info, force=force)
from_info = self.checksum_to_path_info(checksum)
self.copy(from_info, path_info)
self.link(cache_info, path_info)
self.state.save_link(path_info)
if progress_callback:
progress_callback.update(path_info["url"])
def makedirs(self, path_info):
raise NotImplementedError
def _checkout_dir(
self, path_info, checksum, force, progress_callback=None
):
# Create dir separately so that dir is created
# even if there are no files in it
if not self.exists(path_info):
self.makedirs(path_info)
dir_info = self.get_dir_cache(checksum)
logger.debug("Linking directory '{}'.".format(path_info))
entry_info = path_info.copy()
for entry in dir_info:
relpath = entry[self.PARAM_RELPATH]
checksum = entry[self.PARAM_CHECKSUM]
entry_cache_info = self.checksum_to_path_info(checksum)
entry_info["url"] = self.ospath.join(path_info["url"], relpath)
entry_info["path"] = self.ospath.join(path_info["path"], relpath)
entry_checksum_info = {self.PARAM_CHECKSUM: checksum}
if self.changed(entry_info, entry_checksum_info):
if self.exists(entry_info):
self.safe_remove(entry_info, force=force)
self.link(entry_cache_info, entry_info)
if progress_callback:
progress_callback.update(entry_info["url"])
self._remove_redundant_files(path_info, dir_info, force)
self.state.save_link(path_info)
def _remove_redundant_files(self, path_info, dir_info, force):
existing_files = set(
self.ospath.join(root, fname)
for root, _, files in self.walk(path_info)
for fname in files
)
needed_files = set(
self.ospath.join(path_info["path"], entry[self.PARAM_RELPATH])
for entry in dir_info
)
delta = existing_files - needed_files
d_info = path_info.copy()
for path in delta:
d_info["path"] = path
self.safe_remove(d_info, force)
def checkout(
self, path_info, checksum_info, force=False, progress_callback=None
@@ -363,11 +657,17 @@ class RemoteBase(object):
msg = "Checking out '{}' with cache '{}'."
logger.debug(msg.format(str(path_info), checksum))
self.do_checkout(
path_info,
checksum,
force=force,
progress_callback=progress_callback,
return self._checkout(path_info, checksum, force, progress_callback)
def _checkout(
self, path_info, checksum, force=False, progress_callback=None
):
if not self.is_dir_checksum(checksum):
return self._checkout_file(
path_info, checksum, force, progress_callback=progress_callback
)
return self._checkout_dir(
path_info, checksum, force, progress_callback=progress_callback
)
@staticmethod
dvc/remote/gs.py
+6 -13
View File
@@ -55,10 +55,12 @@ class RemoteGS(RemoteBase):
else storage.Client(self.projectname)
)
def get_md5(self, bucket, path):
def get_file_checksum(self, path_info):
import base64
import codecs
bucket = path_info["bucket"]
path = path_info["path"]
blob = self.gs.bucket(bucket).get_blob(path)
if not blob:
return None
@@ -67,16 +69,6 @@ class RemoteGS(RemoteBase):
md5 = base64.b64decode(b64_md5)
return codecs.getencoder("hex")(md5)[0].decode("utf-8")
def save_info(self, path_info):
if path_info["scheme"] != "gs":
raise NotImplementedError
return {
self.PARAM_CHECKSUM: self.get_md5(
path_info["bucket"], path_info["path"]
)
}
def copy(self, from_info, to_info, gs=None):
gs = gs if gs else self.gs
@@ -120,7 +112,7 @@ class RemoteGS(RemoteBase):
paths = self._list_paths(path_info["bucket"], path_info["path"])
return any(path_info["path"] == path for path in paths)
def upload(self, from_infos, to_infos, names=None):
def upload(self, from_infos, to_infos, names=None, no_progress_bar=False):
names = self._verify_path_args(to_infos, from_infos, names)
gs = self.gs
@@ -141,7 +133,8 @@ class RemoteGS(RemoteBase):
if not name:
name = os.path.basename(from_info["path"])
progress.update_target(name, 0, None)
if not no_progress_bar:
progress.update_target(name, 0, None)
try:
bucket = gs.bucket(to_info["bucket"])
dvc/remote/hdfs.py
+2 -10
View File
@@ -63,7 +63,7 @@ class RemoteHDFS(RemoteBase):
assert match is not None
return match.group(gname)
def checksum(self, path_info):
def get_file_checksum(self, path_info):
regex = r".*\t.*\t(?P<checksum>.*)"
stdout = self.hadoop_fs(
"checksum {}".format(path_info["path"]), user=path_info["user"]
@@ -83,14 +83,6 @@ class RemoteHDFS(RemoteBase):
"rm -f {}".format(path_info["path"]), user=path_info["user"]
)
def save_info(self, path_info):
if path_info["scheme"] != "hdfs":
raise NotImplementedError
assert path_info.get("path")
return {self.PARAM_CHECKSUM: self.checksum(path_info)}
def remove(self, path_info):
if path_info["scheme"] != "hdfs":
raise NotImplementedError
@@ -111,7 +103,7 @@ class RemoteHDFS(RemoteBase):
except RemoteCmdError:
return False
def upload(self, from_infos, to_infos, names=None):
def upload(self, from_infos, to_infos, names=None, no_progress_bar=False):
names = self._verify_path_args(to_infos, from_infos, names)
for from_info, to_info, name in zip(from_infos, to_infos, names):
dvc/remote/http.py
+2 -7
View File
@@ -109,16 +109,11 @@ class RemoteHTTP(RemoteBase):
return list(filter(func, md5s))
def save_info(self, path_info):
if path_info["scheme"] not in ["http", "https"]:
raise NotImplementedError
return {self.PARAM_CHECKSUM: self._etag(path_info["path"])}
def _content_length(self, url):
return self._request("HEAD", url).headers.get("Content-Length")
def _etag(self, url):
def get_file_checksum(self, path_info):
url = path_info["path"]
etag = self._request("HEAD", url).headers.get("ETag") or self._request(
"HEAD", url
).headers.get("Content-MD5")
dvc/remote/local.py
+52 -300
View File
@@ -7,12 +7,10 @@ from dvc.utils.compat import str, makedirs
import os
import stat
import uuid
import json
import ntpath
import shutil
import posixpath
import logging
from operator import itemgetter
from dvc.system import System
from dvc.remote.base import (
@@ -26,12 +24,11 @@ from dvc.utils import (
remove,
move,
copyfile,
dict_md5,
to_chunks,
tmp_fname,
file_md5,
walk_files,
)
from dvc.utils import LARGE_DIR_SIZE
from dvc.config import Config
from dvc.exceptions import DvcException
from dvc.progress import progress
@@ -48,8 +45,6 @@ class RemoteLOCAL(RemoteBase):
REGEX = r"^(?P<path>.*)$"
PARAM_CHECKSUM = "md5"
PARAM_PATH = "path"
PARAM_RELPATH = "relpath"
MD5_DIR_SUFFIX = ".dir"
DEFAULT_CACHE_TYPES = ["reflink", "copy"]
CACHE_TYPE_MAP = {
@@ -116,44 +111,21 @@ class RemoteLOCAL(RemoteBase):
if not md5:
return None
return os.path.join(self.cache_dir, md5[0:2], md5[2:])
def changed_cache_file(self, md5):
cache = self.get(md5)
if self.state.changed(cache, md5=md5):
if os.path.exists(cache):
msg = "Corrupted cache file {}."
logger.warning(msg.format(os.path.relpath(cache)))
remove(cache)
return True
return False
return self.checksum_to_path(md5)
def exists(self, path_info):
assert not isinstance(path_info, list)
assert path_info["scheme"] == "local"
return os.path.lexists(path_info["path"])
def changed_cache(self, md5):
cache = self.get(md5)
clist = [(cache, md5)]
def makedirs(self, path_info):
if not self.exists(path_info):
os.makedirs(path_info["path"])
while True:
if len(clist) == 0:
break
def link(self, cache_info, path_info):
cache = cache_info["path"]
path = path_info["path"]
cache, md5 = clist.pop()
if self.changed_cache_file(md5):
return True
if self.is_dir_cache(cache) and self._cache_metadata_changed():
for entry in self.load_dir_cache(md5):
md5 = entry[self.PARAM_CHECKSUM]
cache = self.get(md5)
clist.append((cache, md5))
return False
def link(self, cache, path):
assert os.path.isfile(cache)
dname = os.path.dirname(path)
@@ -174,7 +146,7 @@ class RemoteLOCAL(RemoteBase):
self.CACHE_TYPE_MAP[self.cache_types[0]](cache, path)
if self.protected:
os.chmod(path, stat.S_IREAD | stat.S_IRGRP | stat.S_IROTH)
self.protect(path_info)
msg = "Created {}'{}': {} -> {}".format(
"protected " if self.protected else "",
@@ -200,276 +172,38 @@ class RemoteLOCAL(RemoteBase):
return ntpath
return posixpath
@classmethod
def to_ospath(cls, path):
if os.name == "nt":
return cls.ntpath(path)
return cls.unixpath(path)
@staticmethod
def unixpath(path):
assert not ntpath.isabs(path)
assert not posixpath.isabs(path)
return path.replace("\\", "/")
@staticmethod
def ntpath(path):
assert not ntpath.isabs(path)
assert not posixpath.isabs(path)
return path.replace("/", "\\")
def collect_dir_cache(self, dname):
dir_info = []
for root, dirs, files in os.walk(str(dname)):
bar = False
if len(files) > LARGE_DIR_SIZE:
msg = (
"Computing md5 for a large directory {}. "
"This is only done once."
)
logger.info(msg.format(os.path.relpath(root)))
bar = True
title = os.path.relpath(root)
processed = 0
total = len(files)
progress.update_target(title, 0, total)
for fname in files:
path = os.path.join(root, fname)
relpath = self.unixpath(os.path.relpath(path, dname))
if bar:
progress.update_target(title, processed, total)
processed += 1
md5 = self.state.update(path)
dir_info.append(
{self.PARAM_RELPATH: relpath, self.PARAM_CHECKSUM: md5}
)
if bar:
progress.finish_target(title)
# NOTE: sorting the list by path to ensure reproducibility
dir_info = sorted(dir_info, key=itemgetter(self.PARAM_RELPATH))
md5 = dict_md5(dir_info) + self.MD5_DIR_SUFFIX
if self.changed_cache_file(md5):
self.dump_dir_cache(md5, dir_info)
return (md5, dir_info)
def load_dir_cache(self, md5):
path = self.get(md5)
assert self.is_dir_cache(path)
dir_info = self._dir_info.get(md5)
if dir_info:
return dir_info
try:
with open(path, "r") as fd:
d = json.load(fd)
except Exception:
msg = "Failed to load dir cache '{}'"
logger.exception(msg.format(os.path.relpath(path)))
return []
if not isinstance(d, list):
msg = "dir cache file format error '{}' [skipping the file]"
logger.error(msg.format(os.path.relpath(path)))
return []
for info in d:
info["relpath"] = self.to_ospath(info["relpath"])
self._dir_info[md5] = d
return d
def dump_dir_cache(self, md5, dir_info):
path = self.get(md5)
dname = os.path.dirname(path)
assert self.is_dir_cache(path)
assert isinstance(dir_info, list)
if not os.path.isdir(dname):
os.makedirs(dname)
# NOTE: Writing first and renaming after that
# to make sure that the operation is atomic.
tmp = "{}.{}".format(path, str(uuid.uuid4()))
with open(tmp, "w+") as fd:
json.dump(dir_info, fd, sort_keys=True)
move(tmp, path)
@classmethod
def is_dir_cache(cls, cache):
return cache.endswith(cls.MD5_DIR_SUFFIX)
def do_checkout(
self, path_info, checksum, force=False, progress_callback=None
):
path = path_info["path"]
cache = self.get(checksum)
if not self.is_dir_cache(cache):
if self.exists(path_info):
self.safe_remove(path_info, force=force)
self.link(cache, path)
self.state.update_link(path)
if progress_callback:
progress_callback.update(os.path.relpath(path))
return
# Create dir separately so that dir is created
# even if there are no files in it
if not os.path.exists(path):
os.makedirs(path)
dir_relpath = os.path.relpath(path)
logger.debug("Linking directory '{}'.".format(dir_relpath))
dir_info = self.load_dir_cache(checksum)
for processed, entry in enumerate(dir_info):
relpath = entry[self.PARAM_RELPATH]
m = entry[self.PARAM_CHECKSUM]
p = os.path.join(path, relpath)
c = self.get(m)
entry_info = {"scheme": path_info["scheme"], self.PARAM_PATH: p}
entry_checksum_info = {self.PARAM_CHECKSUM: m}
if self.changed(entry_info, entry_checksum_info):
if self.exists(entry_info):
self.safe_remove(entry_info, force=force)
self.link(c, p)
if progress_callback:
progress_callback.update(os.path.relpath(p))
self._discard_working_directory_changes(path, dir_info, force=force)
self.state.update_link(path)
def already_cached(self, path_info):
assert path_info["scheme"] in ["", "local"]
current_md5 = self.state.update(path_info["path"])
current_md5 = self.get_checksum(path_info)
if not current_md5:
return False
return not self.changed_cache(current_md5)
def _discard_working_directory_changes(self, path, dir_info, force=False):
working_dir_files = set(path for path in walk_files(path))
cached_files = set(
os.path.join(path, file["relpath"]) for file in dir_info
)
delta = working_dir_files - cached_files
for file in delta:
self.safe_remove({"scheme": "local", "path": file}, force=force)
def _move(self, inp, outp):
# moving in two stages to make the whole operation atomic in
# case inp and outp are in different filesystems and actual
# physical copying of data is happening
tmp = "{}.{}".format(outp, str(uuid.uuid4()))
move(inp, tmp)
move(tmp, outp)
def _save_file(self, path, md5):
assert md5 is not None
cache = self.get(md5)
if self.changed_cache(md5):
self._move(path, cache)
else:
remove(path)
self.link(cache, path)
self.state.update_link(path)
# we need to update path and cache, since in case of reflink,
# or copy cache type moving original file results in updates on
# next executed command, which causes md5 recalculation
self.state.update(path, md5)
self.state.update(cache, md5)
return {self.PARAM_CHECKSUM: md5}
def _save_dir(self, path, md5):
dir_info = self.load_dir_cache(md5)
dir_relpath = os.path.relpath(path)
dir_size = len(dir_info)
bar = dir_size > LARGE_DIR_SIZE
logger.info("Linking directory '{}'.".format(dir_relpath))
for processed, entry in enumerate(dir_info):
relpath = entry[self.PARAM_RELPATH]
m = entry[self.PARAM_CHECKSUM]
p = os.path.join(path, relpath)
c = self.get(m)
if self.changed_cache(m):
self._move(p, c)
else:
remove(p)
self.link(c, p)
self.state.update(p, m)
self.state.update(c, m)
if bar:
progress.update_target(dir_relpath, processed, dir_size)
self.state.update_link(path)
cache = self.get(md5)
self.state.update(cache)
self.state.update(path, md5)
if bar:
progress.finish_target(dir_relpath)
def save(self, path_info, checksum_info):
if path_info["scheme"] != "local":
raise NotImplementedError
def is_empty(self, path_info):
path = path_info["path"]
msg = "Saving '{}' to cache '{}'."
logger.info(
msg.format(os.path.relpath(path), os.path.relpath(self.cache_dir))
)
if self.isfile(path_info) and os.path.getsize(path) == 0:
return True
md5 = checksum_info[self.PARAM_CHECKSUM]
if os.path.isdir(path):
self._save_dir(path, md5)
else:
self._save_file(path, md5)
if self.isdir(path_info) and len(os.listdir(path)) == 0:
return True
def save_info(self, path_info):
if path_info["scheme"] != "local":
raise NotImplementedError
return False
return {self.PARAM_CHECKSUM: self.state.update(path_info["path"])}
def isfile(self, path_info):
return os.path.isfile(path_info["path"])
def isdir(self, path_info):
return os.path.isdir(path_info["path"])
def walk(self, path_info):
return os.walk(path_info["path"])
def get_file_checksum(self, path_info):
return file_md5(path_info["path"])[0]
def remove(self, path_info):
if path_info["scheme"] != "local":
@@ -481,13 +215,21 @@ class RemoteLOCAL(RemoteBase):
if from_info["scheme"] != "local" or to_info["scheme"] != "local":
raise NotImplementedError
move(from_info["path"], to_info["path"])
inp = from_info["path"]
outp = to_info["path"]
# moving in two stages to make the whole operation atomic in
# case inp and outp are in different filesystems and actual
# physical copying of data is happening
tmp = "{}.{}".format(outp, str(uuid.uuid4()))
move(inp, tmp)
move(tmp, outp)
def cache_exists(self, md5s):
assert isinstance(md5s, list)
return list(filter(lambda md5: not self.changed_cache_file(md5), md5s))
def upload(self, from_infos, to_infos, names=None):
def upload(self, from_infos, to_infos, names=None, no_progress_bar=False):
names = self._verify_path_args(to_infos, from_infos, names)
for from_info, to_info, name in zip(from_infos, to_infos, names):
@@ -510,7 +252,12 @@ class RemoteLOCAL(RemoteBase):
tmp_file = tmp_fname(to_info["path"])
try:
copyfile(from_info["path"], tmp_file, name=name)
copyfile(
from_info["path"],
tmp_file,
name=name,
no_progress_bar=no_progress_bar,
)
os.rename(tmp_file, to_info["path"])
except Exception:
logger.exception(
@@ -724,17 +471,18 @@ class RemoteLOCAL(RemoteBase):
download=True,
)
def _cache_metadata_changed(self):
def _changed_cache_dir(self):
mtime, size = get_mtime_and_size(self.cache_dir)
inode = get_inode(self.cache_dir)
existing_record = self.state.get_state_record_for_inode(inode)
if existing_record:
cached_mtime, cached_size, _, _ = existing_record
return not (mtime == cached_mtime and size == cached_size)
changed = not (mtime == cached_mtime and size == cached_size)
else:
changed = True
return True
return changed
def _log_missing_caches(self, checksum_info_dict):
missing_caches = [
@@ -799,3 +547,7 @@ class RemoteLOCAL(RemoteBase):
RemoteLOCAL._unprotect_dir(path)
else:
RemoteLOCAL._unprotect_file(path)
@staticmethod
def protect(path_info):
os.chmod(path_info["path"], stat.S_IREAD | stat.S_IRGRP | stat.S_IROTH)
dvc/remote/s3.py
+6 -12
View File
@@ -94,8 +94,12 @@ class RemoteS3(RemoteBase):
return obj["ETag"].strip('"')
def get_file_checksum(self, path_info):
return self.get_etag(self.s3, path_info["bucket"], path_info["path"])
@staticmethod
def get_head_object(s3, bucket, path, *args, **kwargs):
try:
obj = s3.head_object(Bucket=bucket, Key=path, *args, **kwargs)
except Exception as exc:
@@ -104,16 +108,6 @@ class RemoteS3(RemoteBase):
)
return obj
def save_info(self, path_info):
if path_info["scheme"] != "s3":
raise NotImplementedError
return {
self.PARAM_CHECKSUM: self.get_etag(
self.s3, path_info["bucket"], path_info["path"]
)
}
@classmethod
def _copy_multipart(cls, s3, from_info, to_info, size, n_parts):
mpu = s3.create_multipart_upload(
@@ -241,7 +235,7 @@ class RemoteS3(RemoteBase):
paths = self._list_paths(path_info["bucket"], path_info["path"])
return any(path_info["path"] == path for path in paths)
def upload(self, from_infos, to_infos, names=None):
def upload(self, from_infos, to_infos, names=None, no_progress_bar=False):
names = self._verify_path_args(to_infos, from_infos, names)
s3 = self.s3
@@ -263,7 +257,7 @@ class RemoteS3(RemoteBase):
name = os.path.basename(from_info["path"])
total = os.path.getsize(from_info["path"])
cb = Callback(name, total)
cb = None if no_progress_bar else Callback(name, total)
try:
s3.upload_file(
dvc/remote/ssh/__init__.py
+26 -9
View File
@@ -93,15 +93,19 @@ class RemoteSSH(RemoteBase):
assert path_info["scheme"] == self.scheme
with self.ssh(**path_info) as ssh:
return ssh.file_exists(path_info["path"])
return ssh.exists(path_info["path"])
def md5(self, path_info):
def get_file_checksum(self, path_info):
if path_info["scheme"] != self.scheme:
raise NotImplementedError
with self.ssh(**path_info) as ssh:
return ssh.md5(path_info["path"])
def isdir(self, path_info):
with self.ssh(**path_info) as ssh:
return ssh.isdir(path_info["path"])
def copy(self, from_info, to_info):
if (
from_info["scheme"] != self.scheme
@@ -112,12 +116,6 @@ class RemoteSSH(RemoteBase):
with self.ssh(**from_info) as ssh:
ssh.cp(from_info["path"], to_info["path"])
def save_info(self, path_info):
if path_info["scheme"] != self.scheme:
raise NotImplementedError
return {self.PARAM_CHECKSUM: self.md5(path_info)}
def remove(self, path_info):
if path_info["scheme"] != self.scheme:
raise NotImplementedError
@@ -125,6 +123,16 @@ class RemoteSSH(RemoteBase):
with self.ssh(**path_info) as ssh:
ssh.remove(path_info["path"])
def move(self, from_info, to_info):
if (
from_info["scheme"] != self.scheme
or to_info["scheme"] != self.scheme
):
raise NotImplementedError
with self.ssh(**from_info) as ssh:
ssh.move(from_info["path"], to_info["path"])
def download(
self,
from_infos,
@@ -171,7 +179,7 @@ class RemoteSSH(RemoteBase):
ssh.close()
def upload(self, from_infos, to_infos, names=None):
def upload(self, from_infos, to_infos, names=None, no_progress_bar=False):
names = self._verify_path_args(to_infos, from_infos, names)
with self.ssh(**to_infos[0]) as ssh:
@@ -199,3 +207,12 @@ class RemoteSSH(RemoteBase):
def list_cache_paths(self):
with self.ssh(**self.path_info) as ssh:
return list(ssh.walk_files(self.prefix))
def walk(self, path_info):
with self.ssh(**path_info) as ssh:
for entry in ssh.walk(path_info["path"]):
yield entry
def makedirs(self, path_info):
with self.ssh(**path_info) as ssh:
ssh.makedirs(path_info["path"])
dvc/remote/ssh/connection.py
+58 -16
View File
@@ -2,6 +2,7 @@ import errno
import os
import posixpath
import logging
from stat import S_ISDIR
try:
import paramiko
@@ -72,7 +73,7 @@ class SSHConnection:
self._ssh.close()
def file_exists(self, path):
def exists(self, path):
self._sftp_connect()
try:
return self._sftp.stat(path)
@@ -118,7 +119,7 @@ class SSHConnection:
"a file with the same name '{}' already exists".format(path)
)
head, tail = os.path.split(path)
head, tail = posixpath.split(path)
if head:
self.makedirs(head)
@@ -126,8 +127,11 @@ class SSHConnection:
if tail:
self._sftp.mkdir(path)
def walk_files(self, directory):
from stat import S_ISLNK, S_ISDIR, S_ISREG
def walk(self, directory, topdown=True):
# NOTE: original os.walk() implementation [1] with default options was
# used as a template.
#
# [1] https://github.com/python/cpython/blob/master/Lib/os.py
self._sftp_connect()
@@ -141,28 +145,61 @@ class SSHConnection:
cause=exc,
)
dirs = []
nondirs = []
for entry in dir_entries:
path = posixpath.join(directory, entry.filename)
if S_ISLNK(entry.st_mode):
path = self._sftp.readlink(directory)
entry = self._sftp.stat(path)
name = entry.filename
if S_ISDIR(entry.st_mode):
for inner_path in self.walk_files(path):
yield inner_path
dirs.append(name)
else:
nondirs.append(name)
elif S_ISREG(entry.st_mode):
yield path
if topdown:
yield directory, dirs, nondirs
def remove(self, path):
self._sftp_connect()
for dname in dirs:
newpath = posixpath.join(directory, dname)
for entry in self.walk(newpath, topdown=topdown):
yield entry
if not topdown:
yield directory, dirs, nondirs
def walk_files(self, directory):
for root, dirs, files in self.walk(directory):
for fname in files:
yield posixpath.join(root, fname)
def _remove_file(self, path):
try:
self._sftp.remove(path)
except IOError as exc:
if exc.errno != errno.ENOENT:
raise
def _remove_dir(self, path):
for root, dirs, files in self.walk(path, topdown=False):
for fname in files:
path = posixpath.join(root, fname)
self._remove_file(path)
for dname in dirs:
path = posixpath.join(root, dname)
self._sftp.rmdir(dname)
try:
self._sftp.rmdir(path)
except IOError as exc:
if exc.errno != errno.ENOENT:
raise
def remove(self, path):
self._sftp_connect()
if self.isdir(path):
self._remove_dir(path)
else:
self._remove_file(path)
def download(self, src, dest, no_progress_bar=False, progress_title=None):
self._sftp_connect()
@@ -183,6 +220,11 @@ class SSHConnection:
os.rename(tmp_file, dest)
def move(self, src, dst):
self.makedirs(posixpath.dirname(dst))
self._sftp_connect()
self._sftp.rename(src, dst)
def upload(self, src, dest, no_progress_bar=False, progress_title=None):
self._sftp_connect()
dvc/repo/__init__.py
+2 -4
View File
@@ -205,7 +205,7 @@ class Repo(object):
else:
return ret
for i in self.cache.local.load_dir_cache(md5):
for i in out.dir_cache:
i["branch"] = branch
i[r.PARAM_PATH] = os.path.join(
info[r.PARAM_PATH], i[r.PARAM_RELPATH]
@@ -234,9 +234,7 @@ class Repo(object):
if out.scheme != "local":
return ret
md5 = info[out.remote.PARAM_CHECKSUM]
cache = self.cache.local.get(md5)
if not out.remote.is_dir_cache(cache):
if not out.is_dir_checksum:
return ret
return self._collect_dir_cache(
dvc/repo/diff.py
+2 -2
View File
@@ -133,9 +133,9 @@ def _check_local_cache(a_out, is_checked):
def _is_dir(path, a_outs, b_outs):
if a_outs.get(path):
return a_outs[path].is_dir_cache
return a_outs[path].is_dir_checksum
else:
return b_outs[path].is_dir_cache
return b_outs[path].is_dir_checksum
def _get_diff_outs(self, diff_dct):
dvc/repo/move.py
+2 -2
View File
@@ -5,7 +5,7 @@ from dvc.repo.scm_context import scm_context
def _expand_target_path(from_path, to_path):
if os.path.isdir(to_path) and not os.path.isdir(from_path):
if os.path.isdir(to_path):
return os.path.join(to_path, os.path.basename(from_path))
return to_path
@@ -59,7 +59,7 @@ def move(self, from_path, to_path):
)
to_out = Output.loads_from(
stage, [os.path.basename(to_path)], out.cache, out.metric
stage, [os.path.basename(to_path)], out.use_cache, out.metric
)[0]
with self.state:
dvc/stage.py
+2 -2
View File
@@ -601,14 +601,14 @@ class Stage(object):
return stage
def dumpd(self):
from dvc.remote.local import RemoteLOCAL
from dvc.remote.base import RemoteBase
return {
key: value
for key, value in {
Stage.PARAM_MD5: self.md5,
Stage.PARAM_CMD: self.cmd,
Stage.PARAM_WDIR: RemoteLOCAL.unixpath(
Stage.PARAM_WDIR: RemoteBase.to_posixpath(
os.path.relpath(self.wdir, os.path.dirname(self.path))
),
Stage.PARAM_LOCKED: self.locked,
dvc/state.py
+73 -97
View File
@@ -7,7 +7,7 @@ import sqlite3
import logging
from dvc.config import Config
from dvc.utils import file_md5, remove, current_timestamp
from dvc.utils import remove, current_timestamp
from dvc.exceptions import DvcException
from dvc.utils.fs import get_mtime_and_size, get_inode
@@ -29,11 +29,18 @@ class StateVersionTooNewError(DvcException):
)
def _file_metadata_changed(actual_mtime, mtime, actual_size, size):
return actual_mtime != mtime or actual_size != size
class StateBase(object):
def save(self, path_info, checksum):
pass
def get(self, path_info):
return None
def save_link(self, path_info):
pass
class State(object): # pylint: disable=too-many-instance-attributes
class State(StateBase): # pylint: disable=too-many-instance-attributes
"""Class for the state database.
Args:
@@ -112,11 +119,6 @@ class State(object): # pylint: disable=too-many-instance-attributes
def __exit__(self, typ, value, tbck):
self.dump()
def _collect(self, path):
if os.path.isdir(path):
return self.repo.cache.local.collect_dir_cache(path)
return (file_md5(path)[0], None)
def changed(self, path, md5):
"""Check if file/directory has the expected md5.
@@ -299,53 +301,9 @@ class State(object): # pylint: disable=too-many-instance-attributes
self.cursor = None
self.inserts = 0
def _do_update(self, path, known_checksum=None):
"""
Make sure the stored info for the given path is up to date.
"""
if not os.path.exists(path):
return None, None
actual_mtime, actual_size = get_mtime_and_size(path)
actual_inode = get_inode(path)
existing_record = self.get_state_record_for_inode(actual_inode)
if existing_record:
md5, info = self._update_existing_state_record(
path,
actual_inode,
actual_mtime,
actual_size,
existing_record,
known_checksum,
)
else:
md5, info = self._insert_new_state_record(
path, actual_inode, actual_mtime, actual_size, known_checksum
)
return md5, info
def _update_existing_state_record(
self,
path,
actual_inode,
actual_mtime,
actual_size,
existing_record,
known_checksum=None,
):
mtime, size, md5, _ = existing_record
if _file_metadata_changed(actual_mtime, mtime, actual_size, size):
md5, info = self._update_state_for_path_changed(
path, actual_inode, actual_mtime, actual_size, known_checksum
)
else:
info = None
self._update_state_record_timestamp_for_inode(actual_inode)
return md5, info
@staticmethod
def _file_metadata_changed(actual_mtime, mtime, actual_size, size):
return actual_mtime != mtime or actual_size != size
def _update_state_record_timestamp_for_inode(self, actual_inode):
cmd = 'UPDATE {} SET timestamp = "{}" WHERE inode = {}'
@@ -358,17 +316,8 @@ class State(object): # pylint: disable=too-many-instance-attributes
)
def _update_state_for_path_changed(
self,
path,
actual_inode,
actual_mtime,
actual_size,
known_checksum=None,
self, path, actual_inode, actual_mtime, actual_size, checksum
):
if known_checksum:
md5, info = known_checksum, None
else:
md5, info = self._collect(path)
cmd = (
"UPDATE {} SET "
'mtime = "{}", size = "{}", '
@@ -380,20 +329,17 @@ class State(object): # pylint: disable=too-many-instance-attributes
self.STATE_TABLE,
actual_mtime,
actual_size,
md5,
checksum,
current_timestamp(),
self._to_sqlite(actual_inode),
)
)
return md5, info
def _insert_new_state_record(
self, path, actual_inode, actual_mtime, actual_size, known_checksum
self, path, actual_inode, actual_mtime, actual_size, checksum
):
if known_checksum:
md5, info = known_checksum, None
else:
md5, info = self._collect(path)
assert checksum is not None
cmd = (
"INSERT INTO {}(inode, mtime, size, md5, timestamp) "
'VALUES ({}, "{}", "{}", "{}", "{}")'
@@ -404,12 +350,11 @@ class State(object): # pylint: disable=too-many-instance-attributes
self._to_sqlite(actual_inode),
actual_mtime,
actual_size,
md5,
checksum,
current_timestamp(),
)
)
self.inserts += 1
return md5, info
def get_state_record_for_inode(self, inode):
cmd = "SELECT mtime, size, md5, timestamp from {} " "WHERE inode={}"
@@ -422,43 +367,74 @@ class State(object): # pylint: disable=too-many-instance-attributes
return results[0]
return None
def update(self, path, known_checksum=None):
"""Gets the checksum for the specified path. Checksum will be
retrieved from the state database if available, otherwise it will be
computed and cached in the state database for the further use.
def save(self, path_info, checksum):
"""Save checksum for the specified path info.
Args:
path (str): path to get the checksum for.
Returns:
str: checksum for the specified path.
path_info (dict): path_info to save checksum for.
checksum (str): checksum to save.
"""
return self._do_update(path, known_checksum)[0]
assert path_info["scheme"] == "local"
assert checksum is not None
def update_info(self, path):
"""Gets the checksum and the directory info (if applicable) for the
specified path.
path = path_info["path"]
assert os.path.exists(path)
actual_mtime, actual_size = get_mtime_and_size(path)
actual_inode = get_inode(path)
existing_record = self.get_state_record_for_inode(actual_inode)
if not existing_record:
self._insert_new_state_record(
path, actual_inode, actual_mtime, actual_size, checksum
)
return
self._update_state_for_path_changed(
path, actual_inode, actual_mtime, actual_size, checksum
)
def get(self, path_info):
"""Gets the checksum for the specified path info. Checksum will be
retrieved from the state database if available.
Args:
path (str): path to get the checksum and the directory info for.
path_info (dict): path info to get the checksum for.
Returns:
tuple: checksum for the specified path along with a directory info
(list of {relative_path: checksum} entries for each file in the
directory) if applicable, otherwise None.
str or None: checksum for the specified path info or None if it
doesn't exist in the state database.
"""
md5, info = self._do_update(path)
if not info:
info = self.repo.cache.local.load_dir_cache(md5)
return (md5, info)
assert path_info["scheme"] == "local"
path = path_info["path"]
def update_link(self, path):
if not os.path.exists(path):
return None
actual_mtime, actual_size = get_mtime_and_size(path)
actual_inode = get_inode(path)
existing_record = self.get_state_record_for_inode(actual_inode)
if not existing_record:
return None
mtime, size, checksum, _ = existing_record
if self._file_metadata_changed(actual_mtime, mtime, actual_size, size):
return None
self._update_state_record_timestamp_for_inode(actual_inode)
return checksum
def save_link(self, path_info):
"""Adds the specified path to the list of links created by dvc. This
list is later used on `dvc checkout` to cleanup old links.
Args:
path (str): path to add to the list of links.
path_info (dict): path info to add to the list of links.
"""
assert path_info["scheme"] == "local"
path = path_info["path"]
if not os.path.exists(path):
return
dvc/utils/compat.py
-1
View File
@@ -5,7 +5,6 @@ import sys
import os
import errno
# Syntax sugar.
_ver = sys.version_info
tests/func/test_add.py
+8 -22
View File
@@ -224,9 +224,8 @@ class TestDoubleAddUnchanged(TestDvc):
class TestShouldUpdateStateEntryForFileAfterAdd(TestDvc):
def test(self):
file_md5_counter = spy(dvc.state.file_md5)
with patch.object(dvc.state, "file_md5", file_md5_counter):
file_md5_counter = spy(dvc.remote.local.file_md5)
with patch.object(dvc.remote.local, "file_md5", file_md5_counter):
ret = main(["config", "cache.type", "copy"])
self.assertEqual(ret, 0)
@@ -245,8 +244,8 @@ class TestShouldUpdateStateEntryForFileAfterAdd(TestDvc):
class TestShouldUpdateStateEntryForDirectoryAfterAdd(TestDvc):
def test(self):
file_md5_counter = spy(dvc.state.file_md5)
with patch.object(dvc.state, "file_md5", file_md5_counter):
file_md5_counter = spy(dvc.remote.local.file_md5)
with patch.object(dvc.remote.local, "file_md5", file_md5_counter):
ret = main(["config", "cache.type", "copy"])
self.assertEqual(ret, 0)
@@ -303,27 +302,14 @@ class TestShouldNotCheckCacheForDirIfCacheMetadataDidNotChange(TestDvc):
class TestShouldCollectDirCacheOnlyOnce(TestDvc):
NEW_LARGE_DIR_SIZE = 1
@patch("dvc.remote.local.LARGE_DIR_SIZE", NEW_LARGE_DIR_SIZE)
def test(self):
from dvc.remote.local import RemoteLOCAL
collect_dir_counter = spy(RemoteLOCAL.collect_dir_cache)
get_dir_checksum_counter = spy(RemoteLOCAL.get_dir_checksum)
with patch.object(
RemoteLOCAL, "collect_dir_cache", collect_dir_counter
RemoteLOCAL, "get_dir_checksum", get_dir_checksum_counter
):
LARGE_DIR_FILES_NUM = self.NEW_LARGE_DIR_SIZE + 1
data_dir = "dir"
os.makedirs(data_dir)
for i in range(LARGE_DIR_FILES_NUM):
with open(os.path.join(data_dir, str(i)), "w+") as f:
f.write(str(i))
ret = main(["add", data_dir])
ret = main(["add", self.DATA_DIR])
self.assertEqual(0, ret)
ret = main(["status"])
@@ -331,7 +317,7 @@ class TestShouldCollectDirCacheOnlyOnce(TestDvc):
ret = main(["status"])
self.assertEqual(0, ret)
self.assertEqual(1, collect_dir_counter.mock.call_count)
self.assertEqual(1, get_dir_checksum_counter.mock.call_count)
class SymlinkAddTestBase(TestDvc):
tests/func/test_checkout.py
+1 -1
View File
@@ -69,7 +69,7 @@ class TestCheckoutSingleStage(TestCheckout):
class TestCheckoutCorruptedCacheFile(TestRepro):
def test(self):
cache = self.foo_stage.outs[0].cache
cache = self.foo_stage.outs[0].cache_path
with open(cache, "a") as fd:
fd.write("1")
tests/func/test_data_cloud.py
+9 -8
View File
@@ -9,7 +9,6 @@ import platform
import copy
import logging
from dvc.state import State
from mock import patch
from dvc.utils.compat import str
@@ -244,7 +243,7 @@ class TestDataCloudBase(TestDvc):
self.assertEqual(len(stages), 1)
stage = stages[0]
self.assertTrue(stage is not None)
cache = stage.outs[0].cache
cache = stage.outs[0].cache_path
info = stage.outs[0].dumpd()
md5 = info["md5"]
@@ -253,7 +252,7 @@ class TestDataCloudBase(TestDvc):
stage_dir = stages[0]
self.assertTrue(stage_dir is not None)
cache_dir = stage_dir.outs[0].cache
cache_dir = stage_dir.outs[0].cache_path
info_dir = stage_dir.outs[0].dumpd()
md5_dir = info_dir["md5"]
@@ -423,13 +422,13 @@ class TestDataCloudCLIBase(TestDvc):
self.assertEqual(len(stages), 1)
stage = stages[0]
self.assertTrue(stage is not None)
cache = stage.outs[0].cache
cache = stage.outs[0].cache_path
stages = self.dvc.add(self.DATA_DIR)
self.assertEqual(len(stages), 1)
stage_dir = stages[0]
self.assertTrue(stage_dir is not None)
cache_dir = stage_dir.outs[0].cache
cache_dir = stage_dir.outs[0].cache_path
# FIXME check status output
self.main(["status", "-c", "--show-checksums"] + args)
@@ -730,8 +729,10 @@ class TestRecursiveSyncOperations(TestDataCloudBase):
class TestCheckSumRecalculation(TestDvc):
def test(self):
test_collect = spy(State._collect)
with patch.object(State, "_collect", test_collect):
test_get_file_checksum = spy(RemoteLOCAL.get_file_checksum)
with patch.object(
RemoteLOCAL, "get_file_checksum", test_get_file_checksum
):
url = get_local_url()
ret = main(["remote", "add", "-d", TEST_REMOTE, url])
self.assertEqual(ret, 0)
@@ -743,7 +744,7 @@ class TestCheckSumRecalculation(TestDvc):
self.assertEqual(ret, 0)
ret = main(["run", "-d", self.FOO, "echo foo"])
self.assertEqual(ret, 0)
self.assertEqual(test_collect.mock.call_count, 1)
self.assertEqual(test_get_file_checksum.mock.call_count, 1)
class TestShouldWarnOnNoChecksumInLocalAndRemoteCache(TestDvc):
tests/func/test_dependency.py
+14 -8
View File
@@ -1,15 +1,21 @@
import copy
import pytest
from dvc.stage import Stage
from dvc.dependency import _get, DEP_MAP
from tests.func.test_output import TestOutScheme
from tests.func.test_output import TESTS as OUT_TESTS
class TestDepScheme(TestOutScheme):
MAP = DEP_MAP
TESTS = copy.copy(OUT_TESTS)
TESTS.append(("http://example.com/path/to/file", "http"))
TESTS.append(("https://example.com/path/to/file", "https"))
TESTS = TestOutScheme.TESTS.copy()
TESTS["http://example.com/path/to/file"] = "http"
TESTS["https://example.com/path/to/file"] = "https"
def _get(self, path):
return _get(Stage(self.dvc), path, None)
def _get_dep(dvc, path):
return _get(Stage(dvc), path, None)
@pytest.mark.parametrize("url,scheme", TESTS)
def test_scheme(dvc, url, scheme):
assert type(_get_dep(dvc, url)) == DEP_MAP[scheme]
tests/func/test_metrics.py
+1
View File
@@ -582,6 +582,7 @@ class TestCachedMetrics(TestDvc):
def _do_add(self, branch):
self.dvc.scm.checkout(branch)
self.dvc.checkout(force=True)
assert not os.path.exists("metrics.json")
with open("metrics.json", "w+") as fd:
json.dump({"metrics": branch}, fd)
tests/func/test_move.py
+15 -7
View File
@@ -124,23 +124,31 @@ class TestMoveFileToDirectoryWithoutSpecifiedTargetName(TestDvc):
class TestMoveDirectoryShouldNotOverwriteExisting(TestDvc):
def test(self):
new_dir_name = "data_dir2"
os.mkdir(new_dir_name)
dir_name = "dir"
orig_listdir = set(os.listdir(self.DATA_DIR))
ret = main(["add", self.DATA_DIR])
self.assertEqual(ret, 0)
ret = main(["move", self.DATA_DIR, new_dir_name])
os.mkdir(dir_name)
new_dir_name = os.path.join(dir_name, self.DATA_DIR)
ret = main(["move", self.DATA_DIR, dir_name])
self.assertEqual(ret, 0)
data_dir_stage = self.DATA_DIR + Stage.STAGE_FILE_SUFFIX
self.assertFalse(os.path.exists(self.DATA_DIR))
self.assertFalse(
os.path.exists(self.DATA_DIR + Stage.STAGE_FILE_SUFFIX)
self.assertFalse(os.path.exists(data_dir_stage))
self.assertTrue(os.path.exists(dir_name))
self.assertEqual(
set(os.listdir(dir_name)),
set([".gitignore", data_dir_stage, self.DATA_DIR]),
)
self.assertTrue(os.path.exists(new_dir_name))
self.assertTrue(os.path.isfile(new_dir_name + Stage.STAGE_FILE_SUFFIX))
self.assertEqual(os.listdir(new_dir_name), [self.DATA_DIR])
self.assertEqual(set(os.listdir(new_dir_name)), orig_listdir)
class TestMoveFileBetweenDirectories(TestDvc):
tests/func/test_output.py
+22 -23
View File
@@ -1,30 +1,29 @@
import pytest
from dvc.stage import Stage
from dvc.output import _get, OUTS_MAP
from tests.basic_env import TestDvc
TESTS = [
("s3://bucket/path", "s3"),
("gs://bucket/path", "gs"),
("ssh://example.com:/dir/path", "ssh"),
("hdfs://example.com/dir/path", "hdfs"),
("path/to/file", "local"),
("path\\to\\file", "local"),
("file", "local"),
("./file", "local"),
(".\\file", "local"),
("../file", "local"),
("..\\file", "local"),
("unknown://path", "local"),
]
class TestOutScheme(TestDvc):
MAP = OUTS_MAP
def _get_out(dvc, path):
return _get(Stage(dvc), path, None, None, None, None)
TESTS = {
"s3://bucket/path": "s3",
"gs://bucket/path": "gs",
"ssh://example.com:/dir/path": "ssh",
"hdfs://example.com/dir/path": "hdfs",
"path/to/file": "local",
"path\\to\\file": "local",
"file": "local",
"./file": "local",
".\\file": "local",
"../file": "local",
"..\\file": "local",
"unknown://path": "local",
}
def _get(self, path):
return _get(Stage(self.dvc), path, None, None, None, None)
def test(self):
for path, cls in self.TESTS.items():
self.assertIsInstance(self._get(path), self.MAP[cls])
@pytest.mark.parametrize("url,scheme", TESTS)
def test_scheme(dvc, url, scheme):
assert type(_get_out(dvc, url)) == OUTS_MAP[scheme]
tests/func/test_state.py
+11 -3
View File
@@ -14,20 +14,27 @@ from tests.basic_env import TestDvc
class TestState(TestDvc):
def test_update(self):
path = os.path.join(self.dvc.root_dir, self.FOO)
path_info = {"scheme": "local", "path": path}
md5 = file_md5(path)[0]
state = State(self.dvc, self.dvc.config.config)
with state:
entry_md5 = state.update(path)
state.save(path_info, md5)
entry_md5 = state.get(path_info)
self.assertEqual(entry_md5, md5)
os.unlink(path)
with open(path, "a") as fd:
fd.write("1")
entry_md5 = state.get(path_info)
self.assertTrue(entry_md5 is None)
md5 = file_md5(path)[0]
entry_md5 = state.update(path)
state.save(path_info, md5)
entry_md5 = state.get(path_info)
self.assertEqual(entry_md5, md5)
@@ -66,9 +73,10 @@ class TestGetStateRecordForInode(TestDvc):
self.assertNotEqual(inode, state._to_sqlite(inode))
path = os.path.join(self.dvc.root_dir, self.FOO)
md5 = file_md5(path)[0]
get_inode_mock.side_effect = self.mock_get_inode(path, inode)
with state:
state.update(path)
state.save({"scheme": "local", "path": path}, md5)
ret = state.get_state_record_for_inode(inode)
self.assertIsNotNone(ret)
tests/unit/output/test_local.py
+4 -6
View File
@@ -26,22 +26,20 @@ class TestGetFilesNumber(TestDvc):
stage = Stage(self.dvc)
return OutputLOCAL(stage, "path")
@patch.object(OutputLOCAL, "cache", None)
def test_return_0_on_no_cache(self):
o = self._get_output()
o.use_cache = False
self.assertEqual(0, o.get_files_number())
@patch.object(OutputLOCAL, "cache", "cache")
@patch.object(OutputLOCAL, "is_dir_cache", True)
@patch.object(OutputLOCAL, "checksum", "12345678.dir")
@patch.object(OutputLOCAL, "dir_cache", [{"md5": "asdf"}, {"md5": "qwe"}])
def test_return_mutiple_for_dir(self):
o = self._get_output()
self.assertEqual(2, o.get_files_number())
@patch.object(OutputLOCAL, "cache", "cache")
@patch.object(OutputLOCAL, "is_dir_cache", False)
@patch.object(OutputLOCAL, "checksum", "12345678")
@patch.object(OutputLOCAL, "is_dir_checksum", False)
def test_return_1_on_single_file_cache(self):
o = self._get_output()
tests/unit/remote/ssh/test_connection.py
+65 -23
View File
@@ -1,6 +1,9 @@
from __future__ import unicode_literals
import mockssh
import os
from pytest import yield_fixture
import posixpath
import pytest
from dvc.remote.ssh.connection import SSHConnection
@@ -10,35 +13,74 @@ user = "user"
key_path = os.path.join(here, "{0}.key".format(user))
@yield_fixture()
def server():
@pytest.yield_fixture()
def ssh():
users = {user: key_path}
with mockssh.Server(users) as s:
yield s
yield SSHConnection(
s.host, username=user, port=s.port, key_filename=key_path
)
def _client(server):
client = SSHConnection(
server.host, username=user, port=server.port, key_filename=key_path
)
return client
def test_connection(ssh):
assert ssh.execute("ls /")
def test_connection(server):
client = _client(server)
ls = client.execute("ls /")
assert ls
def test_isdir(ssh):
assert ssh.isdir(here)
assert not ssh.isdir(__file__)
def test_isdir(server):
client = _client(server)
path = here
isdir = client.isdir(path=path)
assert isdir is True
def test_exists(ssh):
assert not ssh.exists("/path/to/non/existent/file")
assert ssh.exists(__file__)
def test_file_exists(server):
client = _client(server)
path = "/path/to/file"
file_path = client.file_exists(path=path)
assert file_path is False
def test_isfile(ssh):
assert ssh.isfile(__file__)
assert not ssh.isfile(here)
def test_makedirs(tmp_path, ssh):
tmp = tmp_path.absolute().as_posix()
path = posixpath.join(tmp, "dir", "subdir")
ssh.makedirs(path)
assert os.path.isdir(path)
def test_walk(tmp_path, ssh):
root_path = tmp_path
dir_path = root_path / "dir"
subdir_path = dir_path / "subdir"
dir_path.mkdir()
subdir_path.mkdir()
root_data_path = root_path / "root_data"
dir_data_path = dir_path / "dir_data"
subdir_data_path = subdir_path / "subdir_data"
with root_data_path.open("w+") as fobj:
fobj.write("")
with dir_data_path.open("w+") as fobj:
fobj.write("")
with subdir_data_path.open("w+") as fobj:
fobj.write("")
entries = [
dir_path,
subdir_path,
root_data_path,
dir_data_path,
subdir_data_path,
]
expected = set([entry.absolute().as_posix() for entry in entries])
paths = set()
for root, dirs, files in ssh.walk(root_path.absolute().as_posix()):
for entry in dirs + files:
paths.add(posixpath.join(root, entry))
assert paths == expected
tests/unit/remote/test_base.py
+50 -43
View File
@@ -1,61 +1,68 @@
import mock
from unittest import TestCase
from dvc.remote import REMOTES, RemoteLOCAL, RemoteSSH, RemoteHDFS
from dvc.remote.base import RemoteBase, RemoteCmdError, RemoteMissingDepsError
class TestMissingDeps(TestCase):
class TestRemoteBase(object):
REMOTE_CLS = RemoteBase
class TestMissingDeps(TestCase, TestRemoteBase):
def test(self):
remotes = REMOTES + [RemoteLOCAL, RemoteBase]
for remote_class in remotes:
REQUIRES = {"foo": None, "bar": None, "mock": mock}
with mock.patch.object(remote_class, "REQUIRES", REQUIRES):
with self.assertRaises(RemoteMissingDepsError):
remote_class(None, {})
REQUIRES = {"foo": None, "bar": None, "mock": mock}
with mock.patch.object(self.REMOTE_CLS, "REQUIRES", REQUIRES):
with self.assertRaises(RemoteMissingDepsError):
self.REMOTE_CLS(None, {})
class TestCmdError(TestCase):
class TestCmdError(TestCase, TestRemoteBase):
def test(self):
for remote_class in [RemoteSSH, RemoteHDFS]:
repo = None
config = {}
repo = None
config = {}
remote_name = remote_class.scheme
cmd = "sed 'hello'"
ret = "1"
err = "sed: expression #1, char 2: extra characters after command"
cmd = "sed 'hello'"
ret = "1"
err = "sed: expression #1, char 2: extra characters after command"
with mock.patch.object(
remote_class,
"remove",
side_effect=RemoteCmdError(remote_name, cmd, ret, err),
):
with self.assertRaises(RemoteCmdError):
remote_class(repo, config).remove("file")
with mock.patch.object(
self.REMOTE_CLS,
"remove",
side_effect=RemoteCmdError("base", cmd, ret, err),
):
with self.assertRaises(RemoteCmdError):
self.REMOTE_CLS(repo, config).remove("file")
class TestCacheExists(TestCase):
def test(self):
for remote_class in REMOTES:
config = {
"url": remote_class.scheme + "://example/prefix",
"connection_string": "1234567",
}
remote = remote_class(None, config)
config = {
"url": "base://example/prefix",
"connection_string": "1234567",
}
remote = RemoteBase(None, config)
with mock.patch.object(remote, "changed_cache", return_value=True):
with mock.patch.object(remote, "copy") as cp:
remote.save(
remote.path_info, {remote.PARAM_CHECKSUM: "1234567890"}
)
cp.assert_called_once()
remote.PARAM_CHECKSUM = "checksum"
remote.path_info = {}
remote.url = ""
remote.prefix = ""
path_info = {"scheme": None, "path": "example"}
checksum_info = {remote.PARAM_CHECKSUM: "1234567890"}
with mock.patch.object(
remote, "changed_cache", return_value=False
):
with mock.patch.object(remote, "copy") as cp:
remote.save(
remote.path_info, {remote.PARAM_CHECKSUM: "1234567890"}
)
cp.assert_not_called()
with mock.patch.object(remote, "_checkout") as mock_checkout:
with mock.patch.object(remote, "_save") as mock_save:
with mock.patch.object(
remote, "changed_cache", return_value=True
):
remote.save(path_info, checksum_info)
mock_save.assert_called_once()
mock_checkout.assert_not_called()
with mock.patch.object(remote, "_checkout") as mock_checkout:
with mock.patch.object(remote, "_save") as mock_save:
with mock.patch.object(
remote, "changed_cache", return_value=False
):
remote.save(path_info, checksum_info)
mock_save.assert_not_called()
mock_checkout.assert_called_once()