diff --git a/.dvc/.gitignore b/.dvc/.gitignore index a53edd2f3..5b2f952a6 100644 --- a/.dvc/.gitignore +++ b/.dvc/.gitignore @@ -1,10 +1,8 @@ -link.state -link.state.lock -lock -config.local -updater -cache -state -state.lock -state-journal -state-wal +/state +/lock +/config.local +/updater +/updater.lock +/state-journal +/state-wal +/cache diff --git a/dvc/dependency/__init__.py b/dvc/dependency/__init__.py index 7377f94f0..52c16bf30 100644 --- a/dvc/dependency/__init__.py +++ b/dvc/dependency/__init__.py @@ -39,11 +39,11 @@ DEP_MAP = { 'https': DependencyHTTP, } -# We are skipping RemoteHTTP.PARAM_ETAG because is the same as RemoteS3 +# We are skipping RemoteHTTP.PARAM_CHECKSUM because is the same as RemoteS3 SCHEMA = { DependencyBase.PARAM_PATH: str, - schema.Optional(RemoteLOCAL.PARAM_MD5): schema.Or(str, None), - schema.Optional(RemoteS3.PARAM_ETAG): schema.Or(str, None), + schema.Optional(RemoteLOCAL.PARAM_CHECKSUM): schema.Or(str, None), + schema.Optional(RemoteS3.PARAM_CHECKSUM): schema.Or(str, None), schema.Optional(RemoteHDFS.PARAM_CHECKSUM): schema.Or(str, None), } diff --git a/dvc/dependency/base.py b/dvc/dependency/base.py index 62637cb14..4700e8fc8 100644 --- a/dvc/dependency/base.py +++ b/dvc/dependency/base.py @@ -60,7 +60,7 @@ class DependencyBase(object): @property def exists(self): - return self.remote.exists([self.path_info])[0] + return self.remote.exists(self.path_info) def changed(self): raise NotImplementedError diff --git a/dvc/dependency/gs.py b/dvc/dependency/gs.py index 2c0be604e..6f31f3d9e 100644 --- a/dvc/dependency/gs.py +++ b/dvc/dependency/gs.py @@ -17,9 +17,9 @@ class DependencyGS(DependencyS3): self.info = info self.remote = remote if remote else RemoteGS(stage.project, {}) bucket = remote.bucket if remote else urlparse(path).netloc - key = urlparse(path).path.lstrip('/') + path = urlparse(path).path.lstrip('/') if remote: - key = posixpath.join(remote.prefix, key) + path = posixpath.join(remote.prefix, path) self.path_info = {'scheme': 'gs', 'bucket': bucket, - 'key': key} + 'path': path} diff --git a/dvc/dependency/hdfs.py b/dvc/dependency/hdfs.py index 4ea959309..81506ba97 100644 --- a/dvc/dependency/hdfs.py +++ b/dvc/dependency/hdfs.py @@ -21,7 +21,7 @@ class DependencyHDFS(DependencyBase): user = remote.user if remote else self.group('user') self.path_info = {'scheme': 'hdfs', 'user': user, - 'url': path} + 'path': path} def changed(self): if not self.exists: diff --git a/dvc/dependency/http.py b/dvc/dependency/http.py index f05bcbed4..a2cccf454 100644 --- a/dvc/dependency/http.py +++ b/dvc/dependency/http.py @@ -21,7 +21,7 @@ class DependencyHTTP(DependencyBase): self.path_info = { 'scheme': urlparse(path).scheme, - 'url': path, + 'path': path, } def save(self): diff --git a/dvc/dependency/local.py b/dvc/dependency/local.py index b5671cbee..a9ffb6efc 100644 --- a/dvc/dependency/local.py +++ b/dvc/dependency/local.py @@ -30,7 +30,7 @@ class DependencyLOCAL(DependencyBase): path = os.path.join(remote.prefix, urlparse(path).path.lstrip('/')) if not os.path.isabs(path): - path = self.remote.ospath(path) + path = self.remote.to_ospath(path) path = os.path.join(stage.cwd, path) self.path = os.path.abspath(os.path.normpath(path)) diff --git a/dvc/dependency/s3.py b/dvc/dependency/s3.py index 9eb2a48d0..d2f0c4eef 100644 --- a/dvc/dependency/s3.py +++ b/dvc/dependency/s3.py @@ -18,12 +18,12 @@ class DependencyS3(DependencyBase): self.remote = remote or RemoteS3(stage.project, {}) bucket = remote.bucket if remote else urlparse(path).netloc - key = urlparse(path).path.lstrip('/') + path = urlparse(path).path.lstrip('/') if remote: - key = posixpath.join(remote.prefix, key) + path = posixpath.join(remote.prefix, path) self.path_info = {'scheme': 's3', 'bucket': bucket, - 'key': key} + 'path': path} def changed(self): if not self.exists: diff --git a/dvc/output/local.py b/dvc/output/local.py index d52bbaea5..6c1cf6541 100644 --- a/dvc/output/local.py +++ b/dvc/output/local.py @@ -34,7 +34,7 @@ class OutputLOCAL(DependencyLOCAL): @property def md5(self): - return self.info.get(self.project.cache.local.PARAM_MD5, None) + return self.info.get(self.project.cache.local.PARAM_CHECKSUM, None) @property def cache(self): diff --git a/dvc/project.py b/dvc/project.py index dcd4a6ac1..dcd307487 100644 --- a/dvc/project.py +++ b/dvc/project.py @@ -602,7 +602,7 @@ class Project(object): info = out.dumpd() ret = [info] r = out.remote - md5 = info[r.PARAM_MD5] + md5 = info[r.PARAM_CHECKSUM] if self.cache.local.changed_cache_file(md5): try: @@ -649,7 +649,7 @@ class Project(object): if out.path_info['scheme'] != 'local': return ret - md5 = info[out.remote.PARAM_MD5] + md5 = info[out.remote.PARAM_CHECKSUM] cache = self.cache.local.get(md5) if not out.remote.is_dir_cache(cache): return ret @@ -896,12 +896,15 @@ class Project(object): remote=remote, jobs=jobs)['local'] - status = {} - for md5, ret in self.cloud.status(used, - jobs, - remote=remote, - show_checksums=show_checksums): - if ret == cloud.STATUS_OK: + ret = {} + status_info = self.cloud.status(used, + jobs, + remote=remote, + show_checksums=show_checksums) + for md5, info in status_info.items(): + name = info['name'] + status = info['status'] + if status == cloud.STATUS_OK: continue prefix_map = { @@ -909,9 +912,9 @@ class Project(object): cloud.STATUS_NEW: 'new', } - status[md5] = prefix_map[ret] + ret[name] = prefix_map[status] - return status + return ret def status(self, target=None, diff --git a/dvc/remote/azure.py b/dvc/remote/azure.py index 228c68551..482c254dd 100644 --- a/dvc/remote/azure.py +++ b/dvc/remote/azure.py @@ -1,7 +1,6 @@ from __future__ import absolute_import import os import re -import posixpath try: from azure.storage.blob import BlockBlobService @@ -49,7 +48,7 @@ class RemoteAzure(RemoteBase): or match.group('container_name') # backward compatibility or os.getenv('AZURE_STORAGE_CONTAINER_NAME')) - self.prefix = urlparse(self.url).path.lstrip('/') or '/' + self.prefix = urlparse(self.url).path.lstrip('/') self.connection_string = ( config.get(Config.SECTION_AZURE_CONNECTION_STRING) @@ -64,6 +63,8 @@ class RemoteAzure(RemoteBase): self.__blob_service = None + self.path_info = {'scheme': self.scheme, 'bucket': self.bucket} + @property def blob_service(self): if self.__blob_service is None: @@ -72,104 +73,33 @@ class RemoteAzure(RemoteBase): self.__blob_service.create_container(self.bucket) return self.__blob_service -# FIXME: temporarily disabled because of the lack of test for external azure -# dependencies/outputs/cache. -# -# def _get_etag(self, bucket, key): -# try: -# blob = self.blob_service.get_blob_properties(bucket, key) -# return blob.properties.etag -# except Exception: -# return None -# -# def save_info(self, path_info): -# if path_info['scheme'] != self.scheme: -# raise NotImplementedError -# -# return {self.PARAM_ETAG: self._get_etag( -# path_info['bucket'], path_info['key'])} -# -# def save(self, path_info): -# if path_info['scheme'] != self.scheme: -# raise NotImplementedError -# -# etag = self._get_etag(path_info['bucket'], path_info['key']) -# dest_key = '{}/{}'.format(etag[0:2], etag[2:]) -# -# self._copy_blob(to_bucket=self.bucket, -# to_key=dest_key, -# from_bucket=path_info['bucket'], -# from_key=path_info['key']) -# -# return {self.PARAM_ETAG: etag} -# -# def changed(self, path_info, checksum_info): -# return checksum_info != self.save_info(path_info) -# -# def _copy_blob(self, to_bucket, to_key, from_bucket, from_key): -# source = self.blob_service.make_blob_url(from_bucket, from_key) -# -# copy = self.blob_service.copy_blob(to_bucket, to_key, source) -# -# if self.COPY_POLL_SECONDS <= 0: -# return -# -# while copy.status != 'success': -# time.sleep(self.COPY_POLL_SECONDS) -# copy = self.blob_service.get_blob_properties( -# to_bucket, to_key).properties.copy -# -# def checkout(self, path_info, checksum_info): -# if path_info['scheme'] != self.scheme: -# raise NotImplementedError -# -# etag = checksum_info.get(self.PARAM_ETAG, None) -# if not etag: -# return -# -# self._copy_blob(to_bucket=path_info['bucket'], -# to_key=path_info['key'], -# from_bucket=self.bucket, -# from_key='{}/{}'.format(etag[0:2], etag[2:])) -# def remove(self, path_info): if path_info['scheme'] != self.scheme: raise NotImplementedError logger.debug('Removing azure://{}/{}'.format(path_info['bucket'], - path_info['key'])) + path_info['path'])) - self.blob_service.delete_blob(path_info['bucket'], path_info['key']) + self.blob_service.delete_blob(path_info['bucket'], path_info['path']) - def md5s_to_path_infos(self, md5s): - return [{ - 'scheme': self.scheme, - 'bucket': self.bucket, - 'key': '{}/{}/{}'.format(self.prefix, md5[0:2], md5[2:]) - } for md5 in md5s] + def _list_paths(self, bucket, prefix): + blob_service = self.blob_service + next_marker = None + while True: + blobs = blob_service.list_blobs(bucket, + prefix=prefix, + marker=next_marker) - def _all_keys(self, bucket, prefix): - return {blob.name - for blob in self.blob_service.list_blobs(bucket, - prefix=prefix)} + for blob in blobs: + yield blob.name - def exists(self, path_infos): - ret = [] + if not blobs.next_marker: + break - if len(path_infos) == 0: - return ret + next_marker = blobs.next_marker - bucket = path_infos[0]['bucket'] - key = path_infos[0]['key'] - - for path_info in path_infos: - assert path_info['scheme'] == self.scheme - assert path_info['bucket'] == bucket - - key = path_info['key'] - ret.append(key in self._all_keys(bucket, key)) - - return ret + def list_cache_paths(self): + return self._list_paths(self.bucket, self.prefix) def upload(self, from_infos, to_infos, names=None): names = self._verify_path_args(to_infos, from_infos, names) @@ -182,10 +112,10 @@ class RemoteAzure(RemoteBase): raise NotImplementedError bucket = to_info['bucket'] - key = to_info['key'] + path = to_info['path'] logger.debug("Uploading '{}' to '{}/{}'".format( - from_info['path'], bucket, key)) + from_info['path'], bucket, path)) if not name: name = os.path.basename(from_info['path']) @@ -194,7 +124,7 @@ class RemoteAzure(RemoteBase): try: self.blob_service.create_blob_from_path( - bucket, key, from_info['path'], progress_callback=cb) + bucket, path, from_info['path'], progress_callback=cb) except Exception as ex: msg = "Failed to upload '{}'".format(from_info['path']) logger.warn(msg, ex) @@ -216,10 +146,10 @@ class RemoteAzure(RemoteBase): raise NotImplementedError bucket = from_info['bucket'] - key = from_info['key'] + path = from_info['path'] logger.debug("Downloading '{}/{}' to '{}'".format( - bucket, key, to_info['path'])) + bucket, path, to_info['path'])) tmp_file = self.tmp_file(to_info['path']) if not name: @@ -231,40 +161,12 @@ class RemoteAzure(RemoteBase): try: self.blob_service.get_blob_to_path( - bucket, key, tmp_file, progress_callback=cb) + bucket, path, tmp_file, progress_callback=cb) except Exception as exc: - msg = "Failed to download '{}/{}'".format(bucket, key) + msg = "Failed to download '{}/{}'".format(bucket, path) logger.warn(msg, exc) else: os.rename(tmp_file, to_info['path']) if not no_progress_bar: progress.finish_target(name) - - def _path_to_etag(self, path): - relpath = posixpath.relpath(path, self.prefix) - return posixpath.dirname(relpath) + posixpath.basename(relpath) - - def _all(self): - keys = self._all_keys(self.bucket, self.prefix) - return [self._path_to_etag(key) for key in keys] - - def gc(self, cinfos): - from dvc.remote.local import RemoteLOCAL - - used = [info[self.PARAM_ETAG] for info in cinfos['azure']] - used += [info[RemoteLOCAL.PARAM_MD5] for info in cinfos['local']] - - removed = False - for etag in self._all(): - if etag in used: - continue - path_info = {'scheme': self.scheme, - 'key': posixpath.join(self.prefix, - etag[0:2], - etag[2:]), - 'bucket': self.bucket} - self.remove(path_info) - removed = True - - return removed diff --git a/dvc/remote/base.py b/dvc/remote/base.py index 02ab6b588..53eb27beb 100644 --- a/dvc/remote/base.py +++ b/dvc/remote/base.py @@ -1,8 +1,10 @@ import os import re import errno +import posixpath from multiprocessing import cpu_count +import dvc.prompt as prompt from dvc.config import Config from dvc.logger import logger from dvc.exceptions import DvcException @@ -104,9 +106,6 @@ class RemoteBase(object): def save(self, path_info): raise NotImplementedError - def checkout(self, path_info, checksum_info, force=False): - raise NotImplementedError - def download(self, from_infos, to_infos, no_progress_bar=False, name=None): raise NotImplementedError @@ -119,6 +118,9 @@ class RemoteBase(object): def move(self, path_info): raise NotImplementedError + def copy(self, from_info, to_info): + raise NotImplementedError + def _makedirs(self, fname): dname = os.path.dirname(fname) @@ -128,9 +130,6 @@ class RemoteBase(object): if e.errno != errno.EEXIST: raise - def md5s_to_path_infos(self, md5s): - raise NotImplementedError - def exists(self, path_infos): raise NotImplementedError @@ -147,3 +146,145 @@ class RemoteBase(object): assert len(names) == len(to_infos) return names + + @property + def ospath(self): + return posixpath + + def checksum_to_path(self, checksum): + return self.ospath.join(self.prefix, checksum[0:2], checksum[2:]) + + def path_to_checksum(self, path): + relpath = self.ospath.relpath(path, self.prefix) + return self.ospath.dirname(relpath) + self.ospath.basename(relpath) + + def checksum_to_path_info(self, checksum): + path_info = self.path_info.copy() + path_info['path'] = self.checksum_to_path(checksum) + return path_info + + def md5s_to_path_infos(self, md5s): + return [ + self.checksum_to_path_info(md5) + for md5 in md5s + ] + + def list_cache_paths(self): + raise NotImplementedError + + def all(self): + # NOTE: The list might be way too big(e.g. 100M entries, md5 for each + # is 32 bytes, so ~3200Mb list) and we don't really need all of it at + # the same time, so it makes sense to use a generator to gradually + # iterate over it, without keeping all of it in memory. + return ( + self.path_to_checksum(path) + for path in self.list_cache_paths() + ) + + def gc(self, cinfos): + from dvc.remote.local import RemoteLOCAL + + used = [info[RemoteLOCAL.PARAM_CHECKSUM] for info in cinfos['local']] + + if self.scheme != '': + used += [info[self.PARAM_CHECKSUM] for info in cinfos[self.scheme]] + + removed = False + for checksum in self.all(): + if checksum in used: + continue + path_info = self.checksum_to_path_info(checksum) + self.remove(path_info) + removed = True + return removed + + def changed_cache(self, checksum): + cache = self.checksum_to_path_info(checksum) + expected = {self.PARAM_CHECKSUM: checksum} + actual = self.save_info(cache) + + logger.debug("Cache '{}' actual '{}'.".format(str(expected), + str(actual))) + + if expected != actual: + if self.exists(cache): + msg = 'Corrupted cache file {}' + logger.warn(msg.format(str(cache))) + self.remove(cache) + return True + + return False + + 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 + # a certain prefix(e.g. s3, gs, ssh, hdfs). Other remotes that can + # check if particular file exists much quicker, use their own + # implementation of cache_exists(see http, local). + # + # Result of all() might be way too big, so we should walk through + # it in one pass. + 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] + + if not current: + return False + + return not self.changed_cache(current) + + def safe_remove(self, path_info, force=False): + if not self.exists(path_info): + return + + if not force and not self.already_cached(path_info): + msg = ( + 'File "{}" is going to be removed. ' + 'Are you sure you want to proceed?' + .format(str(path_info)) + ) + + if not prompt.confirm(msg): + raise DvcException("Unable to remove {} without a confirmation" + " from the user. Use '-f' to force." + .format(str(path_info))) + + self.remove(path_info) + + def do_checkout(self, path_info, checksum, force=False): + if self.exists(path_info): + msg = "Data '{}' exists. Removing before checkout." + logger.warn(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) + + def checkout(self, path_info, checksum_info, force=False): + scheme = path_info['scheme'] + if scheme not in ['', 'local'] and scheme != self.scheme: + raise NotImplementedError + + checksum = checksum_info.get(self.PARAM_CHECKSUM) + if not checksum: + msg = "No checksum info for '{}'." + logger.info(msg.format(str(path_info))) + return + + if not self.changed(path_info, checksum_info): + msg = "Data '{}' didn't change." + logger.info(msg.format(str(path_info))) + return + + if self.changed_cache(checksum): + msg = "Cache '{}' not found. File '{}' won't be created." + logger.warn(msg.format(checksum, str(path_info))) + self.safe_remove(path_info, force=force) + return + + msg = "Checking out '{}' with cache '{}'." + logger.info(msg.format(str(path_info), checksum)) + + self.do_checkout(path_info, checksum, force=force) diff --git a/dvc/remote/gs.py b/dvc/remote/gs.py index 23830071f..531e4e7e8 100644 --- a/dvc/remote/gs.py +++ b/dvc/remote/gs.py @@ -1,5 +1,4 @@ import os -import posixpath try: from google.cloud import storage @@ -13,7 +12,6 @@ except ImportError: from dvc.logger import logger from dvc.remote.base import RemoteBase -from dvc.remote.local import RemoteLOCAL from dvc.config import Config from dvc.progress import progress from dvc.exceptions import DvcException @@ -23,7 +21,7 @@ class RemoteGS(RemoteBase): scheme = 'gs' REGEX = r'^gs://(?P.*)$' REQUIRES = {'google.cloud.storage': storage} - PARAM_MD5 = 'md5' + PARAM_CHECKSUM = 'md5' def __init__(self, project, config): self.project = project @@ -33,6 +31,12 @@ class RemoteGS(RemoteBase): self.url = config.get(Config.SECTION_REMOTE_URL, storagepath) self.projectname = config.get(Config.SECTION_GCP_PROJECTNAME, None) + parsed = urlparse(self.url) + self.bucket = parsed.netloc + self.prefix = parsed.path.lstrip('/') + + self.path_info = {'scheme': 'gs', 'bucket': self.bucket} + @staticmethod def compat_config(config): ret = config.copy() @@ -40,23 +44,15 @@ class RemoteGS(RemoteBase): ret[Config.SECTION_REMOTE_URL] = url return ret - @property - def bucket(self): - return urlparse(self.url).netloc - - @property - def prefix(self): - return urlparse(self.url).path.lstrip('/') - @property def gs(self): return storage.Client() - def get_md5(self, bucket, key): + def get_md5(self, bucket, path): import base64 import codecs - blob = self.gs.bucket(bucket).get_blob(key) + blob = self.gs.bucket(bucket).get_blob(path) if not blob: return None @@ -68,33 +64,14 @@ class RemoteGS(RemoteBase): if path_info['scheme'] != 'gs': raise NotImplementedError - return {self.PARAM_MD5: self.get_md5(path_info['bucket'], - path_info['key'])} - - @staticmethod - def to_string(path_info): - return "{}://{}/{}".format(path_info['scheme'], - path_info['bucket'], - path_info['key']) - - def changed_cache(self, md5): - key = posixpath.join(self.prefix, md5[0:2], md5[2:]) - cache = {'scheme': 'gs', 'bucket': self.bucket, 'key': key} - - if {self.PARAM_MD5: md5} != self.save_info(cache): - if self.exists([cache])[0]: - msg = 'Corrupted cache file {}' - logger.warn(msg.format(self.to_string(cache))) - self.remove(cache) - return True - - return False + return {self.PARAM_CHECKSUM: self.get_md5(path_info['bucket'], + path_info['path'])} def changed(self, path_info, checksum_info): - if not self.exists([path_info])[0]: + if not self.exists(path_info): return True - md5 = checksum_info.get(self.PARAM_MD5, None) + md5 = checksum_info.get(self.PARAM_CHECKSUM, None) if md5 is None: return True @@ -103,103 +80,57 @@ class RemoteGS(RemoteBase): return checksum_info != self.save_info(path_info) - def _copy(self, from_info, to_info, gs=None): + def copy(self, from_info, to_info, gs=None): gs = gs if gs else self.gs - blob = gs.bucket(from_info['bucket']).get_blob(from_info['key']) + blob = gs.bucket(from_info['bucket']).get_blob(from_info['path']) if not blob: msg = '{} doesn\'t exist in the cloud' - raise DvcException(msg.format(from_info['key'])) + raise DvcException(msg.format(from_info['path'])) bucket = self.gs.bucket(to_info['bucket']) bucket.copy_blob(blob, self.gs.bucket(to_info['bucket']), - new_name=to_info['key']) + new_name=to_info['path']) def save(self, path_info): if path_info['scheme'] != 'gs': raise NotImplementedError - md5 = self.get_md5(path_info['bucket'], path_info['key']) - key = posixpath.join(self.prefix, md5[0:2], md5[2:]) - to_info = {'scheme': 'gs', 'bucket': self.bucket, 'key': key} + md5 = self.get_md5(path_info['bucket'], path_info['path']) + path = self.checksum_to_path(md5) + to_info = {'scheme': 'gs', 'bucket': self.bucket, 'path': path} - self._copy(path_info, to_info) + self.copy(path_info, to_info) - return {self.PARAM_MD5: md5} - - def checkout(self, path_info, checksum_info): - if path_info['scheme'] != 'gs': - raise NotImplementedError - - md5 = checksum_info.get(self.PARAM_MD5, None) - if not md5: - return - - if not self.changed(path_info, checksum_info): - msg = "Data '{}' didn't change." - logger.info(msg.format(self.to_string(path_info))) - return - - if self.changed_cache(md5): - msg = "Cache '{}' not found. File '{}' won't be created." - logger.warn(msg.format(md5, self.to_string(path_info))) - return - - if self.exists([path_info])[0]: - msg = "Data '{}' exists. Removing before checkout." - logger.warn(msg.format(self.to_string(path_info))) - self.remove(path_info) - return - - msg = "Checking out '{}' with cache '{}'." - logger.info(msg.format(self.to_string(path_info), md5)) - - key = posixpath.join(self.prefix, md5[0:2], md5[2:]) - from_info = {'scheme': 'gs', 'bucket': self.bucket, 'key': key} - - self._copy(from_info, path_info) + return {self.PARAM_CHECKSUM: md5} def remove(self, path_info): if path_info['scheme'] != 'gs': raise NotImplementedError logger.debug("Removing gs://{}/{}".format(path_info['bucket'], - path_info['key'])) + path_info['path'])) - blob = self.gs.bucket(path_info['bucket']).get_blob(path_info['key']) + blob = self.gs.bucket(path_info['bucket']).get_blob(path_info['path']) if not blob: return blob.delete() - def md5s_to_path_infos(self, md5s): - return [{'scheme': 'gs', - 'bucket': self.bucket, - 'key': posixpath.join(self.prefix, - md5[0:2], md5[2:])} for md5 in md5s] + def _list_paths(self, bucket, prefix): + for blob in self.gs.bucket(bucket).list_blobs(prefix=prefix): + yield blob.name - def exists(self, path_infos): - ret = [] + def list_cache_paths(self): + return self._list_paths(self.bucket, self.prefix) - if len(path_infos) == 0: - return ret + def exists(self, path_info): + assert not isinstance(path_info, list) + assert path_info['scheme'] == 'gs' - gs = self.gs - bucket_name = path_infos[0]['bucket'] - bucket = gs.bucket(bucket_name) - - for path_info in path_infos: - assert self.scheme == path_info['scheme'] - assert bucket_name == path_info['bucket'] - - exists = False - key = path_info['key'] - keys = [blob.name for blob in bucket.list_blobs(prefix=key)] - if key in keys: - exists = True - ret.append(exists) - return ret + 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): names = self._verify_path_args(to_infos, from_infos, names) @@ -215,7 +146,7 @@ class RemoteGS(RemoteBase): logger.debug("Uploading '{}' to '{}/{}'".format(from_info['path'], to_info['bucket'], - to_info['key'])) + to_info['path'])) if not name: name = os.path.basename(from_info['path']) @@ -224,13 +155,13 @@ class RemoteGS(RemoteBase): try: bucket = gs.bucket(to_info['bucket']) - blob = bucket.blob(to_info['key']) + blob = bucket.blob(to_info['path']) blob.upload_from_filename(from_info['path']) except Exception as exc: msg = "Failed to upload '{}' to '{}/{}'" logger.warn(msg.format(from_info['path'], to_info['bucket'], - to_info['key']), exc) + to_info['path']), exc) continue progress.finish_target(name) @@ -249,14 +180,14 @@ class RemoteGS(RemoteBase): raise NotImplementedError if to_info['scheme'] == 'gs': - self._copy(from_info, to_info, gs=gs) + self.copy(from_info, to_info, gs=gs) continue if to_info['scheme'] != 'local': raise NotImplementedError msg = "Downloading '{}/{}' to '{}'".format(from_info['bucket'], - from_info['key'], + from_info['path'], to_info['path']) logger.debug(msg) @@ -266,19 +197,19 @@ class RemoteGS(RemoteBase): if not no_progress_bar: # percent_cb is not available for download_to_filename, so - # lets at least update progress at keypoints(start, finish) + # lets at least update progress at pathpoints(start, finish) progress.update_target(name, 0, None) self._makedirs(to_info['path']) try: bucket = gs.bucket(from_info['bucket']) - blob = bucket.get_blob(from_info['key']) + blob = bucket.get_blob(from_info['path']) blob.download_to_filename(tmp_file) except Exception as exc: msg = "Failed to download '{}/{}' to '{}'" logger.warn(msg.format(from_info['bucket'], - from_info['key'], + from_info['path'], to_info['path']), exc) continue @@ -286,29 +217,3 @@ class RemoteGS(RemoteBase): if not no_progress_bar: progress.finish_target(name) - - def _path_to_md5(self, path): - relpath = posixpath.relpath(path, self.prefix) - return posixpath.dirname(relpath) + posixpath.basename(relpath) - - def _all_md5s(self): - blobs = self.gs.bucket(self.bucket).list_blobs(prefix=self.prefix) - blobs = list(blobs) - return [self._path_to_md5(blob.name) for blob in blobs] - - def gc(self, cinfos): - used = [info[self.PARAM_MD5] for info in cinfos['gs']] - used += [info[RemoteLOCAL.PARAM_MD5] for info in cinfos['local']] - - removed = False - for md5 in self._all_md5s(): - if md5 in used: - continue - path_info = {'scheme': 'gs', - 'bucket': self.bucket, - 'key': posixpath.join(self.prefix, - md5[0:2], md5[2:])} - self.remove(path_info) - removed = True - - return removed diff --git a/dvc/remote/hdfs.py b/dvc/remote/hdfs.py index dfc1f331c..dc114f894 100644 --- a/dvc/remote/hdfs.py +++ b/dvc/remote/hdfs.py @@ -6,7 +6,6 @@ from subprocess import Popen, PIPE from dvc.config import Config from dvc.remote.base import RemoteBase, RemoteBaseCmdError -from dvc.remote.local import RemoteLOCAL from dvc.logger import logger from dvc.utils import fix_env @@ -23,11 +22,14 @@ class RemoteHDFS(RemoteBase): def __init__(self, project, config): self.project = project self.url = config.get(Config.SECTION_REMOTE_URL, '/') + self.prefix = self.url self.user = self.group('user') if not self.user: self.user = config.get(Config.SECTION_REMOTE_USER, getpass.getuser()) + self.path_info = {'scheme': 'hdfs', 'user': self.user} + def hadoop_fs(self, cmd, user=None): cmd = 'hadoop fs -' + cmd if user: @@ -59,50 +61,31 @@ class RemoteHDFS(RemoteBase): def checksum(self, path_info): regex = r'.*\t.*\t(?P.*)' - stdout = self.hadoop_fs('checksum {}'.format(path_info['url']), + stdout = self.hadoop_fs('checksum {}'.format(path_info['path']), user=path_info['user']) return self._group(regex, stdout, 'checksum') - def cp(self, from_info, to_info): - self.hadoop_fs('mkdir -p {}'.format(posixpath.dirname(to_info['url'])), - user=to_info['user']) - self.hadoop_fs('cp -f {} {}'.format(from_info['url'], to_info['url']), + def copy(self, from_info, to_info): + dname = posixpath.dirname(to_info['path']) + self.hadoop_fs('mkdir -p {}'.format(dname), user=to_info['user']) + self.hadoop_fs('cp -f {} {}'.format(from_info['path'], + to_info['path']), user=to_info['user']) def rm(self, path_info): - self.hadoop_fs('rm {}'.format(path_info['url']), + self.hadoop_fs('rm {}'.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('url') + assert path_info.get('path') return {self.PARAM_CHECKSUM: self.checksum(path_info)} - @staticmethod - def to_string(path_info): - return "{}://{}".format(path_info['scheme'], - path_info['url']) - - def changed_cache(self, checksum): - cache = {} - cache['scheme'] = 'hdfs' - cache['user'] = self.user - cache['url'] = posixpath.join(self.url, checksum[0:2], checksum[2:]) - - if {self.PARAM_CHECKSUM: checksum} != self.save_info(cache): - if self.exists([cache])[0]: - msg = 'Corrupted cache file {}' - logger.warn(msg.format(self.to_string(cache))) - self.remove(cache) - return True - - return False - def changed(self, path_info, checksum_info): - if not self.exists([path_info])[0]: + if not self.exists(path_info): return True checksum = checksum_info.get(self.PARAM_CHECKSUM, None) @@ -118,82 +101,35 @@ class RemoteHDFS(RemoteBase): if path_info['scheme'] != 'hdfs': raise NotImplementedError - assert path_info.get('url') + assert path_info.get('path') checksum = self.checksum(path_info) dest = path_info.copy() - dest['url'] = posixpath.join(self.url, checksum[0:2], checksum[2:]) + dest['path'] = self.checksum_to_path(checksum) - self.cp(path_info, dest) + self.copy(path_info, dest) return {self.PARAM_CHECKSUM: checksum} - def checkout(self, path_info, checksum_info): - if path_info['scheme'] != 'hdfs': - raise NotImplementedError - - assert path_info.get('url') - - checksum = checksum_info.get(self.PARAM_CHECKSUM, None) - if not checksum: - return - - if not self.changed(path_info, checksum_info): - msg = "Data '{}' didn't change." - logger.info(msg.format(self.to_string(path_info))) - return - - if self.changed_cache(checksum): - msg = "Cache '{}' not found. File '{}' won't be created." - logger.warn(msg.format(checksum, self.to_string(path_info))) - return - - if self.exists([path_info])[0]: - msg = "Data '{}' exists. Removing before checkout." - logger.warn(msg.format(self.to_string(path_info))) - self.remove(path_info) - return - - msg = "Checking out '{}' with cache '{}'." - logger.info(msg.format(self.to_string(path_info), checksum)) - - src = path_info.copy() - src['url'] = posixpath.join(self.url, checksum[0:2], checksum[2:]) - - self.cp(src, path_info) - def remove(self, path_info): if path_info['scheme'] != 'hdfs': raise NotImplementedError - assert path_info.get('url') + assert path_info.get('path') - logger.debug('Removing {}'.format(path_info['url'])) + logger.debug('Removing {}'.format(path_info['path'])) self.rm(path_info) - def md5s_to_path_infos(self, md5s): - return [{'scheme': 'hdfs', - 'user': self.user, - 'url': posixpath.join(self.url, - md5[0:2], md5[2:])} for md5 in md5s] + def exists(self, path_info): + assert not isinstance(path_info, list) + assert path_info['scheme'] == 'hdfs' - def exists(self, path_infos): - ret = [] - - if len(path_infos) == 0: - return ret - - for path_info in path_infos: - assert path_info['scheme'] == self.scheme - try: - self.hadoop_fs('test -e {}'.format(path_info['url'])) - exists = True - except RemoteHDFSCmdError: - exists = False - ret.append(exists) - - return ret + try: + self.hadoop_fs('test -e {}'.format(path_info['path'])) + return True + except RemoteHDFSCmdError: + return False def upload(self, from_infos, to_infos, names=None): names = self._verify_path_args(to_infos, from_infos, names) @@ -205,11 +141,11 @@ class RemoteHDFS(RemoteBase): if from_info['scheme'] != 'local': raise NotImplementedError - cmd = 'mkdir -p {}'.format(posixpath.dirname(to_info['url'])) + cmd = 'mkdir -p {}'.format(posixpath.dirname(to_info['path'])) self.hadoop_fs(cmd, user=to_info['user']) cmd = 'copyFromLocal {} {}'.format(from_info['path'], - to_info['url']) + to_info['path']) self.hadoop_fs(cmd, user=to_info['user']) def download(self, @@ -224,7 +160,7 @@ class RemoteHDFS(RemoteBase): raise NotImplementedError if to_info['scheme'] == 'hdfs': - self.cp(from_info, to_info) + self.copy(from_info, to_info) continue if to_info['scheme'] != 'local': @@ -234,36 +170,21 @@ class RemoteHDFS(RemoteBase): if not os.path.exists(dname): os.makedirs(dname) - cmd = 'copyToLocal {} {}'.format(from_info['url'], to_info['path']) + cmd = 'copyToLocal {} {}'.format(from_info['path'], + to_info['path']) self.hadoop_fs(cmd, user=from_info['user']) - def _path_to_checksum(self, path): - relpath = posixpath.relpath(path, self.url) - return posixpath.dirname(relpath) + posixpath.basename(relpath) + def list_cache_paths(self): + try: + self.hadoop_fs('test -e {}'.format(self.prefix)) + except RemoteHDFSCmdError: + return [] - def _all_checksums(self): - stdout = self.hadoop_fs('ls -R {}'.format(self.url)) + stdout = self.hadoop_fs('ls -R {}'.format(self.prefix)) lines = stdout.split('\n') flist = [] for line in lines: if not line.startswith('-'): continue flist.append(line.split()[-1]) - return [self._path_to_checksum(path) for path in flist] - - def gc(self, cinfos): - used = [info[self.PARAM_CHECKSUM] for info in cinfos['hdfs']] - used += [info[RemoteLOCAL.PARAM_MD5] for info in cinfos['local']] - - removed = False - for checksum in self._all_checksums(): - if checksum in used: - continue - path_info = {'scheme': 'hdfs', - 'user': self.user, - 'url': posixpath.join(self.url, - checksum[0:2], checksum[2:])} - self.remove(path_info) - removed = True - - return removed + return flist diff --git a/dvc/remote/http.py b/dvc/remote/http.py index 50bf1910c..d2d80a0cf 100644 --- a/dvc/remote/http.py +++ b/dvc/remote/http.py @@ -1,7 +1,6 @@ import os import threading import requests -import posixpath from dvc.logger import logger from dvc.progress import progress @@ -34,6 +33,7 @@ class RemoteHTTP(RemoteBase): self.project = project self.cache_dir = config.get(Config.SECTION_REMOTE_URL) self.url = self.cache_dir + self.path_info = {'scheme': 'http'} @property def prefix(self): @@ -53,7 +53,7 @@ class RemoteHTTP(RemoteBase): if to_info['scheme'] != 'local': raise NotImplementedError - msg = "Downloading '{}' to '{}'".format(from_info['url'], + msg = "Downloading '{}' to '{}'".format(from_info['path'], to_info['path']) logger.debug(msg) @@ -63,7 +63,7 @@ class RemoteHTTP(RemoteBase): self._makedirs(to_info['path']) - total = self._content_length(from_info['url']) + total = self._content_length(from_info['path']) if no_progress_bar or not total: cb = None @@ -71,9 +71,9 @@ class RemoteHTTP(RemoteBase): cb = ProgressBarCallback(name, total) try: - self._download_to(from_info['url'], tmp_file, callback=cb) + self._download_to(from_info['path'], tmp_file, callback=cb) except Exception as exc: - msg = "Failed to download '{}'".format(from_info['url']) + msg = "Failed to download '{}'".format(from_info['path']) logger.warn(msg, exc) continue @@ -82,26 +82,24 @@ class RemoteHTTP(RemoteBase): if not no_progress_bar: progress.finish_target(name) - def exists(self, path_infos): - return [ - bool(self._request('HEAD', path_info.get('url'))) - for path_info in path_infos - ] + def exists(self, path_info): + assert not isinstance(path_info, list) + assert path_info['scheme'] in ['http', 'https'] + return bool(self._request('HEAD', path_info.get('path'))) + + def cache_exists(self, md5s): + assert isinstance(md5s, list) + + def func(md5): + return bool(self._request('HEAD', self.checksum_to_path(md5))) + + return list(filter(func, md5s)) def save_info(self, path_info): if path_info['scheme'] not in ['http', 'https']: raise NotImplementedError - return {self.PARAM_ETAG: self._etag(path_info['url'])} - - def md5s_to_path_infos(self, md5s): - return [ - { - 'scheme': 'http', - 'url': posixpath.join(self.prefix, md5[0:2], md5[2:]), - } - for md5 in md5s - ] + return {self.PARAM_ETAG: self._etag(path_info['path'])} def _content_length(self, url): return self._request('HEAD', url).headers.get('Content-Length') @@ -141,3 +139,6 @@ class RemoteHTTP(RemoteBase): return requests.request(method, url, **kwargs) except requests.exceptions.RequestException: raise DvcException('Could not perform a {} request'.format(method)) + + def gc(self): + raise NotImplementedError diff --git a/dvc/remote/local.py b/dvc/remote/local.py index 369820fab..18c53f34e 100644 --- a/dvc/remote/local.py +++ b/dvc/remote/local.py @@ -7,9 +7,8 @@ import shutil import posixpath from operator import itemgetter -import dvc.prompt as prompt from dvc.system import System -from dvc.remote.base import RemoteBase, STATUS_MAP +from dvc.remote.base import RemoteBase, STATUS_MAP, STATUS_NEW, STATUS_DELETED from dvc.logger import logger from dvc.utils import remove, move, copyfile, dict_md5, to_chunks from dvc.utils import LARGE_DIR_SIZE @@ -22,7 +21,7 @@ from concurrent.futures import ThreadPoolExecutor class RemoteLOCAL(RemoteBase): scheme = '' REGEX = r'^(?P(/+|.:\\+).*)$' - PARAM_MD5 = 'md5' + PARAM_CHECKSUM = 'md5' PARAM_PATH = 'path' PARAM_RELPATH = 'relpath' MD5_DIR_SUFFIX = '.dir' @@ -53,6 +52,8 @@ class RemoteLOCAL(RemoteBase): if self.cache_dir is not None and not os.path.exists(self.cache_dir): os.mkdir(self.cache_dir) + self.path_info = {'scheme': 'local'} + @staticmethod def compat_config(config): ret = config.copy() @@ -68,7 +69,7 @@ class RemoteLOCAL(RemoteBase): def prefix(self): return self.cache_dir - def all(self): + def list_cache_paths(self): clist = [] for entry in os.listdir(self.cache_dir): subdir = os.path.join(self.cache_dir, entry) @@ -76,8 +77,7 @@ class RemoteLOCAL(RemoteBase): continue for cache in os.listdir(subdir): - path = os.path.join(subdir, cache) - clist.append(self.path_to_md5(path)) + clist.append(os.path.join(subdir, cache)) return clist @@ -87,10 +87,6 @@ class RemoteLOCAL(RemoteBase): return os.path.join(self.cache_dir, md5[0:2], md5[2:]) - def path_to_md5(self, path): - relpath = os.path.relpath(path, self.cache_dir) - return os.path.dirname(relpath) + os.path.basename(relpath) - def changed_cache_file(self, md5): cache = self.get(md5) if self.state.changed(cache, md5=md5): @@ -101,6 +97,11 @@ class RemoteLOCAL(RemoteBase): return True return False + def exists(self, path_info): + assert not isinstance(path_info, list) + assert path_info['scheme'] == 'local' + return os.path.exists(path_info['path']) + def changed_cache(self, md5): cache = self.get(md5) clist = [(cache, md5)] @@ -117,7 +118,7 @@ class RemoteLOCAL(RemoteBase): continue for entry in self.load_dir_cache(md5): - md5 = entry[self.PARAM_MD5] + md5 = entry[self.PARAM_CHECKSUM] cache = self.get(md5) clist.append((cache, md5)) @@ -160,8 +161,14 @@ class RemoteLOCAL(RemoteBase): raise DvcException('No possible cache types left to try out.') + @property + def ospath(self): + if os.name == 'nt': + return ntpath + return posixpath + @classmethod - def ospath(cls, path): + def to_ospath(cls, path): if os.name == 'nt': return cls.ntpath(path) return cls.unixpath(path) @@ -204,7 +211,7 @@ class RemoteLOCAL(RemoteBase): md5 = self.state.update(path) dir_info.append({self.PARAM_RELPATH: relpath, - self.PARAM_MD5: md5}) + self.PARAM_CHECKSUM: md5}) if bar: progress.finish_target(title) @@ -237,7 +244,7 @@ class RemoteLOCAL(RemoteBase): return [] for info in d: - info['relpath'] = self.ospath(info['relpath']) + info['relpath'] = self.to_ospath(info['relpath']) return d @@ -262,36 +269,14 @@ class RemoteLOCAL(RemoteBase): def is_dir_cache(cls, cache): return cache.endswith(cls.MD5_DIR_SUFFIX) - def checkout(self, path_info, checksum_info, force=False): + def do_checkout(self, path_info, checksum, force=False): path = path_info['path'] - md5 = checksum_info.get(self.PARAM_MD5) + md5 = checksum cache = self.get(md5) - if not cache: - msg = 'No cache info for \'{}\'. Skipping checkout.' - logger.warn(msg.format(os.path.relpath(path))) - return - - if not self.changed(path_info, checksum_info): - msg = "Data '{}' didn't change." - logger.info(msg.format(os.path.relpath(path))) - return - - if self.changed_cache(md5): - msg = u'Cache \'{}\' not found. File \'{}\' won\'t be created.' - logger.warn(msg.format(md5, os.path.relpath(path))) - remove(path) - return - - msg = u'Checking out \'{}\' with cache \'{}\'.' - logger.info(msg.format(os.path.relpath(path), md5)) - if not self.is_dir_cache(cache): if os.path.exists(path): - if force or self._already_cached(path): - remove(path) - else: - self._safe_remove(path) + self.safe_remove(path_info, force=force) self.link(cache, path) self.state.update_link(path) @@ -311,20 +296,17 @@ class RemoteLOCAL(RemoteBase): for processed, entry in enumerate(dir_info): relpath = entry[self.PARAM_RELPATH] - m = entry[self.PARAM_MD5] + 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_MD5: m} + entry_checksum_info = {self.PARAM_CHECKSUM: m} if self.changed(entry_info, entry_checksum_info): if os.path.exists(p): - if force or self._already_cached(p): - remove(p) - else: - self._safe_remove(p) + self.safe_remove(entry_info, force=force) self.link(c, p) @@ -338,8 +320,10 @@ class RemoteLOCAL(RemoteBase): if bar: progress.finish_target(dir_relpath) - def _already_cached(self, path): - current_md5 = self.state.update(path) + def already_cached(self, path_info): + assert path_info['scheme'] in ['', 'local'] + + current_md5 = self.state.update(path_info['path']) if not current_md5: return False @@ -361,24 +345,7 @@ class RemoteLOCAL(RemoteBase): delta = working_dir_files - cached_files for file in delta: - if force or self._already_cached(file): - remove(file) - else: - self._safe_remove(file) - - def _safe_remove(self, file): - msg = ( - 'File "{}" is going to be removed. ' - 'Are you sure you want to proceed?' - .format(file) - ) - - if not prompt.confirm(msg): - raise DvcException('Unable to remove {} without a confirmation' - " from the user. Use '-f' to force." - .format(file)) - - remove(file) + self.safe_remove({'scheme': 'local', 'path': file}, force=force) def _move(self, inp, outp): # moving in two stages to make last the move atomic in @@ -402,7 +369,7 @@ class RemoteLOCAL(RemoteBase): self.link(cache, path) self.state.update_link(path) - return {self.PARAM_MD5: md5} + return {self.PARAM_CHECKSUM: md5} def _save_dir(self, path_info): path = path_info['path'] @@ -415,7 +382,7 @@ class RemoteLOCAL(RemoteBase): for processed, entry in enumerate(dir_info): relpath = entry[self.PARAM_RELPATH] - m = entry[self.PARAM_MD5] + m = entry[self.PARAM_CHECKSUM] p = os.path.join(path, relpath) c = self.get(m) @@ -434,7 +401,7 @@ class RemoteLOCAL(RemoteBase): if bar: progress.finish_target(dir_relpath) - return {self.PARAM_MD5: md5} + return {self.PARAM_CHECKSUM: md5} def save(self, path_info): if path_info['scheme'] != 'local': @@ -455,7 +422,7 @@ class RemoteLOCAL(RemoteBase): if path_info['scheme'] != 'local': raise NotImplementedError - return {self.PARAM_MD5: self.state.update(path_info['path'])} + return {self.PARAM_CHECKSUM: self.state.update(path_info['path'])} def changed(self, path_info, checksum_info): """ @@ -465,10 +432,10 @@ class RemoteLOCAL(RemoteBase): - The checkusm stored in the State is different from the given one - There's no file in the cache """ - if not self.exists([path_info])[0]: + if not self.exists(path_info): return True - md5 = checksum_info.get(self.PARAM_MD5, None) + md5 = checksum_info.get(self.PARAM_CHECKSUM, None) if md5 is None: return True @@ -489,17 +456,9 @@ class RemoteLOCAL(RemoteBase): move(from_info['path'], to_info['path']) - def md5s_to_path_infos(self, md5s): - return [{'scheme': 'local', - 'path': os.path.join(self.prefix, - md5[0:2], md5[2:])} for md5 in md5s] - - def exists(self, path_infos): - ret = [] - for path_info in path_infos: - assert path_info['scheme'] == 'local' - ret.append(os.path.exists(path_info['path'])) - return ret + 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): names = self._verify_path_args(to_infos, from_infos, names) @@ -565,10 +524,10 @@ class RemoteLOCAL(RemoteBase): by_md5 = {} for info in checksum_infos: - md5 = info[self.PARAM_MD5] + md5 = info[self.PARAM_CHECKSUM] if show_checksums: - by_md5[md5] = md5 + by_md5[md5] = {'name': md5} continue name = info[self.PARAM_PATH] @@ -577,187 +536,136 @@ class RemoteLOCAL(RemoteBase): name += '({})'.format(branch) if md5 not in by_md5.keys(): - by_md5[md5] = '' + by_md5[md5] = {'name': name} else: - by_md5[md5] += ' ' + by_md5[md5]['name'] += ' ' + name - by_md5[md5] += name - - return list(by_md5.keys()), list(by_md5.values()) - - def gc(self, checksum_infos): - checksum_infos = checksum_infos['local'] - used_md5s = [info[self.PARAM_MD5] for info in checksum_infos] - - removed = False - for md5 in self.all(): - if md5 in used_md5s: - continue - remove(self.get(md5)) - removed = True - - return removed + return by_md5 def status(self, checksum_infos, remote, jobs=None, show_checksums=False): - logger.info("Preparing to pull data from {}".format(remote.url)) + logger.info("Preparing to collect status from {}".format(remote.url)) title = "Collecting information" + ret = {} + progress.set_n_total(1) progress.update_target(title, 0, 100) progress.update_target(title, 10, 100) - md5s, names = self._group(checksum_infos, - show_checksums=show_checksums) - - progress.update_target(title, 20, 100) - - path_infos = remote.md5s_to_path_infos(md5s) + ret = self._group(checksum_infos, show_checksums=show_checksums) + md5s = list(ret.keys()) progress.update_target(title, 30, 100) - remote_exists = remote.exists(path_infos) + remote_exists = list(remote.cache_exists(md5s)) progress.update_target(title, 90, 100) - local_exists = [not self.changed_cache_file(md5) for md5 in md5s] + local_exists = self.cache_exists(md5s) progress.finish_target(title) - return [(name, STATUS_MAP[l, r]) for name, l, r in zip(names, - local_exists, - remote_exists)] + for md5, info in ret.items(): + info['status'] = STATUS_MAP[(md5 in local_exists, + md5 in remote_exists)] + + return ret + + def _get_chunks(self, download, remote, status_info, status, jobs): + title = "Analysing status." + + progress.set_n_total(1) + total = len(status_info) + current = 0 + + cache = [] + path_infos = [] + names = [] + for md5, info in status_info.items(): + if info['status'] == status: + cache.append(self.checksum_to_path_info(md5)) + path_infos.append(remote.checksum_to_path_info(md5)) + names.append(info['name']) + current += 1 + progress.update_target(title, current, total) + + progress.finish_target(title) + + progress.set_n_total(len(names)) + + if download: + to_infos = cache + from_infos = path_infos + else: + to_infos = path_infos + from_infos = cache + + return list(zip(to_chunks(from_infos, jobs), + to_chunks(to_infos, jobs), + to_chunks(names, jobs))) + + def _process(self, + checksum_infos, + remote, + jobs=None, + show_checksums=False, + download=False): + msg = "Preparing to {} data {} '{}'" + logger.info(msg.format('download' if download else 'upload', + 'from' if download else 'to', + remote.url)) + + if download: + func = remote.download + status = STATUS_DELETED + else: + func = remote.upload + status = STATUS_NEW + + if jobs is None: + jobs = remote.JOBS + + status_info = self.status(checksum_infos, + remote, + jobs=jobs, + show_checksums=show_checksums) + + chunks = self._get_chunks(download, remote, status_info, status, jobs) + + if len(chunks) == 0: + return + + futures = [] + with ThreadPoolExecutor(max_workers=jobs) as executor: + for from_infos, to_infos, names in chunks: + res = executor.submit(func, + from_infos, + to_infos, + names=names) + futures.append(res) + + for f in futures: + f.result() + + def push(self, + checksum_infos, + remote, + jobs=None, + show_checksums=False): + self._process(checksum_infos, + remote, + jobs=jobs, + show_checksums=show_checksums, + download=False) def pull(self, checksum_infos, remote, jobs=None, show_checksums=False): - title = "Collecting information" - - progress.set_n_total(1) - progress.update_target(title, 0, 100) - - grouped = zip(*self._group(checksum_infos, - show_checksums=show_checksums)) - - progress.update_target(title, 10, 100) - - md5s = [] - names = [] - # NOTE: filter files that are not corrupted - for md5, name in grouped: - if self.changed_cache_file(md5): - md5s.append(md5) - names.append(name) - - progress.update_target(title, 30, 100) - - cache = [{'scheme': 'local', 'path': self.get(md5)} for md5 in md5s] - - progress.update_target(title, 50, 100) - - path_infos = remote.md5s_to_path_infos(md5s) - - progress.update_target(title, 60, 100) - - # NOTE: dummy call to try to establish a connection - # to see if we need to ask user for a password. - remote.exists(remote.md5s_to_path_infos(['000'])) - - progress.update_target(title, 70, 100) - - assert len(path_infos) == len(cache) == len(md5s) == len(names) - - if jobs is None: - jobs = remote.JOBS - - chunks = list(zip(to_chunks(path_infos, jobs), - to_chunks(cache, jobs), - to_chunks(names, jobs))) - - progress.finish_target(title) - - progress.set_n_total(len(names)) - - if len(chunks) == 0: - return - - futures = [] - with ThreadPoolExecutor(max_workers=len(chunks)) as executor: - for from_infos, to_infos, names in chunks: - res = executor.submit(remote.download, - from_infos, - to_infos, - names=names) - futures.append(res) - - for f in futures: - f.result() - - def push(self, checksum_infos, remote, jobs=None, show_checksums=False): - logger.info("Preparing to push data to {}".format(remote.url)) - title = "Collecting information" - - progress.set_n_total(1) - progress.update_target(title, 0, 100) - - # NOTE: verifying that our cache is not corrupted - def func(info): - return not self.changed_cache_file(info[self.PARAM_MD5]) - checksum_infos = list(filter(func, checksum_infos)) - - progress.update_target(title, 20, 100) - - # NOTE: filter files that are already uploaded - md5s = [i[self.PARAM_MD5] for i in checksum_infos] - exists = remote.exists(remote.md5s_to_path_infos(md5s)) - - progress.update_target(title, 30, 100) - - def func(entry): - return not entry[0] - - assert len(exists) == len(checksum_infos) - infos_exist = list(filter(func, zip(exists, checksum_infos))) - checksum_infos = [i for e, i in infos_exist] - - progress.update_target(title, 70, 100) - - md5s, names = self._group(checksum_infos, - show_checksums=show_checksums) - cache = [{'scheme': 'local', 'path': self.get(md5)} for md5 in md5s] - - progress.update_target(title, 80, 100) - - path_infos = remote.md5s_to_path_infos(md5s) - - assert len(path_infos) == len(cache) == len(md5s) == len(names) - - progress.update_target(title, 90, 100) - - if jobs is None: - jobs = remote.JOBS - - chunks = list(zip(to_chunks(path_infos, jobs), - to_chunks(cache, jobs), - to_chunks(names, jobs))) - - progress.finish_target(title) - - progress.set_n_total(len(names)) - - if len(chunks) == 0: - return - - futures = [] - with ThreadPoolExecutor(max_workers=len(chunks)) as executor: - for to_infos, from_infos, names in chunks: - res = executor.submit(remote.upload, - from_infos, - to_infos, - names=names) - futures.append(res) - - for f in futures: - f.result() + self._process(checksum_infos, + remote, + jobs=jobs, + show_checksums=show_checksums, + download=True) diff --git a/dvc/remote/s3.py b/dvc/remote/s3.py index 3242f42a9..3ca2ac5e9 100644 --- a/dvc/remote/s3.py +++ b/dvc/remote/s3.py @@ -1,6 +1,5 @@ import os import threading -import posixpath try: import boto3 @@ -16,7 +15,6 @@ from dvc.logger import logger from dvc.progress import progress from dvc.config import Config from dvc.remote.base import RemoteBase -from dvc.remote.local import RemoteLOCAL from dvc.exceptions import DvcException @@ -37,7 +35,7 @@ class RemoteS3(RemoteBase): scheme = 's3' REGEX = r'^s3://(?P.*)$' REQUIRES = {'boto3': boto3} - PARAM_ETAG = 'etag' + PARAM_CHECKSUM = 'etag' def __init__(self, project, config): self.project = project @@ -66,6 +64,12 @@ class RemoteS3(RemoteBase): if shared_creds: os.environ.setdefault('AWS_SHARED_CREDENTIALS_FILE', shared_creds) + parsed = urlparse(self.url) + self.bucket = parsed.netloc + self.prefix = parsed.path.lstrip('/') + + self.path_info = {'scheme': self.scheme, 'bucket': self.bucket} + @staticmethod def compat_config(config): ret = config.copy() @@ -73,14 +77,6 @@ class RemoteS3(RemoteBase): ret[Config.SECTION_REMOTE_URL] = url return ret - @property - def bucket(self): - return urlparse(self.url).netloc - - @property - def prefix(self): - return urlparse(self.url).path.lstrip('/') - @property def s3(self): session = boto3.session.Session(profile_name=self.profile, @@ -90,11 +86,12 @@ class RemoteS3(RemoteBase): endpoint_url=self.endpoint_url, use_ssl=self.use_ssl) - def get_etag(self, bucket, key): + def get_etag(self, bucket, path): try: - obj = self.s3.head_object(Bucket=bucket, Key=key) + obj = self.s3.head_object(Bucket=bucket, Key=path) except Exception: - raise DvcException('s3://{}/{} does not exist'.format(bucket, key)) + raise DvcException('s3://{}/{} does not exist'.format(bucket, + path)) return obj['ETag'].strip('"') @@ -102,14 +99,14 @@ class RemoteS3(RemoteBase): if path_info['scheme'] != 's3': raise NotImplementedError - return {self.PARAM_ETAG: self.get_etag(path_info['bucket'], - path_info['key'])} + return {self.PARAM_CHECKSUM: self.get_etag(path_info['bucket'], + path_info['path'])} def changed(self, path_info, checksum_info): - if not self.exists([path_info])[0]: + if not self.exists(path_info): return True - etag = checksum_info.get(self.PARAM_ETAG, None) + etag = checksum_info.get(self.PARAM_CHECKSUM, None) if etag is None: return True @@ -118,94 +115,38 @@ class RemoteS3(RemoteBase): return checksum_info != self.save_info(path_info) - def _copy(self, from_info, to_info, s3=None): + def copy(self, from_info, to_info, s3=None): s3 = s3 if s3 else self.s3 source = {'Bucket': from_info['bucket'], - 'Key': from_info['key']} - self.s3.copy(source, to_info['bucket'], to_info['key']) + 'Key': from_info['path']} + self.s3.copy(source, to_info['bucket'], to_info['path']) def save(self, path_info): if path_info['scheme'] != 's3': raise NotImplementedError - etag = self.get_etag(path_info['bucket'], path_info['key']) - key = posixpath.join(self.prefix, etag[0:2], etag[2:]) - to_info = {'scheme': 's3', 'bucket': self.bucket, 'key': key} + etag = self.get_etag(path_info['bucket'], path_info['path']) + path = self.checksum_to_path(etag) + to_info = {'scheme': 's3', 'bucket': self.bucket, 'path': path} - self._copy(path_info, to_info) + self.copy(path_info, to_info) - return {self.PARAM_ETAG: etag} - - @staticmethod - def to_string(path_info): - return "s3://{}/{}".format(path_info['bucket'], path_info['key']) - - def changed_cache(self, etag): - key = posixpath.join(self.prefix, etag[0:2], etag[2:]) - cache = {'scheme': 's3', 'bucket': self.bucket, 'key': key} - - if {self.PARAM_ETAG: etag} != self.save_info(cache): - if self.exists([cache])[0]: - msg = 'Corrupted cache file {}' - logger.warn(msg.format(self.to_string(cache))) - self.remove(cache) - return True - - return False - - def checkout(self, path_info, checksum_info): - if path_info['scheme'] != 's3': - raise NotImplementedError - - etag = checksum_info.get(self.PARAM_ETAG, None) - if not etag: - return - - if not self.changed(path_info, checksum_info): - msg = "Data '{}' didn't change." - logger.info(msg.format(self.to_string(path_info))) - return - - if self.changed_cache(etag): - msg = "Cache '{}' not found. File '{}' won't be created." - logger.warn(msg.format(etag, self.to_string(path_info))) - return - - if self.exists([path_info])[0]: - msg = "Data '{}' exists. Removing before checkout." - logger.warn(msg.format(self.to_string(path_info))) - self.remove(path_info) - return - - msg = "Checking out '{}' with cache '{}'." - logger.info(msg.format(self.to_string(path_info), etag)) - - key = posixpath.join(self.prefix, etag[0:2], etag[2:]) - from_info = {'scheme': 's3', 'bucket': self.bucket, 'key': key} - - self._copy(from_info, path_info) + return {self.PARAM_CHECKSUM: etag} def remove(self, path_info): if path_info['scheme'] != 's3': raise NotImplementedError logger.debug('Removing s3://{}/{}'.format(path_info['bucket'], - path_info['key'])) + path_info['path'])) self.s3.delete_object(Bucket=path_info['bucket'], - Key=path_info['key']) + Key=path_info['path']) - def md5s_to_path_infos(self, md5s): - return [{'scheme': self.scheme, - 'bucket': self.bucket, - 'key': posixpath.join(self.prefix, - md5[0:2], md5[2:])} for md5 in md5s] - - def _all_keys(self, bucket, prefix): + def _list_paths(self, bucket, prefix): s3 = self.s3 - keys = [] kwargs = {'Bucket': bucket, 'Prefix': prefix} while True: @@ -219,7 +160,7 @@ class RemoteS3(RemoteBase): break for obj in contents: - keys.append(obj['Key']) + yield obj['Key'] token = resp.get('NextContinuationToken', None) if not token: @@ -227,26 +168,15 @@ class RemoteS3(RemoteBase): kwargs['ContinuationToken'] = token - return keys + def list_cache_paths(self): + return self._list_paths(self.bucket, self.prefix) - def exists(self, path_infos): - ret = [] + def exists(self, path_info): + assert not isinstance(path_info, list) + assert path_info['scheme'] == 's3' - if len(path_infos) == 0: - return ret - - bucket = path_infos[0]['bucket'] - - for path_info in path_infos: - assert path_info['scheme'] == self.scheme - assert path_info['bucket'] == bucket - exists = False - key = path_info['key'] - if key in self._all_keys(bucket, key): - exists = True - ret.append(exists) - - return ret + 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): names = self._verify_path_args(to_infos, from_infos, names) @@ -262,7 +192,7 @@ class RemoteS3(RemoteBase): logger.debug("Uploading '{}' to '{}/{}'".format(from_info['path'], to_info['bucket'], - to_info['key'])) + to_info['path'])) if not name: name = os.path.basename(from_info['path']) @@ -273,7 +203,7 @@ class RemoteS3(RemoteBase): try: s3.upload_file(from_info['path'], to_info['bucket'], - to_info['key'], + to_info['path'], Callback=cb) except Exception as exc: msg = "Failed to upload '{}'".format(from_info['path']) @@ -296,14 +226,14 @@ class RemoteS3(RemoteBase): raise NotImplementedError if to_info['scheme'] == 's3': - self._copy(from_info, to_info, s3=s3) + self.copy(from_info, to_info, s3=s3) continue if to_info['scheme'] != 'local': raise NotImplementedError msg = "Downloading '{}/{}' to '{}'".format(from_info['bucket'], - from_info['key'], + from_info['path'], to_info['path']) logger.debug(msg) @@ -319,16 +249,16 @@ class RemoteS3(RemoteBase): else: total = s3.head_object( Bucket=from_info['bucket'], - Key=from_info['key'])['ContentLength'] + Key=from_info['path'])['ContentLength'] cb = Callback(name, total) s3.download_file(from_info['bucket'], - from_info['key'], + from_info['path'], tmp_file, Callback=cb) except Exception as exc: msg = "Failed to download '{}/{}'".format(from_info['bucket'], - from_info['key']) + from_info['path']) logger.warn(msg, exc) continue @@ -336,28 +266,3 @@ class RemoteS3(RemoteBase): if not no_progress_bar: progress.finish_target(name) - - def _path_to_etag(self, path): - relpath = posixpath.relpath(path, self.prefix) - return posixpath.dirname(relpath) + posixpath.basename(relpath) - - def _all(self): - keys = self._all_keys(self.bucket, self.prefix) - return [self._path_to_etag(key) for key in keys] - - def gc(self, cinfos): - used_etags = [info[self.PARAM_ETAG] for info in cinfos['s3']] - used_etags += [info[RemoteLOCAL.PARAM_MD5] for info in cinfos['local']] - - removed = False - for etag in self._all(): - if etag in used_etags: - continue - path_info = {'scheme': 's3', - 'key': posixpath.join(self.prefix, - etag[0:2], etag[2:]), - 'bucket': self.bucket} - self.remove(path_info) - removed = True - - return removed diff --git a/dvc/remote/ssh.py b/dvc/remote/ssh.py index 755468728..8d576cad1 100644 --- a/dvc/remote/ssh.py +++ b/dvc/remote/ssh.py @@ -11,7 +11,6 @@ import dvc.prompt as prompt from dvc.logger import logger from dvc.progress import progress from dvc.remote.base import RemoteBase, RemoteBaseCmdError -from dvc.remote.local import RemoteLOCAL from dvc.config import Config from dvc.exceptions import DvcException @@ -52,7 +51,7 @@ class RemoteSSH(RemoteBase): JOBS = 4 - PARAM_MD5 = 'md5' + PARAM_CHECKSUM = 'md5' DEFAULT_PORT = 22 TIMEOUT = 1800 @@ -73,13 +72,10 @@ class RemoteSSH(RemoteBase): self.ask_password = config.get(Config.SECTION_REMOTE_ASK_PASSWORD, False) - def md5s_to_path_infos(self, md5s): - return [{'scheme': 'ssh', - 'host': self.host, - 'user': self.user, - 'port': self.port, - 'path': posixpath.join(self.prefix, - md5[0:2], md5[2:])} for md5 in md5s] + self.path_info = {'scheme': 'ssh', + 'host': self.host, + 'user': self.user, + 'port': self.port} def ssh(self, host=None, user=None, port=None): msg = "Establishing ssh connection with '{}' " \ @@ -105,33 +101,20 @@ class RemoteSSH(RemoteBase): return ssh - def exists(self, path_infos): - ret = [] + def exists(self, path_info): + assert not isinstance(path_info, list) + assert path_info['scheme'] == 'ssh' - if len(path_infos) == 0: - return ret - - host = path_infos[0]['host'] - user = path_infos[0]['user'] - port = path_infos[0]['port'] - ssh = self.ssh(host=host, - user=user, - port=port) - - for path_info in path_infos: - assert host == path_info['host'] - assert user == path_info['user'] - assert port == path_info['port'] + with self.ssh(path_info['host'], + path_info['user'], + path_info['port']) as ssh: try: self._exec(ssh, 'test -e {}'.format(path_info['path'])) exists = True except RemoteSSHCmdError: exists = False - ret.append(exists) - ssh.close() - - return ret + return exists def _exec(self, ssh, cmd): stdin, stdout, stderr = ssh.exec_command(cmd) @@ -205,7 +188,7 @@ class RemoteSSH(RemoteBase): return md5 - def cp(self, from_info, to_info, ssh=None): + def copy(self, from_info, to_info, ssh=None): if from_info['scheme'] != 'ssh' or to_info['scheme'] != 'ssh': raise NotImplementedError @@ -227,37 +210,13 @@ class RemoteSSH(RemoteBase): if path_info['scheme'] != 'ssh': raise NotImplementedError - return {self.PARAM_MD5: self.md5(path_info)} - - @staticmethod - def to_string(path_info): - return "{}://{}@{}:{}".format(path_info['scheme'], - path_info['user'], - path_info['host'], - path_info['path']) - - def changed_cache(self, md5): - cache = {} - cache['scheme'] = 'ssh' - cache['host'] = self.host - cache['port'] = self.port - cache['user'] = self.user - cache['path'] = posixpath.join(self.prefix, md5[0:2], md5[2:]) - - if {self.PARAM_MD5: md5} != self.save_info(cache): - if self.exists([cache])[0]: - msg = 'Corrupted cache file {}' - logger.warn(msg.format(self.to_string(cache))) - self.remove(cache) - return True - - return False + return {self.PARAM_CHECKSUM: self.md5(path_info)} def changed(self, path_info, checksum_info): - if not self.exists([path_info])[0]: + if not self.exists(path_info): return True - md5 = checksum_info.get(self.PARAM_MD5, None) + md5 = checksum_info.get(self.PARAM_CHECKSUM, None) if md5 is None: return True @@ -272,43 +231,11 @@ class RemoteSSH(RemoteBase): md5 = self.md5(path_info) dest = path_info.copy() - dest['path'] = posixpath.join(self.prefix, md5[0:2], md5[2:]) + dest['path'] = self.checksum_to_path(md5) - self.cp(path_info, dest) + self.copy(path_info, dest) - return {self.PARAM_MD5: md5} - - def checkout(self, path_info, checksum_info): - if path_info['scheme'] != 'ssh': - raise NotImplementedError - - md5 = checksum_info.get(self.PARAM_MD5, None) - if not md5: - return - - if not self.changed(path_info, checksum_info): - msg = "Data '{}' didn't change." - logger.info(msg.format(self.to_string(path_info))) - return - - if self.changed_cache(md5): - msg = "Cache '{}' not found. File '{}' won't be created." - logger.warn(msg.format(md5, self.to_string(path_info))) - return - - if self.exists([path_info])[0]: - msg = "Data '{}' exists. Removing before checkout." - logger.warn(msg.format(self.to_string(path_info))) - self.remove(path_info) - return - - msg = "Checking out '{}' with cache '{}'." - logger.info(msg.format(self.to_string(path_info), md5)) - - src = path_info.copy() - src['path'] = posixpath.join(self.prefix, md5[0:2], md5[2:]) - - self.cp(src, path_info) + return {self.PARAM_CHECKSUM: md5} def remove(self, path_info): if path_info['scheme'] != 'ssh': @@ -343,7 +270,7 @@ class RemoteSSH(RemoteBase): assert from_info['host'] == to_info['host'] assert from_info['port'] == to_info['port'] assert from_info['user'] == to_info['user'] - self.cp(from_info, to_info, ssh=ssh) + self.copy(from_info, to_info, ssh=ssh) continue if to_info['scheme'] != 'local': @@ -416,11 +343,7 @@ class RemoteSSH(RemoteBase): sftp.close() ssh.close() - def _path_to_md5(self, path): - relpath = posixpath.relpath(path, self.prefix) - return posixpath.dirname(relpath) + posixpath.basename(relpath) - - def _all_md5s(self): + def list_cache_paths(self): ssh = self.ssh(host=self.host, user=self.user, port=self.port) @@ -428,24 +351,4 @@ class RemoteSSH(RemoteBase): stdout = self._exec(ssh, cmd) flist = stdout.split() ssh.close() - - return [self._path_to_md5(path) for path in flist] - - def gc(self, cinfos): - used = [info[self.PARAM_MD5] for info in cinfos['ssh']] - used += [info[RemoteLOCAL.PARAM_MD5] for info in cinfos['local']] - - removed = False - for md5 in self._all_md5s(): - if md5 in used: - continue - path_info = {'scheme': 'ssh', - 'user': self.user, - 'host': self.host, - 'port': self.port, - 'path': posixpath.join(self.prefix, - md5[0:2], md5[2:])} - self.remove(path_info) - removed = True - - return removed + return flist diff --git a/dvc/stage.py b/dvc/stage.py index e9a85341b..761378600 100644 --- a/dvc/stage.py +++ b/dvc/stage.py @@ -348,8 +348,8 @@ class Stage(object): new_d.pop(self.PARAM_MD5, None) outs = old_d.get(self.PARAM_OUTS, []) for out in outs: - out.pop(RemoteLOCAL.PARAM_MD5, None) - out.pop(RemoteS3.PARAM_ETAG, None) + out.pop(RemoteLOCAL.PARAM_CHECKSUM, None) + out.pop(RemoteS3.PARAM_CHECKSUM, None) return old_d == new_d diff --git a/tests/test_cache.py b/tests/test_cache.py index 7a9dd3a53..17fb37569 100644 --- a/tests/test_cache.py +++ b/tests/test_cache.py @@ -22,10 +22,10 @@ class TestCache(TestDvc): self.create(self.cache2, '2') def test_all(self): - md5_list = Cache(self.dvc).local.all() + md5_list = list(Cache(self.dvc).local.all()) self.assertEquals(len(md5_list), 2) - self.assertTrue(self.cache1_md5 in md5_list) - self.assertTrue(self.cache2_md5 in md5_list) + self.assertIn(self.cache1_md5, md5_list) + self.assertIn(self.cache2_md5, md5_list) def test_get(self): cache = Cache(self.dvc).local.get(self.cache1_md5) diff --git a/tests/test_checkout.py b/tests/test_checkout.py index da6d496c4..706b05236 100644 --- a/tests/test_checkout.py +++ b/tests/test_checkout.py @@ -81,7 +81,7 @@ class TestCheckoutCorruptedCacheDir(TestDvc): # NOTE: modifying cache file for one of the files inside the directory # to check if dvc will detect that the cache is corrupted. entry = self.dvc.cache.local.load_dir_cache(out.md5)[0] - md5 = entry[self.dvc.cache.local.PARAM_MD5] + md5 = entry[self.dvc.cache.local.PARAM_CHECKSUM] cache = self.dvc.cache.local.get(md5) with open(cache, 'w+') as fobj: @@ -281,8 +281,8 @@ class TestCheckoutMissingMd5InStageFile(TestRepro): with open(self.file1_stage, 'r') as fd: d = yaml.load(fd) - del(d[Stage.PARAM_OUTS][0][RemoteLOCAL.PARAM_MD5]) - del(d[Stage.PARAM_DEPS][0][RemoteLOCAL.PARAM_MD5]) + del(d[Stage.PARAM_OUTS][0][RemoteLOCAL.PARAM_CHECKSUM]) + del(d[Stage.PARAM_DEPS][0][RemoteLOCAL.PARAM_CHECKSUM]) with open(self.file1_stage, 'w') as fd: yaml.dump(d, fd) diff --git a/tests/test_data_cloud.py b/tests/test_data_cloud.py index a13872096..4b94f6407 100644 --- a/tests/test_data_cloud.py +++ b/tests/test_data_cloud.py @@ -233,10 +233,12 @@ class TestDataCloudBase(TestDvc): with self.cloud.project.state: # Check status status = self.cloud.status([info], show_checksums=True) - self.assertEqual([(md5, STATUS_NEW)], status) + expected = {md5: {'name': md5, 'status': STATUS_NEW}} + self.assertEqual(status, expected) status_dir = self.cloud.status([info_dir], show_checksums=True) - self.assertTrue((md5_dir, STATUS_NEW) in status_dir) + expected = {md5_dir: {'name': md5_dir, 'status': STATUS_NEW}} + self.assertEqual(status_dir, expected) # Push and check status self.cloud.push([info]) @@ -247,19 +249,23 @@ class TestDataCloudBase(TestDvc): self.assertTrue(os.path.isfile(cache_dir)) status = self.cloud.status([info], show_checksums=True) - self.assertEqual([(md5, STATUS_OK)], status) + expected = {md5: {'name': md5, 'status': STATUS_OK}} + self.assertEqual(status, expected) status_dir = self.cloud.status([info_dir], show_checksums=True) - self.assertTrue((md5_dir, STATUS_OK) in status_dir) + expected = {md5_dir: {'name': md5_dir, 'status': STATUS_OK}} + self.assertEqual(status_dir, expected) # Remove and check status shutil.rmtree(self.dvc.cache.local.cache_dir) status = self.cloud.status([info], show_checksums=True) - self.assertEqual([(md5, STATUS_DELETED)], status) + expected = {md5: {'name': md5, 'status': STATUS_DELETED}} + self.assertEqual(status, expected) status_dir = self.cloud.status([info_dir], show_checksums=True) - self.assertEqual([(md5_dir, STATUS_DELETED)], status_dir) + expected = {md5_dir: {'name': md5_dir, 'status': STATUS_DELETED}} + self.assertEqual(status_dir, expected) # Pull and check status self.cloud.pull([info]) @@ -272,10 +278,12 @@ class TestDataCloudBase(TestDvc): self.assertTrue(os.path.isfile(cache_dir)) status = self.cloud.status([info], show_checksums=True) - self.assertEqual([(md5, STATUS_OK)], status) + expected = {md5: {'name': md5, 'status': STATUS_OK}} + self.assertEqual(status, expected) status_dir = self.cloud.status([info_dir], show_checksums=True) - self.assertTrue((md5_dir, STATUS_OK) in status_dir) + expected = {md5_dir: {'name': md5_dir, 'status': STATUS_OK}} + self.assertTrue(status_dir, expected) def test(self): self._ensure_should_run() @@ -421,7 +429,7 @@ class TestDataCloudCLIBase(TestDvc): self.main(['fetch'] + args) - self.main(['pull'] + args) + self.main(['pull', '-f'] + args) self.assertTrue(os.path.exists(cache)) self.assertTrue(os.path.isfile(cache)) self.assertTrue(os.path.isfile(cache_dir)) diff --git a/tests/test_repro.py b/tests/test_repro.py index f6f60e637..460863ae0 100644 --- a/tests/test_repro.py +++ b/tests/test_repro.py @@ -544,8 +544,8 @@ class TestReproMissingMd5InStageFile(TestRepro): with open(self.file1_stage, 'r') as fd: d = yaml.load(fd) - del(d[Stage.PARAM_OUTS][0][RemoteLOCAL.PARAM_MD5]) - del(d[Stage.PARAM_DEPS][0][RemoteLOCAL.PARAM_MD5]) + del(d[Stage.PARAM_OUTS][0][RemoteLOCAL.PARAM_CHECKSUM]) + del(d[Stage.PARAM_DEPS][0][RemoteLOCAL.PARAM_CHECKSUM]) with open(self.file1_stage, 'w') as fd: yaml.dump(d, fd) diff --git a/tests/test_stage.py b/tests/test_stage.py index f439641c1..ba227c5d0 100644 --- a/tests/test_stage.py +++ b/tests/test_stage.py @@ -52,8 +52,10 @@ class TestSchemaDepsOuts(TestSchema): Stage.validate(d) def test_list(self): - lst = [{OutputLOCAL.PARAM_PATH: 'foo', RemoteLOCAL.PARAM_MD5: '123'}, - {OutputLOCAL.PARAM_PATH: 'bar', RemoteLOCAL.PARAM_MD5: None}, + lst = [{OutputLOCAL.PARAM_PATH: 'foo', + RemoteLOCAL.PARAM_CHECKSUM: '123'}, + {OutputLOCAL.PARAM_PATH: 'bar', + RemoteLOCAL.PARAM_CHECKSUM: None}, {OutputLOCAL.PARAM_PATH: 'baz'}] d = {Stage.PARAM_DEPS: lst} Stage.validate(d)