diff --git a/dvc/cache.py b/dvc/cache.py index dfa120d80..64c4204be 100644 --- a/dvc/cache.py +++ b/dvc/cache.py @@ -1,205 +1,44 @@ -import os -import json -import uuid -import shutil - -from dvc.state import State -from dvc.system import System -from dvc.logger import Logger -from dvc.utils import move, remove -from dvc.exceptions import DvcException +from dvc.remote.local import RemoteLOCAL +from dvc.remote.s3 import RemoteS3 +from dvc.remote.gs import RemoteGS +from dvc.remote.ssh import RemoteSSH class Cache(object): - CACHE_DIR = 'cache' - CACHE_TYPES = ['reflink', 'hardlink', 'symlink', 'copy'] - CACHE_TYPE_MAP = { - 'copy': shutil.copyfile, - 'symlink': System.symlink, - 'hardlink': System.hardlink, - 'reflink': System.reflink, - } + def __init__(self, project): + #FIXME + from dvc.config import Config - def __init__(self, root_dir, dvc_dir, link_state=None, cache_dir=None, cache_type=None): - self.cache_type = cache_type + config = project.config._config[Config.SECTION_CACHE] - cache_dir = cache_dir if cache_dir else self.CACHE_DIR - if os.path.isabs(cache_dir): - self.cache_dir = cache_dir + local = config.get(Config.SECTION_CACHE_LOCAL, None) + if local: + sect = project.config._config[Config.SECTION_REMOTE_FMT.format(local)] else: - self.cache_dir = os.path.abspath(os.path.realpath(os.path.join(dvc_dir, cache_dir))) + sect = {} + d = config.get(Config.SECTION_CACHE_DIR, None) + if d: + sect[Config.SECTION_REMOTE_URL] = d + t = config.get(Config.SECTION_CACHE_TYPE, None) + if t: + sect[Config.SECTION_CACHE_TYPE] = t - if not os.path.exists(self.cache_dir): - os.mkdir(self.cache_dir) + self.local = RemoteLOCAL(project, sect) - self.state = State(self.cache_dir) - self.link_state = link_state + self.s3 = None + s3 = config.get(Config.SECTION_CACHE_S3, None) + if s3: + sect = project.config._config[Config.SECTION_REMOTE_FMT.format(s3)] + self.s3 = RemoteS3(project, sect) - @staticmethod - def init(root_dir, dvc_dir, cache_dir=None): - return Cache(root_dir, dvc_dir, cache_dir=None) + self.gs = None + gs = config.get(Config.SECTION_CACHE_GS, None) + if gs: + sect = project.config._config[Config.SECTION_REMOTE_FMT.format(gs)] + self.gs = RemoteGS(project, sect) - def all(self): - clist = [] - for entry in os.listdir(self.cache_dir): - subdir = os.path.join(self.cache_dir, entry) - if not os.path.isdir(subdir): - continue - - for cache in os.listdir(subdir): - path = os.path.join(subdir, cache) - clist.append(path) - - return clist - - def get(self, md5): - if not md5: - return None - - 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(self, md5): - cache = self.get(md5) - if self.state.changed(cache, md5=md5): - if os.path.exists(cache): - Logger.warn('Corrupted cache file {}'.format(os.path.relpath(cache))) - remove(cache) - return True - - return False - - def link(self, src, link): - dname = os.path.dirname(link) - if not os.path.exists(dname): - os.makedirs(dname) - - if self.cache_type != None: - types = [self.cache_type] - else: - types = self.CACHE_TYPES - - for typ in types: - try: - self.CACHE_TYPE_MAP[typ](src, link) - self.link_state.update(link) - return - except Exception as exc: - msg = 'Cache type \'{}\' is not supported'.format(typ) - Logger.debug(msg) - if typ == types[-1]: - raise DvcException(msg, cause=exc) - - @staticmethod - def load_dir_cache(path): - if os.path.isabs(path): - relpath = os.path.relpath(path) - else: - relpath = path - - try: - with open(path, 'r') as fd: - d = json.load(fd) - except Exception as exc: - msg = u'Failed to load dir cache \'{}\'' - Logger.error(msg.format(relpath), exc) - return [] - - if not isinstance(d, list): - msg = u'Dir cache file format error \'{}\': skipping the file' - Logger.error(msg.format(relpath)) - return [] - - return d - - @staticmethod - def get_dir_cache(path): - res = {} - d = Cache.load_dir_cache(path) - - for entry in d: - res[entry[State.PARAM_RELPATH]] = entry[State.PARAM_MD5] - - return res - - def dir_cache(self, cache): - res = {} - dir_cache = self.get_dir_cache(cache) - - for relpath, md5 in dir_cache.items(): - res[relpath] = self.get(md5) - - return res - - @staticmethod - def is_dir_cache(cache): - return cache.endswith(State.MD5_DIR_SUFFIX) - - def checkout(self, path, md5): - cache = self.get(md5) - - if not cache or not os.path.exists(cache) or self.changed(md5): - if cache: - Logger.warn(u'\'{}({})\': cache file not found'.format(os.path.relpath(cache), - os.path.relpath(path))) - remove(path) - return - - if os.path.exists(path): - msg = u'Data \'{}\' exists. Removing before checkout' - Logger.debug(msg.format(os.path.relpath(path))) - remove(path) - - msg = u'Checking out \'{}\' with cache \'{}\'' - Logger.debug(msg.format(os.path.relpath(path), os.path.relpath(cache))) - - if not self.is_dir_cache(cache): - self.link(cache, path) - return - - dir_cache = self.dir_cache(cache) - for relpath, c in dir_cache.items(): - p = os.path.join(path, relpath) - self.link(c, p) - - def _move(self, inp, outp): - # moving in two stages to make last the move atomic in - # case inp and outp are in different filesystems - tmp = '{}.{}'.format(outp, str(uuid.uuid4())) - move(inp, tmp) - move(tmp, outp) - - def _save_file(self, path): - md5 = self.state.update(path) - cache = self.get(md5) - if self.changed(md5): - self._move(path, cache) - self.state.update(cache) - self.checkout(path, md5) - - def _save_dir(self, path): - md5 = self.state.update(path) - cache = self.get(md5) - dname = os.path.dirname(cache) - dir_info = self.state.collect_dir(path) - - for entry in dir_info: - relpath = entry[State.PARAM_RELPATH] - p = os.path.join(path, relpath) - - self._save_file(p) - - if not os.path.isdir(dname): - os.makedirs(dname) - - with open(cache, 'w+') as fd: - json.dump(dir_info, fd, sort_keys=True) - - def save(self, path): - if os.path.isdir(path): - self._save_dir(path) - else: - self._save_file(path) +# self.ssh = None +# ssh = config.get(Config.SECTION_CACHE_SSH, None) +# if ssh: +# sect = project.config._config[Config.SECTION_REMOTE_FMT.format(ssh)] +# self.ssh = RemoteSSH(project, sect) diff --git a/dvc/cloud/aws.py b/dvc/cloud/aws.py index 8dd1b2d4a..094a777bb 100644 --- a/dvc/cloud/aws.py +++ b/dvc/cloud/aws.py @@ -4,10 +4,9 @@ import threading import boto3 -from dvc.config import Config from dvc.logger import Logger from dvc.progress import progress -from dvc.cloud.base import DataCloudError, DataCloudBase +from dvc.cloud.base import DataCloudBase def sizeof_fmt(num, suffix='B'): @@ -57,9 +56,13 @@ class DataCloudAWS(DataCloudBase): @property def profile(self): + #FIXME + from dvc.config import Config return self._cloud_settings.cloud_config.get(Config.SECTION_AWS_PROFILE, None) def connect(self): + #FIXME + from dvc.cloud.base import DataCloudError session = boto3.Session(profile_name=self.profile) self.s3 = session.resource('s3') bucket = self.s3.Bucket(self.storage_bucket) diff --git a/dvc/cloud/base.py b/dvc/cloud/base.py index 21346e654..b44446ef6 100644 --- a/dvc/cloud/base.py +++ b/dvc/cloud/base.py @@ -4,8 +4,6 @@ import tempfile from dvc.logger import Logger from dvc.exceptions import DvcException -from dvc.config import Config, ConfigError -from dvc.cache import Cache STATUS_UNKNOWN = 0 @@ -47,6 +45,8 @@ class DataCloudBase(object): @property def url(self): + #FIXME + from dvc.config import Config config = self._cloud_settings.cloud_config return config.get(Config.SECTION_REMOTE_URL, None) @@ -72,6 +72,8 @@ class DataCloudBase(object): Precedence: Storage, then cloud specific """ + #FIXME + from dvc.config import Config if self._cloud_settings.global_storage_path: return self._cloud_settings.global_storage_path @@ -108,7 +110,7 @@ class DataCloudBase(object): def cache_file_key(self, fname): """ Key of a file within the bucket """ - relpath = os.path.relpath(fname, self._cloud_settings.cache.cache_dir) + relpath = os.path.relpath(fname, self._cloud_settings.cache.local.cache_dir) relpath = relpath.replace('\\', '/') return '{}/{}'.format(self.storage_prefix, relpath).strip('/') @@ -130,7 +132,10 @@ class DataCloudBase(object): path, local = arg ret = [path] - if not Cache.is_dir_cache(path): + #FIXME + from dvc.remote.local import RemoteLOCAL + + if not RemoteLOCAL.is_dir_cache(path): return ret if local: @@ -146,15 +151,15 @@ class DataCloudBase(object): self._pull_key(key, tmp, no_progress_bar=True) dir_path = tmp - for relpath, md5 in Cache.get_dir_cache(dir_path).items(): - cache = self._cloud_settings.cache.get(md5) + for relpath, md5 in RemoteLOCAL.get_dir_cache(dir_path).items(): + cache = self._cloud_settings.cache.local.get(md5) ret.append(cache) return ret def _cmp_checksum(self, blob, fname): - md5 = self._cloud_settings.cache.path_to_md5(fname) - if self._cloud_settings.cache.state.changed(fname, md5=md5): + md5 = self._cloud_settings.cache.local.path_to_md5(fname) + if self._cloud_settings.cache.local.state.changed(fname, md5=md5): return False return True diff --git a/dvc/cloud/local.py b/dvc/cloud/local.py index 0a3130596..0d1cabe4a 100644 --- a/dvc/cloud/local.py +++ b/dvc/cloud/local.py @@ -23,7 +23,7 @@ class DataCloudLOCAL(DataCloudBase): REGEX = r'^(?P(/+|.:\\+).*)$' def cache_file_key(self, path): - return os.path.relpath(os.path.abspath(path), self._cloud_settings.cache.cache_dir) + return os.path.relpath(os.path.abspath(path), self._cloud_settings.cache.local.cache_dir) def _get_key(self, path): key_name = self.cache_file_key(path) diff --git a/dvc/cloud/ssh.py b/dvc/cloud/ssh.py index 80f7eed89..91fa2c925 100644 --- a/dvc/cloud/ssh.py +++ b/dvc/cloud/ssh.py @@ -52,7 +52,7 @@ class DataCloudSSH(DataCloudBase): return self.ssh.open_sftp() def cache_file_key(self, path): - relpath = os.path.relpath(os.path.abspath(path), self._cloud_settings.cache.cache_dir) + relpath = os.path.relpath(os.path.abspath(path), self._cloud_settings.cache.local.cache_dir) return relpath.replace('\\', '/') def _isfile_remote(self, path): diff --git a/dvc/config.py b/dvc/config.py index 110a46d38..7de2c5127 100644 --- a/dvc/config.py +++ b/dvc/config.py @@ -5,7 +5,6 @@ import os import schema import configobj -from dvc.cache import Cache from dvc.exceptions import DvcException @@ -32,9 +31,17 @@ class Config(object): SECTION_CACHE = 'cache' SECTION_CACHE_DIR = 'dir' SECTION_CACHE_TYPE = 'type' - SECTION_CACHE_TYPE_SCHEMA = schema.And(schema.Use(str.lower), lambda t: t in Cache.CACHE_TYPE_MAP.keys()) + SECTION_CACHE_TYPE_SCHEMA = schema.And(schema.Use(str.lower), lambda t: t in ('reflink', 'hardlink', 'symlink', 'copy')) + SECTION_CACHE_LOCAL = 'local' + SECTION_CACHE_S3 = 's3' + SECTION_CACHE_GS = 'gs' SECTION_CACHE_SCHEMA = { - schema.Optional(SECTION_CACHE_DIR, default=Cache.CACHE_DIR): str, + schema.Optional(SECTION_CACHE_LOCAL): str, + schema.Optional(SECTION_CACHE_S3): str, + schema.Optional(SECTION_CACHE_GS): str, + + # backward compatibility + schema.Optional(SECTION_CACHE_DIR, default='cache'): str, schema.Optional(SECTION_CACHE_TYPE, default=None): SECTION_CACHE_TYPE_SCHEMA, } @@ -91,6 +98,7 @@ class Config(object): schema.Optional(SECTION_AWS_PROFILE, default='default'): str, schema.Optional(SECTION_AWS_CREDENTIALPATH, default = ''): str, schema.Optional(SECTION_GCP_PROJECTNAME): str, + schema.Optional(SECTION_CACHE_TYPE): SECTION_CACHE_TYPE_SCHEMA, } SCHEMA = { diff --git a/dvc/dependency/__init__.py b/dvc/dependency/__init__.py index b513e31dd..29094aa40 100644 --- a/dvc/dependency/__init__.py +++ b/dvc/dependency/__init__.py @@ -7,13 +7,17 @@ from dvc.dependency.s3 import DependencyS3 from dvc.dependency.gs import DependencyGS from dvc.dependency.local import DependencyLOCAL +#FIXME PARAM_MD5, PARAM_ETAG +#from dvc.remote.local import RemoteLOCAL +#from dvc.remote.s3 import RemoteS3 + DEPS = [DependencyS3, DependencyGS, DependencyLOCAL] SCHEMA = { DependencyBase.PARAM_PATH: str, - schema.Optional(DependencyLOCAL.PARAM_MD5): schema.Or(str, None), - schema.Optional(DependencyS3.PARAM_ETAG): schema.Or(str, None), + schema.Optional('md5'): schema.Or(str, None), + schema.Optional('etag'): schema.Or(str, None), } @@ -27,13 +31,13 @@ def _get(path): def loadd_from(stage, d_list): ret = [] for d in d_list: - p = d[DependencyBase.PARAM_PATH] - ret.append(_get(p)(stage, **d)) + p = d.pop(DependencyBase.PARAM_PATH) + ret.append(_get(p)(stage, p, d)) return ret def loads_from(stage, s_list): ret = [] for s in s_list: - ret.append(_get(s)(stage, s)) + ret.append(_get(s)(stage, s, {})) return ret diff --git a/dvc/dependency/base.py b/dvc/dependency/base.py index 31fd48437..246bc75b0 100644 --- a/dvc/dependency/base.py +++ b/dvc/dependency/base.py @@ -32,6 +32,8 @@ class DependencyBase(object): PARAM_PATH = 'path' def __init__(self, stage, path): + self.stage = stage + self.project = stage.project self.path = path @classmethod @@ -43,7 +45,7 @@ class DependencyBase(object): return cls.match(url) != None def changed(self): - raise NotImplemented + raise NotImplementedError def status(self): if self.changed(): @@ -52,7 +54,7 @@ class DependencyBase(object): return {} def save(self): - raise NotImplemented + raise NotImplementedError def dumpd(self): return {self.PARAM_PATH: self.path} diff --git a/dvc/dependency/gs.py b/dvc/dependency/gs.py index e297f452e..ebe398486 100644 --- a/dvc/dependency/gs.py +++ b/dvc/dependency/gs.py @@ -1,25 +1,21 @@ -from google.cloud import storage as gc - try: from urlparse import urlparse except ImportError: from urllib.parse import urlparse -from dvc.dependency.s3 import DependencyS3 from dvc.cloud.gcp import DataCloudGCP +from dvc.dependency.s3 import DependencyS3 +from dvc.remote.gs import RemoteGS +from dvc.config import Config class DependencyGS(DependencyS3): REGEX = DataCloudGCP.REGEX - @property - def client(self): - return gc.Client() - - def get_etag(self): - bucket = self.client.bucket(self.bucket) - blob = bucket.get_blob(self.key) - if not blob: - return None - - return blob.etag + def __init__(self, stage, path, info=None): + super(DependencyGS, self).__init__(stage, path) + self.info = info + self.remote = RemoteGS(stage.project, {Config.SECTION_REMOTE_URL: '/'}) + self.path_info = {'scheme': 'gs', + 'bucket': urlparse(path).netloc, + 'key': urlparse(path).path.lstrip('/')} diff --git a/dvc/dependency/local.py b/dvc/dependency/local.py index 847635658..f8029dd5b 100644 --- a/dvc/dependency/local.py +++ b/dvc/dependency/local.py @@ -12,45 +12,37 @@ from dvc.cloud.local import DataCloudLOCAL class DependencyLOCAL(DependencyBase): REGEX = r'^(?P(/+|.:\\+)?.*)$' - PARAM_PATH = 'path' - PARAM_MD5 = 'md5' - MD5_DIR_SUFFIX = '.dir' - DoesNotExistError = DependencyDoesNotExistError IsNotFileOrDirError = DependencyIsNotFileOrDirError - def __init__(self, stage, path, md5=None): + def __init__(self, stage, path, info=None): + #FIXME + from dvc.config import Config + from dvc.remote.local import RemoteLOCAL + self.stage = stage self.project = stage.project if not os.path.isabs(path): path = self.unixpath(path) path = os.path.join(stage.cwd, path) self.path = os.path.normpath(path) - self.md5 = md5 + self.info = info + self.remote = RemoteLOCAL(stage.project, + {Config.SECTION_REMOTE_URL: self.project.dvc_dir}) + self.path_info = {'scheme': 'local', + 'path': self.path} @property def rel_path(self): return os.path.relpath(self.path) - def _changed_md5(self): + def changed(self): if not os.path.exists(self.path): return True - return self.project.state.changed(self.path, self.md5) + info = self.remote.save_info(self.path_info) - @staticmethod - def _changed_msg(changed): - if changed: - return 'changed' - return "didn't change" - - def changed(self): - ret = self._changed_md5() - - msg = u'Dependency \'{}\' {}'.format(self.rel_path, self._changed_msg(ret)) - self.project.logger.debug(msg) - - return ret + return self.info != info def save(self): if not os.path.exists(self.path): @@ -63,7 +55,7 @@ class DependencyLOCAL(DependencyBase): (os.path.isdir(self.path) and len(os.listdir(self.path)) == 0): self.project.logger.warn("File/directory '{}' is empty.".format(self.rel_path)) - self.md5 = self.project.state.update(self.path) + self.info = self.remote.save_info(self.path_info) @staticmethod def unixpath(path): @@ -77,5 +69,6 @@ class DependencyLOCAL(DependencyBase): else: path = self.path - return {self.PARAM_PATH: path, - self.PARAM_MD5:self.md5} + info = self.info + info[self.PARAM_PATH] = path + return info diff --git a/dvc/dependency/s3.py b/dvc/dependency/s3.py index 52974057b..78beb4c89 100644 --- a/dvc/dependency/s3.py +++ b/dvc/dependency/s3.py @@ -1,5 +1,3 @@ -import boto3 - try: from urlparse import urlparse except ImportError: @@ -7,44 +5,28 @@ except ImportError: from dvc.dependency.base import DependencyBase from dvc.cloud.aws import DataCloudAWS +from dvc.remote.s3 import RemoteS3 +from dvc.config import Config class DependencyS3(DependencyBase): REGEX = DataCloudAWS.REGEX - PARAM_ETAG = 'etag' - - def __init__(self, stage, path, etag=None): + def __init__(self, stage, path, info=None): super(DependencyS3, self).__init__(stage, path) - self.etag = etag - - @property - def bucket(self): - return urlparse(self.path).netloc - - @property - def key(self): - return urlparse(self.path).path.lstrip('/') - - @property - def s3(self): - session = boto3.Session() - return session.resource('s3') - - def get_etag(self): - try: - obj = self.s3.Object(self.bucket, self.key).get() - except Exception: - return None - - return obj['ETag'].strip('"') + self.info = info + self.remote = RemoteS3(stage.project, {Config.SECTION_REMOTE_URL: '/'}) + self.path_info = {'scheme': 's3', + 'bucket': urlparse(path).netloc, + 'key': urlparse(path).path.lstrip('/')} def changed(self): - return self.etag != self.get_etag() + return self.info != self.remote.save_info(self.path_info) def save(self): - self.etag = self.get_etag() + self.info = self.remote.save_info(self.path_info) def dumpd(self): - return {self.PARAM_PATH: self.path, - self.PARAM_ETAG: self.etag} + ret = self.info + ret[self.PARAM_PATH] = self.path + return ret diff --git a/dvc/output/__init__.py b/dvc/output/__init__.py index 0153fee5f..0b402c83b 100644 --- a/dvc/output/__init__.py +++ b/dvc/output/__init__.py @@ -25,13 +25,14 @@ def _get(path): def loadd_from(stage, d_list): ret = [] for d in d_list: - p = d[DependencyBase.PARAM_PATH] - ret.append(_get(p)(stage, **d)) + p = d.pop(DependencyBase.PARAM_PATH) + cache = d.pop(OutputLOCAL.PARAM_CACHE, True) + ret.append(_get(p)(stage, p, info=d, cache=cache)) return ret -def loads_from(stage, s_list, use_cache=False): +def loads_from(stage, s_list, use_cache=True): ret = [] for s in s_list: - ret.append(_get(s)(stage, s, cache=use_cache)) + ret.append(_get(s)(stage, s, info={}, cache=use_cache)) return ret diff --git a/dvc/output/gs.py b/dvc/output/gs.py index fc84d78bb..f0d037824 100644 --- a/dvc/output/gs.py +++ b/dvc/output/gs.py @@ -4,19 +4,10 @@ from dvc.dependency.gs import DependencyGS class OutputGS(DependencyGS): PARAM_CACHE = 'cache' - def __init__(self, stage, path, etag=None, cache=True): - super(OutputGS, self).__init__(stage, path, etag=etag) + def __init__(self, stage, path, info=None, cache=True): + super(OutputGS, self).__init__(stage, path, info) self.use_cache = cache - @property - def cache(self): - if not self.use_cache: - return None - - raise NotImplementedError - - return self.project.cache.get(self.etag) - def dumpd(self): ret = super(OutputGS, self).dumpd() ret[self.PARAM_CACHE] = self.use_cache @@ -26,7 +17,7 @@ class OutputGS(DependencyGS): if super(OutputGS, self).changed(): return True - if self.use_cache and self.project.cache.changed(self.etag): + if self.use_cache and self.info != self.project.cache.gs.save_info(self.path_info): return True return False @@ -35,7 +26,7 @@ class OutputGS(DependencyGS): if not self.use_cache: return - raise NotImplementedError + self.project.cache.gs.checkout(self.path_info, self.info) def save(self): super(OutputGS, self).save() @@ -43,7 +34,8 @@ class OutputGS(DependencyGS): if not self.use_cache: return - raise NotImplementedError + + self.info = self.project.cache.gs.save(self.path_info) def remove(self): blob = self.client.bucket(self.bucket).get_blob(self.key) diff --git a/dvc/output/local.py b/dvc/output/local.py index e9dd2876a..262e2d642 100644 --- a/dvc/output/local.py +++ b/dvc/output/local.py @@ -13,19 +13,21 @@ from dvc.utils import remove class OutputLOCAL(DependencyLOCAL): PARAM_CACHE = 'cache' - DoesNotExistError = OutputDoesNotExistError - IsNotFileOrDirError = OutputIsNotFileOrDirError + DoesNotExistError = OutputDoesNotExistError + IsNotFileOrDirError = OutputIsNotFileOrDirError - def __init__(self, stage, path, md5=None, cache=True): - super(OutputLOCAL, self).__init__(stage, path, md5=md5) + def __init__(self, stage, path, info=None, cache=True): + super(OutputLOCAL, self).__init__(stage, path, info) self.use_cache = cache @property - def cache(self): - if not self.use_cache: - return None + def md5(self): + #FIXME + return self.info.get(self.project.cache.local.PARAM_MD5, None) - return self.project.cache.get(self.md5) + @property + def cache(self): + return self.project.cache.local.get(self.md5) def dumpd(self): ret = super(OutputLOCAL, self).dumpd() @@ -36,7 +38,7 @@ class OutputLOCAL(DependencyLOCAL): if super(OutputLOCAL, self).changed(): return True - if self.use_cache and self.project.cache.changed(self.md5): + if self.use_cache and self.info != self.project.cache.local.save_info(self.path_info): return True return False @@ -45,31 +47,27 @@ class OutputLOCAL(DependencyLOCAL): if not self.use_cache: return - msg = u'Checking out \'{}\' with cache \'{}\'' - self.project.logger.debug(msg.format(self.rel_path, self.md5)) - if not self.changed(): - msg = u'Data file \'{}\' with cache \'{}\' didn\'t change, skipping checkout.' - self.project.logger.debug(msg.format(self.rel_path, self.md5)) + msg = u'Data file \'{}\' didn\'t change, skipping checkout.' + self.project.logger.debug(msg.format(self.rel_path)) return - self.project.cache.checkout(self.path, self.md5) + self.project.cache.local.checkout(self.path_info, self.info) def save(self): + if not self.changed(): + msg = 'Output \'{}\' didn\'t change. Skipping saving.' + self.project.logger.debug(msg.format(self.rel_path)) + return + super(OutputLOCAL, self).save() if not self.use_cache: + msg = 'Output \'{}\' doesn\'t use cache. Skipping saving.' + self.project.logger.debug(msg.format(self.rel_path)) return - self.project.logger.debug(u'Saving \'{}\' to \'{}\''.format(self.rel_path, self.md5)) - - if self.project.scm.is_tracked(self.path): - raise OutputAlreadyTrackedError(self.rel_path) - - if not self.changed(): - return - - self.project.cache.save(self.path) + self.info = self.project.cache.local.save(self.path_info) def remove(self): remove(self.path) diff --git a/dvc/output/s3.py b/dvc/output/s3.py index 8c9fd4d29..12db2a8b7 100644 --- a/dvc/output/s3.py +++ b/dvc/output/s3.py @@ -4,19 +4,10 @@ from dvc.dependency.s3 import DependencyS3 class OutputS3(DependencyS3): PARAM_CACHE = 'cache' - def __init__(self, stage, path, etag=None, cache=True): - super(OutputS3, self).__init__(stage, path, etag=etag) + def __init__(self, stage, path, info=None, cache=True): + super(OutputS3, self).__init__(stage, path, info) self.use_cache = cache - @property - def cache(self): - if not self.use_cache: - return None - - raise NotImplementedError - - return self.project.cache.get(self.etag) - def dumpd(self): ret = super(OutputS3, self).dumpd() ret[self.PARAM_CACHE] = self.use_cache @@ -26,7 +17,7 @@ class OutputS3(DependencyS3): if super(OutputS3, self).changed(): return True - if self.use_cache and self.project.cache.changed(self.etag): + if self.use_cache and self.info != self.project.cache.s3.save_info(self.path_info): return True return False @@ -35,7 +26,7 @@ class OutputS3(DependencyS3): if not self.use_cache: return - raise NotImplementedError + self.project.cache.s3.checkout(self.path_info, self.info) def save(self): super(OutputS3, self).save() @@ -43,7 +34,7 @@ class OutputS3(DependencyS3): if not self.use_cache: return - raise NotImplementedError + self.info = self.project.cache.s3.save(self.path_info) def remove(self): self.s3.Object(self.bucket, self.key).delete() diff --git a/dvc/project.py b/dvc/project.py index 7708c6e2b..2718974a0 100644 --- a/dvc/project.py +++ b/dvc/project.py @@ -11,7 +11,7 @@ from dvc.logger import Logger from dvc.exceptions import DvcException from dvc.stage import Stage from dvc.config import Config -from dvc.state import State, LinkState +from dvc.state import LinkState from dvc.lock import Lock from dvc.scm import SCM, Base from dvc.cache import Cache @@ -48,11 +48,8 @@ class Project(object): self.scm = SCM(self.root_dir) self.lock = Lock(self.dvc_dir) self.link_state = LinkState(self.root_dir, self.dvc_dir) - self.cache = Cache(self.root_dir, self.dvc_dir, link_state=self.link_state, - cache_dir=self.config._config[Config.SECTION_CACHE].get(Config.SECTION_CACHE_DIR, None), - cache_type=self.config._config[Config.SECTION_CACHE].get(Config.SECTION_CACHE_TYPE, None)) - self.state = State(self.dvc_dir) self.logger = Logger(self.config._config[Config.SECTION_CORE].get(Config.SECTION_CORE_LOGLEVEL, None)) + self.cache = Cache(self) self.cloud = DataCloud(cache=self.cache, config=self.config._config) self.updater = Updater(self.dvc_dir) @@ -85,8 +82,6 @@ class Project(object): os.mkdir(dvc_dir) config = Config.init(dvc_dir) - Cache.init(root_dir, dvc_dir) - State.init(dvc_dir) proj = Project(root_dir) scm.add([config.config_file]) @@ -97,15 +92,13 @@ class Project(object): def _ignore(self): l = [self.link_state.state_file, - self.cache.link_state._lock_file.lock_file, - self.state.state_file, - self.state._lock_file.lock_file, + self.link_state._lock_file.lock_file, self.lock.lock_file, self.config.config_local_file, self.updater.updater_file] - if self.cache.cache_dir.startswith(self.root_dir): - l += [self.cache.cache_dir] + if self.cache.local.cache_dir.startswith(self.root_dir): + l += [self.cache.local.cache_dir] self.scm.ignore_list(l) @@ -212,15 +205,15 @@ class Project(object): if not out.use_cache or not out.cache: continue cache_set |= set([out.cache]) - if self.cache.is_dir_cache(out.cache) and os.path.isfile(out.cache): - dir_cache = self.cache.dir_cache(out.cache) + if self.cache.local.is_dir_cache(out.cache) and os.path.isfile(out.cache): + dir_cache = self.cache.local.dir_cache(out.cache) cache_set |= set(dir_cache.values()) return list(cache_set) def gc(self): clist = self._used_cache() - for cache in self.cache.all(): + for cache in self.cache.local.all(): if cache in clist: continue os.unlink(cache) @@ -262,7 +255,7 @@ class Project(object): cloud.STATUS_NEW: 'new', } - path = os.path.relpath(target, self.cache.cache_dir) + path = os.path.relpath(target, self.cache.local.cache_dir) status[path] = prefix_map[ret] diff --git a/dvc/remote/__init__.py b/dvc/remote/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/dvc/remote/base.py b/dvc/remote/base.py index 31a7bf96e..1906cb356 100644 --- a/dvc/remote/base.py +++ b/dvc/remote/base.py @@ -1,3 +1,12 @@ class Remote(object): - def __init__(self + def __init__(self, project, config): + pass + + def save_info(self, path_info): + pass + + def save(self, path_info): + pass + + def checkout(self, path_info, checksum_info): pass diff --git a/dvc/remote/gs.py b/dvc/remote/gs.py new file mode 100644 index 000000000..74b85c4c1 --- /dev/null +++ b/dvc/remote/gs.py @@ -0,0 +1,76 @@ +import posixpath +from google.cloud import storage + +try: + from urlparse import urlparse +except ImportError: + from urllib.parse import urlparse + +from dvc.remote.base import Remote + + +class RemoteGS(Remote): + PARAM_ETAG = 'etag' + + def __init__(self, project, config): + #FIXME + from dvc.config import Config + + self.project = project + self.url = config.get[Config.SECTION_REMOTE_URL] + self.projectname = config.get(Config.SECTION_GCP_PROJECTNAME, None) + + @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_etag(self, bucket, key): + blob = self.gs.bucket(bucket).get_blob(key) + if not blob: + return None + + return blob.etag + + def save_info(self, path_info): + if path_info['scheme'] != 'gs': + raise NotImplementedError + + return {self.PARAM_ETAG: self.get_etag(path_info['bucket'], path_info['key'])} + + def save(self, path_info): + if path_info['scheme'] != 'gs': + raise NotImplementedError + + etag = self.get_etag(path_info['bucket'], path_info['key']) + dest_key = posixpath.join(self.prefix, etag[0:2], etag[2:]) + + blob = self.gs.bucket(path_info['bucket']).get_blob(path_info['key']) + if not blob: + raise DvcException('{} doesn\'t exist in the cloud'.format(path_info['key'])) + + self.gs.bucket(self.bucket).copy_blob(blob, self.bucket, name=dest_key) + + return {PARAM_ETAG: etag} + + def checkout(self, path_info, checksum_info): + if path_info['scheme'] != 'gs': + raise NotImplementedError + + etag = checksum_info.get(PARAM_ETAG, None) + if not etag: + return + + key = posixpath.join(self.prefix, etag[0:2], etag[2:]) + blob = self.gs.bucket(self.bucket).get_blob(key) + if not blob: + raise DvcException('{} doesn\'t exist in the cloud'.format(key)) + + self.gs.bucket(path_info['bucket']).copy_blob(blob, path_info['bucket'], name=path_info['key']) diff --git a/dvc/remote/local.py b/dvc/remote/local.py new file mode 100644 index 000000000..1b3abe78e --- /dev/null +++ b/dvc/remote/local.py @@ -0,0 +1,242 @@ +import os +import uuid +import json +import shutil + +from dvc.system import System +from dvc.remote.base import Remote +from dvc.state import State +from dvc.logger import Logger +from dvc.utils import remove, move +from dvc.output.base import OutputAlreadyTrackedError + + +class RemoteLOCAL(Remote): + PARAM_MD5 = State.PARAM_MD5 + + CACHE_DIR = 'cache' + CACHE_TYPES = ['reflink', 'hardlink', 'symlink', 'copy'] + CACHE_TYPE_MAP = { + 'copy': shutil.copyfile, + 'symlink': System.symlink, + 'hardlink': System.hardlink, + 'reflink': System.reflink, + } + + def __init__(self, project, config): + #FIXME + from dvc.config import Config + + self.project = project + link_state = project.link_state + cache_dir = config.get(Config.SECTION_REMOTE_URL, self.CACHE_DIR) + cache_type = config.get(Config.SECTION_CACHE_TYPE, None) + + self.cache_type = cache_type + + cache_dir = cache_dir if cache_dir else self.CACHE_DIR + if os.path.isabs(cache_dir): + self.cache_dir = cache_dir + else: + self.cache_dir = os.path.abspath(os.path.realpath(os.path.join(project.dvc_dir, cache_dir))) + + if not os.path.exists(self.cache_dir): + os.mkdir(self.cache_dir) + + self.state = State(self.cache_dir) + self.link_state = link_state + + def all(self): + clist = [] + for entry in os.listdir(self.cache_dir): + subdir = os.path.join(self.cache_dir, entry) + if not os.path.isdir(subdir): + continue + + for cache in os.listdir(subdir): + path = os.path.join(subdir, cache) + clist.append(path) + + return clist + + def get(self, md5): + if not md5: + return None + + 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(self, md5): + cache = self.get(md5) + if self.state.changed(cache, md5=md5): + if os.path.exists(cache): + Logger.warn('Corrupted cache file {}'.format(os.path.relpath(cache))) + remove(cache) + return True + + return False + + def link(self, src, link): + dname = os.path.dirname(link) + if not os.path.exists(dname): + os.makedirs(dname) + + if self.cache_type != None: + types = [self.cache_type] + else: + types = self.CACHE_TYPES + + for typ in types: + try: + msg = u'Checking out \'{}\' with cache \'{}\'' + Logger.debug(msg.format(os.path.relpath(src), os.path.relpath(link))) + self.CACHE_TYPE_MAP[typ](src, link) + self.link_state.update(link) + return + except Exception as exc: + msg = 'Cache type \'{}\' is not supported'.format(typ) + Logger.debug(msg) + if typ == types[-1]: + raise DvcException(msg, cause=exc) + + @staticmethod + def load_dir_cache(path): + if os.path.isabs(path): + relpath = os.path.relpath(path) + else: + relpath = path + + try: + with open(path, 'r') as fd: + d = json.load(fd) + except Exception as exc: + msg = u'Failed to load dir cache \'{}\'' + Logger.error(msg.format(relpath), exc) + return [] + + if not isinstance(d, list): + msg = u'Dir cache file format error \'{}\': skipping the file' + Logger.error(msg.format(relpath)) + return [] + + return d + + @classmethod + def get_dir_cache(cls, path): + res = {} + + d = cls.load_dir_cache(path) + + for entry in d: + res[entry[State.PARAM_RELPATH]] = entry[State.PARAM_MD5] + + return res + + def dir_cache(self, cache): + res = {} + dir_cache = self.get_dir_cache(cache) + + for relpath, md5 in dir_cache.items(): + res[relpath] = self.get(md5) + + return res + + @staticmethod + def is_dir_cache(cache): + return cache.endswith(State.MD5_DIR_SUFFIX) + + def checkout(self, path_info, checksum_info): + path = path_info['path'] + md5 = checksum_info.get(self.PARAM_MD5, None) + cache = self.get(md5) + + if not cache or not os.path.exists(cache) or self.changed(md5): + if cache: + Logger.warn(u'\'{}({})\': cache file not found'.format(os.path.relpath(cache), + os.path.relpath(path))) + remove(path) + return + + if os.path.exists(path): + msg = u'Data \'{}\' exists. Removing before checkout' + Logger.debug(msg.format(os.path.relpath(path))) + remove(path) + + if not self.is_dir_cache(cache): + self.link(cache, path) + return + + msg = u'Checking out directory \'{}\' with cache \'{}\'' + Logger.debug(msg.format(os.path.relpath(path), os.path.relpath(cache))) + + dir_cache = self.dir_cache(cache) + for relpath, c in dir_cache.items(): + p = os.path.join(path, relpath) + self.link(c, p) + + def _move(self, inp, outp): + # moving in two stages to make last the move atomic in + # case inp and outp are in different filesystems + tmp = '{}.{}'.format(outp, str(uuid.uuid4())) + move(inp, tmp) + move(tmp, outp) + + def _save_file(self, path_info): + path = path_info['path'] + md5 = self.state.update(path) + cache = self.get(md5) + if self.changed(md5): + Logger.debug(u'Saving \'{}\' to \'{}\''.format(os.path.relpath(path), + os.path.relpath(cache))) + self._move(path, cache) + self.state.update(cache) + self.checkout(path_info, {self.PARAM_MD5: md5}) + + return {self.PARAM_MD5: md5} + + def _save_dir(self, path_info): + path = path_info['path'] + md5 = self.state.update(path) + cache = self.get(md5) + dname = os.path.dirname(cache) + dir_info = self.state.collect_dir(path) + + for entry in dir_info: + relpath = entry[State.PARAM_RELPATH] + p = os.path.join(path, relpath) + + self._save_file({'scheme': 'local', 'path': p}) + + if not os.path.isdir(dname): + os.makedirs(dname) + + Logger.debug(u'Saving directory \'{}\' to \'{}\''.format(os.path.relpath(path), + os.path.relpath(cache))) + + with open(cache, 'w+') as fd: + json.dump(dir_info, fd, sort_keys=True) + + return {self.PARAM_MD5: md5} + + def save(self, path_info): + if path_info['scheme'] != 'local': + raise NotImplementedError + + path = path_info['path'] + + if self.project.scm.is_tracked(path): + raise OutputAlreadyTrackedError(os.path.relpath(path)) + + if os.path.isdir(path): + return self._save_dir(path_info) + else: + return self._save_file(path_info) + + def save_info(self, path_info): + if path_info['scheme'] != 'local': + raise NotImplementedError + + return {self.PARAM_MD5: self.state.update(path_info['path'])} diff --git a/dvc/remote/s3.py b/dvc/remote/s3.py new file mode 100644 index 000000000..95cfe03f5 --- /dev/null +++ b/dvc/remote/s3.py @@ -0,0 +1,77 @@ +import boto3 +import posixpath + +try: + from urlparse import urlparse +except ImportError: + from urllib.parse import urlparse + +from dvc.remote.base import Remote + + +class RemoteS3(Remote): + PARAM_ETAG = 'etag' + + def __init__(self, project, config): + #FIXME + from dvc.config import Config + + + self.project = project + self.url = config[Config.SECTION_REMOTE_URL] + self.region = config.get(Config.SECTION_AWS_REGION, None) + self.profile = config.get(Config.SECTION_AWS_PROFILE, None) + self.credentialpath = config.get(Config.SECTION_AWS_CREDENTIALPATH, None) + + @property + def bucket(self): + return urlparse(self.url).netloc + + @property + def prefix(self): + return urlparse(self.url).path.lstrip('/') + + @property + def s3(self): + return boto3.resource('s3') + + def get_etag(self, bucket, key): + try: + obj = self.s3.Object(bucket, key).get() + except Exception: + return None + + return obj['ETag'].strip('"') + + def save_info(self, path_info): + if path_info['scheme'] != 's3': + raise NotImplementedError + + return {self.PARAM_ETAG: self.get_etag(path_info['bucket'], path_info['key'])} + + def save(self, path_info): + if path_info['scheme'] != 's3': + raise NotImplementedError + + etag = self.get_etag(path_info['bucket'], path_info['key']) + dest_key = posixpath.join(self.prefix, etag[0:2], etag[2:]) + + source = {'Bucket': path_info['bucket'], + 'Key': path_info['key']} + self.s3.Bucket(self.bucket).copy(source, dest_key) + + return {self.PARAM_ETAG: etag} + + 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 + + key = posixpath.join(self.prefix, etag[0:2], etag[2:]) + source = {'Bucket': self.bucket, + 'Key': key} + + self.s3.Bucket(path_info['bucket']).copy(source, path_info['key']) diff --git a/dvc/remote/ssh.py b/dvc/remote/ssh.py new file mode 100644 index 000000000..3e83c423e --- /dev/null +++ b/dvc/remote/ssh.py @@ -0,0 +1,5 @@ +from dvc.remote.base import Remote + + +class RemoteSSH(Remote): + pass diff --git a/dvc/stage.py b/dvc/stage.py index ca7aaec4d..86e5f5ab7 100644 --- a/dvc/stage.py +++ b/dvc/stage.py @@ -219,7 +219,7 @@ class Stage(object): for out in self.outs: out.save() - if out.use_cache: + if out.use_cache and out.path_info['scheme'] == 'local': self.project.scm.ignore(out.path) def run(self): diff --git a/dvc/transport/__init__.py b/dvc/transport/__init__.py deleted file mode 100644 index 08b401df7..000000000 --- a/dvc/transport/__init__.py +++ /dev/null @@ -1,21 +0,0 @@ -def Transport(path): - pass - -def exists(path): - return Transport(path).exists(path) - - -def copy(inp, outp): - return Transport(inp).copy(inp, outp) - - -def is_file(path): - return Transport(path).is_file(path) - - -def is_dir(path): - return Transport(path).is_dir(path) - - -def is_empty(path): - return Transport(path).is_empty(path) diff --git a/tests/test_add.py b/tests/test_add.py index 9711c0891..b1f0736d3 100644 --- a/tests/test_add.py +++ b/tests/test_add.py @@ -25,7 +25,7 @@ class TestAdd(TestDvc): self.assertEqual(len(stage.outs), 1) self.assertEqual(len(stage.deps), 0) self.assertEqual(stage.cmd, None) - self.assertEqual(stage.outs[0].md5, md5) + self.assertEqual(stage.outs[0].info['md5'], md5) class TestAddNonExistentFile(TestDvc): diff --git a/tests/test_cache.py b/tests/test_cache.py index b917b328a..5a82bc5b4 100644 --- a/tests/test_cache.py +++ b/tests/test_cache.py @@ -5,6 +5,7 @@ import tempfile from dvc.cache import Cache from dvc.system import System from dvc.main import main +from dvc.remote.local import RemoteLOCAL from tests.basic_env import TestDvc @@ -14,19 +15,23 @@ class TestCache(TestDvc): super(TestCache, self).setUp() self.cache1_md5 = '123' self.cache2_md5 = '234' - self.cache1 = os.path.join(self.dvc.cache.cache_dir, self.cache1_md5[0:2], self.cache1_md5[2:]) - self.cache2 = os.path.join(self.dvc.cache.cache_dir, self.cache2_md5[0:2], self.cache2_md5[2:]) + self.cache1 = os.path.join(self.dvc.cache.local.cache_dir, + self.cache1_md5[0:2], + self.cache1_md5[2:]) + self.cache2 = os.path.join(self.dvc.cache.local.cache_dir, + self.cache2_md5[0:2], + self.cache2_md5[2:]) self.create(self.cache1, '1') self.create(self.cache2, '2') def test_all(self): - flist = Cache(self.dvc.root_dir, self.dvc.dvc_dir).all() + flist = Cache(self.dvc).local.all() self.assertEquals(len(flist), 2) self.assertTrue(self.cache1 in flist) self.assertTrue(self.cache2 in flist) def test_get(self): - cache = Cache(self.dvc.root_dir, self.dvc.dvc_dir).get(self.cache1_md5) + cache = Cache(self.dvc).local.get(self.cache1_md5) self.assertEquals(cache, self.cache1) @@ -38,11 +43,11 @@ class TestCacheLoadBadDirCache(TestDvc): def test(self): fname = 'not-json' self.create(fname, 'not,json') - self._do_test(Cache.load_dir_cache(fname)) + self._do_test(RemoteLOCAL.load_dir_cache(fname)) fname = 'not-list' self.create(fname, '{"a": "b"}') - self._do_test(Cache.load_dir_cache(fname)) + self._do_test(RemoteLOCAL.load_dir_cache(fname)) class TestExternalCacheDir(TestDvc): diff --git a/tests/test_checkout.py b/tests/test_checkout.py index cd93baed8..f39263ac2 100644 --- a/tests/test_checkout.py +++ b/tests/test_checkout.py @@ -9,7 +9,7 @@ from dvc.main import main from tests.basic_env import TestDvc from tests.test_repro import TestRepro from dvc.stage import Stage -from dvc.output.local import OutputLOCAL +from dvc.remote.local import RemoteLOCAL class TestCheckout(TestRepro): @@ -163,8 +163,8 @@ class TestCheckoutMissingMd5InStageFile(TestRepro): with open(self.file1_stage, 'r') as fd: d = yaml.load(fd) - del(d[Stage.PARAM_OUTS][0][OutputLOCAL.PARAM_MD5]) - del(d[Stage.PARAM_DEPS][0][OutputLOCAL.PARAM_MD5]) + del(d[Stage.PARAM_OUTS][0][RemoteLOCAL.PARAM_MD5]) + del(d[Stage.PARAM_DEPS][0][RemoteLOCAL.PARAM_MD5]) 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 8a831c3b5..a25a523b1 100644 --- a/tests/test_data_cloud.py +++ b/tests/test_data_cloud.py @@ -185,7 +185,7 @@ class TestDataCloudBase(TestDvc): # Remove and check status sleep() - shutil.rmtree(self.dvc.cache.cache_dir) + shutil.rmtree(self.dvc.cache.local.cache_dir) status = self.cloud.status(cache) self.assertEqual(status, STATUS_DELETED) @@ -283,7 +283,7 @@ class TestDataCloudCLIBase(TestDvc): sleep() self.main(['status', '-c'] + args) - shutil.rmtree(self.dvc.cache.cache_dir) + shutil.rmtree(self.dvc.cache.local.cache_dir) sleep() self.main(['status', '-c'] + args) diff --git a/tests/test_gc.py b/tests/test_gc.py index e45e4fa6e..53ac95f9e 100644 --- a/tests/test_gc.py +++ b/tests/test_gc.py @@ -17,11 +17,11 @@ class TestGC(TestDvc): self.dvc.add(self.FOO) self.dvc.add(self.DATA_DIR) - self.good_cache = self.dvc.cache.all() + self.good_cache = self.dvc.cache.local.all() self.bad_cache = [] for i in ['123', '234', '345']: - path = os.path.join(self.dvc.cache.cache_dir, i[0:2], i[2:]) + path = os.path.join(self.dvc.cache.local.cache_dir, i[0:2], i[2:]) self.create(path, i) self.bad_cache.append(path) @@ -34,7 +34,7 @@ class TestGC(TestDvc): self._test_gc() def _test_gc(self): - self.assertTrue(os.path.isdir(self.dvc.cache.cache_dir)) + self.assertTrue(os.path.isdir(self.dvc.cache.local.cache_dir)) for c in self.bad_cache: self.assertFalse(os.path.exists(c)) diff --git a/tests/test_repro.py b/tests/test_repro.py index b5f5f9cd8..d32fd261c 100644 --- a/tests/test_repro.py +++ b/tests/test_repro.py @@ -12,7 +12,7 @@ from dvc.main import main from dvc.command.repro import CmdRepro from dvc.project import ReproductionError from dvc.utils import file_md5 -from dvc.output.local import OutputLOCAL +from dvc.remote.local import RemoteLOCAL from dvc.stage import Stage from tests.basic_env import TestDvc @@ -165,8 +165,8 @@ class TestReproMissingMd5InStageFile(TestRepro): with open(self.file1_stage, 'r') as fd: d = yaml.load(fd) - del(d[Stage.PARAM_OUTS][0][OutputLOCAL.PARAM_MD5]) - del(d[Stage.PARAM_DEPS][0][OutputLOCAL.PARAM_MD5]) + del(d[Stage.PARAM_OUTS][0][RemoteLOCAL.PARAM_MD5]) + del(d[Stage.PARAM_DEPS][0][RemoteLOCAL.PARAM_MD5]) 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 dbef3d336..aee553eb3 100644 --- a/tests/test_stage.py +++ b/tests/test_stage.py @@ -1,4 +1,5 @@ from dvc.output.local import OutputLOCAL +from dvc.remote.local import RemoteLOCAL from dvc.stage import Stage, StageFileFormatError from tests.basic_env import TestDvc @@ -51,8 +52,8 @@ class TestSchemaDepsOuts(TestSchema): Stage.validate(d) def test_list(self): - l = [{OutputLOCAL.PARAM_PATH: 'foo', OutputLOCAL.PARAM_MD5: '123'}, - {OutputLOCAL.PARAM_PATH: 'bar', OutputLOCAL.PARAM_MD5: None}, + l = [{OutputLOCAL.PARAM_PATH: 'foo', RemoteLOCAL.PARAM_MD5: '123'}, + {OutputLOCAL.PARAM_PATH: 'bar', RemoteLOCAL.PARAM_MD5: None}, {OutputLOCAL.PARAM_PATH: 'baz'}] d = {Stage.PARAM_DEPS: l} Stage.validate(d)