From 41b67c3730b3da4783099d16dfb660ac4208ebd6 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Sun, 14 Apr 2019 20:07:12 +0300 Subject: [PATCH 1/6] move: move directory into an existing directory Signed-off-by: Ruslan Kuprieiev --- dvc/repo/move.py | 2 +- tests/func/test_move.py | 22 +++++++++++++++------- 2 files changed, 16 insertions(+), 8 deletions(-) diff --git a/dvc/repo/move.py b/dvc/repo/move.py index 9240880a7..ab40dc64b 100644 --- a/dvc/repo/move.py +++ b/dvc/repo/move.py @@ -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 diff --git a/tests/func/test_move.py b/tests/func/test_move.py index 0ad451115..ccf2d3937 100644 --- a/tests/func/test_move.py +++ b/tests/func/test_move.py @@ -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): From 00cfb04fea046946c0604985b3336e35bf7afd56 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Mon, 15 Apr 2019 01:55:03 +0300 Subject: [PATCH 2/6] remote: generalize file/dir handling Fixes #1654 Fixes #1655 Signed-off-by: Ruslan Kuprieiev --- dvc/dependency/http.py | 3 +- dvc/output/base.py | 95 +++++++-- dvc/output/hdfs.py | 3 +- dvc/output/local.py | 65 +----- dvc/output/s3.py | 8 +- dvc/output/ssh.py | 11 +- dvc/remote/azure.py | 4 +- dvc/remote/base.py | 364 +++++++++++++++++++++++++++++--- dvc/remote/gs.py | 19 +- dvc/remote/hdfs.py | 12 +- dvc/remote/http.py | 9 +- dvc/remote/local.py | 352 +++++------------------------- dvc/remote/s3.py | 18 +- dvc/remote/ssh/__init__.py | 20 +- dvc/remote/ssh/connection.py | 5 + dvc/repo/__init__.py | 6 +- dvc/repo/diff.py | 4 +- dvc/repo/move.py | 2 +- dvc/stage.py | 4 +- dvc/state.py | 170 +++++++-------- tests/func/test_add.py | 30 +-- tests/func/test_checkout.py | 2 +- tests/func/test_data_cloud.py | 17 +- tests/func/test_metrics.py | 1 + tests/func/test_state.py | 14 +- tests/unit/output/test_local.py | 10 +- tests/unit/remote/test_base.py | 93 ++++---- 27 files changed, 676 insertions(+), 665 deletions(-) diff --git a/dvc/dependency/http.py b/dvc/dependency/http.py index 0814ccec3..5b216c0fe 100644 --- a/dvc/dependency/http.py +++ b/dvc/dependency/http.py @@ -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 diff --git a/dvc/output/base.py b/dvc/output/base.py index 1dad6b84b..23704b6e3 100644 --- a/dvc/output/base.py +++ b/dvc/output/base.py @@ -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: diff --git a/dvc/output/hdfs.py b/dvc/output/hdfs.py index 0b076054f..2a657745e 100644 --- a/dvc/output/hdfs.py +++ b/dvc/output/hdfs.py @@ -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 diff --git a/dvc/output/local.py b/dvc/output/local.py index b984083b0..3d9a0cd7f 100644 --- a/dvc/output/local.py +++ b/dvc/output/local.py @@ -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 diff --git a/dvc/output/s3.py b/dvc/output/s3.py index 20c38ded7..1b0305c94 100644 --- a/dvc/output/s3.py +++ b/dvc/output/s3.py @@ -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 diff --git a/dvc/output/ssh.py b/dvc/output/ssh.py index ace06d2ec..2c90b7094 100644 --- a/dvc/output/ssh.py +++ b/dvc/output/ssh.py @@ -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 diff --git a/dvc/remote/azure.py b/dvc/remote/azure.py index da13d942b..9e2a19b6e 100644 --- a/dvc/remote/azure.py +++ b/dvc/remote/azure.py @@ -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( diff --git a/dvc/remote/base.py b/dvc/remote/base.py index abdcf633c..560765589 100644 --- a/dvc/remote/base.py +++ b/dvc/remote/base.py @@ -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 json.JSONDecodeError: + 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 diff --git a/dvc/remote/gs.py b/dvc/remote/gs.py index 9e494ce97..bd3919d36 100644 --- a/dvc/remote/gs.py +++ b/dvc/remote/gs.py @@ -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"]) diff --git a/dvc/remote/hdfs.py b/dvc/remote/hdfs.py index 4a0a66ff6..abbbecad4 100644 --- a/dvc/remote/hdfs.py +++ b/dvc/remote/hdfs.py @@ -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.*)" 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): diff --git a/dvc/remote/http.py b/dvc/remote/http.py index b505eca50..683460141 100644 --- a/dvc/remote/http.py +++ b/dvc/remote/http.py @@ -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") diff --git a/dvc/remote/local.py b/dvc/remote/local.py index 04c043d3a..33f87c68f 100644 --- a/dvc/remote/local.py +++ b/dvc/remote/local.py @@ -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.*)$" 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) diff --git a/dvc/remote/s3.py b/dvc/remote/s3.py index 22f7454d0..0dfe372d0 100644 --- a/dvc/remote/s3.py +++ b/dvc/remote/s3.py @@ -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( diff --git a/dvc/remote/ssh/__init__.py b/dvc/remote/ssh/__init__.py index 0ea9a90a5..2f2e8a9d6 100644 --- a/dvc/remote/ssh/__init__.py +++ b/dvc/remote/ssh/__init__.py @@ -95,7 +95,7 @@ class RemoteSSH(RemoteBase): with self.ssh(**path_info) as ssh: return ssh.file_exists(path_info["path"]) - def md5(self, path_info): + def get_file_checksum(self, path_info): if path_info["scheme"] != self.scheme: raise NotImplementedError @@ -112,12 +112,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 +119,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 +175,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: diff --git a/dvc/remote/ssh/connection.py b/dvc/remote/ssh/connection.py index 3b2b64594..7db8b45b5 100644 --- a/dvc/remote/ssh/connection.py +++ b/dvc/remote/ssh/connection.py @@ -183,6 +183,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() diff --git a/dvc/repo/__init__.py b/dvc/repo/__init__.py index 4d1a7e57d..394cd585e 100644 --- a/dvc/repo/__init__.py +++ b/dvc/repo/__init__.py @@ -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( diff --git a/dvc/repo/diff.py b/dvc/repo/diff.py index 534385f77..6297560ee 100644 --- a/dvc/repo/diff.py +++ b/dvc/repo/diff.py @@ -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): diff --git a/dvc/repo/move.py b/dvc/repo/move.py index ab40dc64b..6b9e812ed 100644 --- a/dvc/repo/move.py +++ b/dvc/repo/move.py @@ -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: diff --git a/dvc/stage.py b/dvc/stage.py index 05691de8a..a569b245a 100644 --- a/dvc/stage.py +++ b/dvc/stage.py @@ -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, diff --git a/dvc/state.py b/dvc/state.py index 70c7317fb..df4052607 100644 --- a/dvc/state.py +++ b/dvc/state.py @@ -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 diff --git a/tests/func/test_add.py b/tests/func/test_add.py index dfad7599f..de4108b44 100644 --- a/tests/func/test_add.py +++ b/tests/func/test_add.py @@ -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): diff --git a/tests/func/test_checkout.py b/tests/func/test_checkout.py index 1d690c5d1..6671bb238 100644 --- a/tests/func/test_checkout.py +++ b/tests/func/test_checkout.py @@ -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") diff --git a/tests/func/test_data_cloud.py b/tests/func/test_data_cloud.py index 21cc54854..c9377482d 100644 --- a/tests/func/test_data_cloud.py +++ b/tests/func/test_data_cloud.py @@ -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): diff --git a/tests/func/test_metrics.py b/tests/func/test_metrics.py index fec94b501..e8f0168b7 100644 --- a/tests/func/test_metrics.py +++ b/tests/func/test_metrics.py @@ -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) diff --git a/tests/func/test_state.py b/tests/func/test_state.py index 093368e6a..3e620f242 100644 --- a/tests/func/test_state.py +++ b/tests/func/test_state.py @@ -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) diff --git a/tests/unit/output/test_local.py b/tests/unit/output/test_local.py index a9b74dc24..8149e83ff 100644 --- a/tests/unit/output/test_local.py +++ b/tests/unit/output/test_local.py @@ -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() diff --git a/tests/unit/remote/test_base.py b/tests/unit/remote/test_base.py index 95216cfad..12be6a37c 100644 --- a/tests/unit/remote/test_base.py +++ b/tests/unit/remote/test_base.py @@ -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() From 44b95f23a36bc67258eaa89a7faeadea08bc481a Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Mon, 15 Apr 2019 02:07:46 +0300 Subject: [PATCH 3/6] remote: ssh: support directories Signed-off-by: Ruslan Kuprieiev --- dvc/remote/ssh/__init__.py | 15 +++++- dvc/remote/ssh/connection.py | 69 ++++++++++++++++++------ tests/unit/remote/ssh/test_connection.py | 4 +- 3 files changed, 69 insertions(+), 19 deletions(-) diff --git a/dvc/remote/ssh/__init__.py b/dvc/remote/ssh/__init__.py index 2f2e8a9d6..6d1b96238 100644 --- a/dvc/remote/ssh/__init__.py +++ b/dvc/remote/ssh/__init__.py @@ -93,7 +93,7 @@ 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 get_file_checksum(self, path_info): if path_info["scheme"] != self.scheme: @@ -102,6 +102,10 @@ class RemoteSSH(RemoteBase): 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 @@ -203,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"]) diff --git a/dvc/remote/ssh/connection.py b/dvc/remote/ssh/connection.py index 7db8b45b5..1241ef0ae 100644 --- a/dvc/remote/ssh/connection.py +++ b/dvc/remote/ssh/connection.py @@ -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() diff --git a/tests/unit/remote/ssh/test_connection.py b/tests/unit/remote/ssh/test_connection.py index 94a1d3eb4..a570f3213 100644 --- a/tests/unit/remote/ssh/test_connection.py +++ b/tests/unit/remote/ssh/test_connection.py @@ -37,8 +37,8 @@ def test_isdir(server): assert isdir is True -def test_file_exists(server): +def test_exists(server): client = _client(server) path = "/path/to/file" - file_path = client.file_exists(path=path) + file_path = client.exists(path=path) assert file_path is False From b97b99cf156adc5a176a8b223d03841a37ae2035 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Wed, 17 Apr 2019 00:01:56 +0300 Subject: [PATCH 4/6] test: func: use pytest for outputs/dependencies Signed-off-by: Ruslan Kuprieiev --- tests/func/test_dependency.py | 22 ++++++++++------- tests/func/test_output.py | 45 +++++++++++++++++------------------ 2 files changed, 36 insertions(+), 31 deletions(-) diff --git a/tests/func/test_dependency.py b/tests/func/test_dependency.py index e1e1385a0..c0f9279fd 100644 --- a/tests/func/test_dependency.py +++ b/tests/func/test_dependency.py @@ -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] diff --git a/tests/func/test_output.py b/tests/func/test_output.py index cf96dd031..300b263f2 100644 --- a/tests/func/test_output.py +++ b/tests/func/test_output.py @@ -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] From 49f87a5bff75d6cbae109f88c259d813f94998eb Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Fri, 19 Apr 2019 16:55:43 +0300 Subject: [PATCH 5/6] use ValueError instead of JSONDecodeError JSONDecodeError is only available in never python 3 versions and still inherits from ValueError, so we could just use it. Signed-off-by: Ruslan Kuprieiev --- dvc/analytics.py | 5 +++-- dvc/remote/base.py | 2 +- dvc/utils/compat.py | 1 - 3 files changed, 4 insertions(+), 4 deletions(-) diff --git a/dvc/analytics.py b/dvc/analytics.py index 2a7ad7c8f..d9805bfd1 100644 --- a/dvc/analytics.py +++ b/dvc/analytics.py @@ -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 diff --git a/dvc/remote/base.py b/dvc/remote/base.py index 560765589..5522711d3 100644 --- a/dvc/remote/base.py +++ b/dvc/remote/base.py @@ -225,7 +225,7 @@ class RemoteBase(object): try: with open(path, "r") as fobj: d = json.load(fobj) - except json.JSONDecodeError: + except ValueError: logger.exception("Failed to load dir cache '{}'".format(path_info)) return [] finally: diff --git a/dvc/utils/compat.py b/dvc/utils/compat.py index 0e104a0f4..088c43a1f 100644 --- a/dvc/utils/compat.py +++ b/dvc/utils/compat.py @@ -5,7 +5,6 @@ import sys import os import errno - # Syntax sugar. _ver = sys.version_info From 37ee55b1f8aa6c89d0120c38541f6e99d414574e Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Wed, 24 Apr 2019 02:54:26 +0300 Subject: [PATCH 6/6] test: test ssh Signed-off-by: Ruslan Kuprieiev --- tests/unit/remote/ssh/test_connection.py | 88 +++++++++++++++++------- 1 file changed, 65 insertions(+), 23 deletions(-) diff --git a/tests/unit/remote/ssh/test_connection.py b/tests/unit/remote/ssh/test_connection.py index a570f3213..775cd79df 100644 --- a/tests/unit/remote/ssh/test_connection.py +++ b/tests/unit/remote/ssh/test_connection.py @@ -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_exists(server): - client = _client(server) - path = "/path/to/file" - file_path = client.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