diff --git a/dvc/dependency/__init__.py b/dvc/dependency/__init__.py index 646f9aaae..131ee2dc4 100644 --- a/dvc/dependency/__init__.py +++ b/dvc/dependency/__init__.py @@ -1,127 +1,39 @@ -import os import schema -import posixpath -import ntpath from dvc.exceptions import DvcException -from dvc.logger import Logger -from dvc.utils import remove + +from dvc.dependency.base import DependencyBase +from dvc.dependency.s3 import DependencyS3 +from dvc.dependency.gs import DependencyGS +from dvc.dependency.local import DependencyLOCAL -class DependencyError(DvcException): - def __init__(self, path, msg): - super(DependencyError, self).__init__('Dependency \'{}\' error: {}'.format(path, msg)) +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), +} -class DependencyOutsideOfRepoError(DependencyError): - def __init__(self, path): - super(DependencyOutsideOfRepoError, self).__init__(path, 'outside of repository') +def _get(path): + for d in DEPS: + if d.supported(path): + return d + raise DvcException('Dependency \'{}\' is not supported'.format(path)) -class DependencyDoesNotExistError(DependencyError): - def __init__(self, path): - super(DependencyDoesNotExistError, self).__init__(path, 'does not exist') +def loadd_from(stage, d_list): + ret = [] + for d in d_list: + p = d[DependencyBase.PARAM_PATH] + ret.append(_get(p).loadd(stage, d)) + return ret -class DependencyIsNotFileOrDirError(DependencyError): - def __init__(self, path): - super(DependencyIsNotFileOrDirError, self).__init__(path, 'not a file or directory') - - -class Dependency(object): - PARAM_PATH = 'path' - PARAM_MD5 = 'md5' - MD5_DIR_SUFFIX = '.dir' - - OutsideOfRepoError = DependencyOutsideOfRepoError - DoesNotExistError = DependencyDoesNotExistError - IsNotFileOrDirError = DependencyIsNotFileOrDirError - - SCHEMA = { - PARAM_PATH: str, - schema.Optional(PARAM_MD5): schema.Or(str, None), - } - - def __init__(self, stage, path, md5=None): - self.stage = stage - self.project = stage.project - self.path = os.path.abspath(os.path.normpath(path)) - - if not self.path.startswith(self.project.root_dir): - raise self.OutsideOfRepoError(self.rel_path) - - self.md5 = md5 - - @property - def rel_path(self): - return os.path.relpath(self.path) - - def _changed_md5(self): - if not os.path.exists(self.path): - return True - - return self.project.state.changed(self.path, self.md5) - - @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 - - def status(self): - if self.changed(): - #FIXME better msgs - return {self.rel_path: 'changed'} - return {} - - def save(self): - if not os.path.exists(self.path): - raise self.DoesNotExistError(self.rel_path) - - if not os.path.isfile(self.path) and not os.path.isdir(self.path): - raise self.NotFileOrDirError(self.rel_path) - - if (os.path.isfile(self.path) and os.path.getsize(self.path) == 0) or \ - (os.path.isdir(self.path) and len(os.listdir(self.path)) == 0): - self.project.logger.warn("File/directory '{}' is empty.".format(self.rel_path)) - - self.md5 = self.project.state.update(self.path) - - @staticmethod - def unixpath(path): - assert not ntpath.isabs(path) - assert not posixpath.isabs(path) - return path.replace('\\', '/') - - def dumpd(self): - return { - self.PARAM_PATH: self.unixpath(os.path.relpath(self.path, self.stage.cwd)), - self.PARAM_MD5: self.md5, - } - - @classmethod - def loadd(cls, stage, d): - relpath = os.path.normpath(Dependency.unixpath(d[Dependency.PARAM_PATH])) - path = os.path.join(stage.cwd, relpath) - md5 = d.get(Dependency.PARAM_MD5, None) - return cls(stage, path, md5=md5) - - @classmethod - def loadd_from(cls, stage, d_list): - return [cls.loadd(stage, x) for x in d_list] - - @classmethod - def loads(cls, stage, s): - return cls(stage, os.path.join(stage.cwd, s), md5=None) - - @classmethod - def loads_from(cls, stage, s_list): - return [cls.loads(stage, x) for x in s_list] +def loads_from(stage, s_list): + ret = [] + for s in s_list: + ret.append(_get(s).loads(stage, s)) + return ret diff --git a/dvc/dependency/base.py b/dvc/dependency/base.py new file mode 100644 index 000000000..15ca934cf --- /dev/null +++ b/dvc/dependency/base.py @@ -0,0 +1,71 @@ +import re +import schema +import posixpath +import ntpath + +from dvc.exceptions import DvcException + + +class DependencyError(DvcException): + def __init__(self, path, msg): + super(DependencyError, self).__init__('Dependency \'{}\' error: {}'.format(path, msg)) + + +class DependencyOutsideOfRepoError(DependencyError): + def __init__(self, path): + super(DependencyOutsideOfRepoError, self).__init__(path, 'outside of repository') + + +class DependencyDoesNotExistError(DependencyError): + def __init__(self, path): + super(DependencyDoesNotExistError, self).__init__(path, 'does not exist') + + +class DependencyIsNotFileOrDirError(DependencyError): + def __init__(self, path): + super(DependencyIsNotFileOrDirError, self).__init__(path, 'not a file or directory') + + +class DependencyBase(object): + REGEX = None + + PARAM_PATH = 'path' + + def __init__(self, stage, path): + self.path = path + + @classmethod + def match(cls, url): + return re.match(cls.REGEX, url) + + @classmethod + def supported(cls, url): + return cls.match(url) != None + + def changed(self): + raise NotImplemented + + def status(self): + if self.changed(): + #FIXME better msgs + return {self.rel_path: 'changed'} + return {} + + def save(self): + raise NotImplemented + + def dumpd(self): + return {self.PARAM_PATH: self.path} + + @classmethod + def loadd(cls, stage, d): + path = d[cls.PARAM_PATH] + return cls(stage, path) + + @classmethod + def loads(cls, stage, s): + return cls(stage, s) + + @classmethod + def loads_from(cls, stage, s_list): + return [cls.loads(stage, x) for x in s_list] diff --git a/dvc/dependency/gs.py b/dvc/dependency/gs.py new file mode 100644 index 000000000..448d6ef8d --- /dev/null +++ b/dvc/dependency/gs.py @@ -0,0 +1,23 @@ +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 + + +class DependencyGS(DependencyS3): + REGEX = DataCloudGCP.REGEX + + def get_etag(self): + o = urlparse(self.path) + bucket_name = o.netloc + key = o.path.lstrip('/') + + client = gc.Client() + bucket = client.bucket(bucket_name) + blob = bucket.get_blob(key) + return blob.etag diff --git a/dvc/dependency/local.py b/dvc/dependency/local.py new file mode 100644 index 000000000..b168bf5c3 --- /dev/null +++ b/dvc/dependency/local.py @@ -0,0 +1,105 @@ +import os +import schema +import posixpath +import ntpath + +from dvc.dependency.base import * +from dvc.logger import Logger +from dvc.utils import remove +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 + + SCHEMA = { + PARAM_PATH: str, + schema.Optional(PARAM_MD5): schema.Or(str, None), + } + + def __init__(self, stage, path, md5=None): + self.stage = stage + self.project = stage.project + self.path = os.path.abspath(os.path.normpath(path)) + self.md5 = md5 + + @property + def rel_path(self): + return os.path.relpath(self.path) + + def _changed_md5(self): + if not os.path.exists(self.path): + return True + + return self.project.state.changed(self.path, self.md5) + + @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 + + def save(self): + if not os.path.exists(self.path): + raise self.DoesNotExistError(self.rel_path) + + if not os.path.isfile(self.path) and not os.path.isdir(self.path): + raise self.NotFileOrDirError(self.rel_path) + + if (os.path.isfile(self.path) and os.path.getsize(self.path) == 0) or \ + (os.path.isdir(self.path) and len(os.listdir(self.path)) == 0): + self.project.logger.warn("File/directory '{}' is empty.".format(self.rel_path)) + + self.md5 = self.project.state.update(self.path) + + @staticmethod + def unixpath(path): + assert not ntpath.isabs(path) + assert not posixpath.isabs(path) + return path.replace('\\', '/') + + def dumpd(self): + if self.path.startswith(self.stage.project.root_dir): + path = self.unixpath(os.path.relpath(self.path, self.stage.cwd)) + else: + path = self.path + + return {self.PARAM_PATH: path, + self.PARAM_MD5:self.md5} + + @classmethod + def loadd(cls, stage, d): + path = d[cls.PARAM_PATH] + if not os.path.isabs(path): + path = cls.unixpath(path) + path = os.path.join(stage.cwd, path) + path = os.path.normpath(path) + md5 = d.get(cls.PARAM_MD5, None) + return cls(stage, path, md5=md5) + + @classmethod + def loadd_from(cls, stage, d_list): + return [cls.loadd(stage, x) for x in d_list] + + @classmethod + def loads(cls, stage, s): + return cls(stage, os.path.join(stage.cwd, s), md5=None) + + @classmethod + def loads_from(cls, stage, s_list): + return [cls.loads(stage, x) for x in s_list] diff --git a/dvc/dependency/s3.py b/dvc/dependency/s3.py new file mode 100644 index 000000000..9b8bfb971 --- /dev/null +++ b/dvc/dependency/s3.py @@ -0,0 +1,53 @@ +import boto3 + +try: + from urlparse import urlparse +except ImportError: + from urllib.parse import urlparse + +from dvc.dependency.base import DependencyBase +from dvc.cloud.aws import DataCloudAWS + + +class DependencyS3(DependencyBase): + REGEX = DataCloudAWS.REGEX + + PARAM_ETAG = 'etag' + + def __init__(self, stage, path, etag=None): + super(DependencyS3, self).__init__(stage, path) + self.etag = etag + + def get_etag(self): + o = urlparse(self.path) + bucket = o.netloc + key = o.path.lstrip('/') + + session = boto3.Session() + s3 = session.resource('s3') + obj = s3.Object(bucket, key).get() + return obj['ETag'].strip('"') + + def changed(self): + return self.etag != self.get_etag() + + def save(self): + self.etag = self.get_etag() + + def dumpd(self): + return {self.PARAM_PATH: self.path, + self.PARAM_ETAG: self.etag} + + @classmethod + def loadd(cls, stage, d): + path = d[cls.PARAM_PATH] + etag = d.get(cls.PARAM_ETAG, None) + return cls(stage, path, etag=etag) + + @classmethod + def loads(cls, stage, s): + return cls(stage, s, etag=None) + + @classmethod + def loads_from(cls, stage, s_list): + return [cls.loads(stage, x) for x in s_list] diff --git a/dvc/output/__init__.py b/dvc/output/__init__.py index 40a6577a1..23dd3b542 100644 --- a/dvc/output/__init__.py +++ b/dvc/output/__init__.py @@ -1,119 +1,37 @@ -import os import schema -import posixpath -import ntpath -from dvc.dependency import Dependency from dvc.exceptions import DvcException -from dvc.logger import Logger -from dvc.utils import remove + +import dvc.dependency as dependency +from dvc.dependency.base import DependencyBase +from dvc.output.s3 import OutputS3 +from dvc.output.gs import OutputGS +from dvc.output.local import OutputLOCAL -class OutputError(DvcException): - def __init__(self, path, msg): - super(OutputError, self).__init__('Output \'{}\' error: {}'.format(path, msg)) +OUTS = [OutputS3, OutputGS, OutputLOCAL] + +SCHEMA = dependency.SCHEMA +SCHEMA[schema.Optional(OutputLOCAL.PARAM_CACHE)] = bool -class OutputOutsideOfRepoError(OutputError): - def __init__(self, path): - super(OutputOutsideOfRepoError, self).__init__(path, 'outside of repository') +def _get(path): + for o in OUTS: + if o.supported(path): + return o + raise DvcException('Output \'{}\' is not supported'.format(path)) -class OutputDoesNotExistError(OutputError): - def __init__(self, path): - super(OutputDoesNotExistError, self).__init__(path, 'does not exist') +def loadd_from(stage, d_list): + ret = [] + for d in d_list: + p = d[DependencyBase.PARAM_PATH] + ret.append(_get(p).loadd(stage, d)) + return ret -class OutputIsNotFileOrDirError(OutputError): - def __init__(self, path): - super(OutputIsNotFileOrDirError, self).__init__(path, 'not a file or directory') - - -class OutputAlreadyTrackedError(OutputError): - def __init__(self, path): - super(OutputAlreadyTrackedError, self).__init__(path, 'already tracked by scm(e.g. git)') - - -class Output(Dependency): - PARAM_CACHE = 'cache' - - OutsideOfRepoError = OutputOutsideOfRepoError - DoesNotExistError = OutputDoesNotExistError - IsNotFileOrDirError = OutputIsNotFileOrDirError - - SCHEMA = Dependency.SCHEMA - SCHEMA[schema.Optional(PARAM_CACHE)] = bool - - def __init__(self, stage, path, md5=None, use_cache=True): - super(Output, self).__init__(stage, path, md5=md5) - self.use_cache = use_cache - - @property - def cache(self): - if not self.use_cache: - return None - - return self.project.cache.get(self.md5) - - def dumpd(self): - ret = super(Output, self).dumpd() - ret[Output.PARAM_CACHE] = self.use_cache - return ret - - @classmethod - def loadd(cls, stage, d): - ret = super(Output, cls).loadd(stage, d) - ret.use_cache = d.get(Output.PARAM_CACHE, True) - return ret - - @classmethod - def loads(cls, stage, s, use_cache=True): - ret = super(Output, cls).loads(stage, s) - ret.use_cache = use_cache - return ret - - @classmethod - def loads_from(cls, stage, s_list, use_cache=False): - return [cls.loads(stage, x, use_cache=use_cache) for x in s_list] - - def changed(self): - if super(Output, self).changed(): - return True - - if self.use_cache and self.project.cache.changed(self.md5): - return True - - return False - - def checkout(self): - 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)) - return - - self.project.cache.checkout(self.path, self.md5) - - def save(self): - super(Output, self).save() - - if not self.use_cache: - 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) - - def remove(self): - remove(self.path) +def loads_from(stage, s_list, use_cache=False): + ret = [] + for s in s_list: + ret.append(_get(s).loads(stage, s, use_cache=use_cache)) + return ret diff --git a/dvc/output/base.py b/dvc/output/base.py new file mode 100644 index 000000000..b4fb074eb --- /dev/null +++ b/dvc/output/base.py @@ -0,0 +1,21 @@ +from dvc.exceptions import DvcException + + +class OutputError(DvcException): + def __init__(self, path, msg): + super(OutputError, self).__init__('Output \'{}\' error: {}'.format(path, msg)) + + +class OutputDoesNotExistError(OutputError): + def __init__(self, path): + super(OutputDoesNotExistError, self).__init__(path, 'does not exist') + + +class OutputIsNotFileOrDirError(OutputError): + def __init__(self, path): + super(OutputIsNotFileOrDirError, self).__init__(path, 'not a file or directory') + + +class OutputAlreadyTrackedError(OutputError): + def __init__(self, path): + super(OutputAlreadyTrackedError, self).__init__(path, 'already tracked by scm(e.g. git)') diff --git a/dvc/output/gs.py b/dvc/output/gs.py new file mode 100644 index 000000000..0cfa59b9c --- /dev/null +++ b/dvc/output/gs.py @@ -0,0 +1,5 @@ +from dvc.dependency.gs import DependencyGS + +class OutputGS(DependencyGS): + def __init__(self, stage, path, etag=None): + raise NotImplemented diff --git a/dvc/output/local.py b/dvc/output/local.py new file mode 100644 index 000000000..a21e44ebf --- /dev/null +++ b/dvc/output/local.py @@ -0,0 +1,94 @@ +import os +import schema +import posixpath +import ntpath + +from dvc.output.base import * +from dvc.dependency.local import DependencyLOCAL +from dvc.exceptions import DvcException +from dvc.logger import Logger +from dvc.utils import remove + + +class OutputLOCAL(DependencyLOCAL): + PARAM_CACHE = 'cache' + + DoesNotExistError = OutputDoesNotExistError + IsNotFileOrDirError = OutputIsNotFileOrDirError + + SCHEMA = DependencyLOCAL.SCHEMA + SCHEMA[schema.Optional(PARAM_CACHE)] = bool + + def __init__(self, stage, path, md5=None, use_cache=True): + super(OutputLOCAL, self).__init__(stage, path, md5=md5) + self.use_cache = use_cache + + @property + def cache(self): + if not self.use_cache: + return None + + return self.project.cache.get(self.md5) + + def dumpd(self): + ret = super(OutputLOCAL, self).dumpd() + ret[self.PARAM_CACHE] = self.use_cache + return ret + + @classmethod + def loadd(cls, stage, d): + ret = super(OutputLOCAL, cls).loadd(stage, d) + ret.use_cache = d.get(cls.PARAM_CACHE, True) + return ret + + @classmethod + def loads(cls, stage, s, use_cache=True): + ret = super(OutputLOCAL, cls).loads(stage, s) + ret.use_cache = use_cache + return ret + + @classmethod + def loads_from(cls, stage, s_list, use_cache=False): + return [cls.loads(stage, x, use_cache=use_cache) for x in s_list] + + def changed(self): + if super(OutputLOCAL, self).changed(): + return True + + if self.use_cache and self.project.cache.changed(self.md5): + return True + + return False + + def checkout(self): + 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)) + return + + self.project.cache.checkout(self.path, self.md5) + + def save(self): + super(OutputLOCAL, self).save() + + if not self.use_cache: + 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) + + def remove(self): + remove(self.path) diff --git a/dvc/output/s3.py b/dvc/output/s3.py new file mode 100644 index 000000000..b0971dc66 --- /dev/null +++ b/dvc/output/s3.py @@ -0,0 +1,6 @@ +from dvc.dependency.s3 import DependencyS3 + + +class OutputS3(DependencyS3): + def __init__(self, stage, path, etag=None): + raise NotImplemented diff --git a/dvc/stage.py b/dvc/stage.py index 62a1ec8ce..ca7aaec4d 100644 --- a/dvc/stage.py +++ b/dvc/stage.py @@ -4,9 +4,9 @@ import itertools import subprocess import schema +import dvc.dependency as dependency +import dvc.output as output from dvc.exceptions import DvcException -from dvc.output import Output -from dvc.dependency import Dependency from dvc.logger import Logger from dvc.utils import dict_md5 @@ -46,8 +46,8 @@ class Stage(object): SCHEMA = { schema.Optional(PARAM_MD5): schema.Or(str, None), schema.Optional(PARAM_CMD): schema.Or(str, None), - schema.Optional(PARAM_DEPS): schema.Or(schema.And(list, schema.Schema([Dependency.SCHEMA])), None), - schema.Optional(PARAM_OUTS): schema.Or(schema.And(list, schema.Schema([Output.SCHEMA])), None), + schema.Optional(PARAM_DEPS): schema.Or(schema.And(list, schema.Schema([dependency.SCHEMA])), None), + schema.Optional(PARAM_OUTS): schema.Or(schema.And(list, schema.Schema([output.SCHEMA])), None), } def __init__(self, project, path=None, cmd=None, cwd=None, deps=[], outs=[], md5=None): @@ -156,8 +156,8 @@ class Stage(object): cwd=cwd, md5=md5) - stage.deps = Dependency.loadd_from(stage, d.get(Stage.PARAM_DEPS, [])) - stage.outs = Output.loadd_from(stage, d.get(Stage.PARAM_OUTS, [])) + stage.deps = dependency.loadd_from(stage, d.get(Stage.PARAM_DEPS, [])) + stage.outs = output.loadd_from(stage, d.get(Stage.PARAM_OUTS, [])) return stage @@ -177,9 +177,9 @@ class Stage(object): cmd=cmd, cwd=cwd) - stage.outs = Output.loads_from(stage, outs, use_cache=True) - stage.outs += Output.loads_from(stage, outs_no_cache, use_cache=False) - stage.deps = Dependency.loads_from(stage, deps) + stage.outs = output.loads_from(stage, outs, use_cache=True) + stage.outs += output.loads_from(stage, outs_no_cache, use_cache=False) + stage.deps = dependency.loads_from(stage, deps) return stage diff --git a/tests/test_add.py b/tests/test_add.py index 3802857c2..9711c0891 100644 --- a/tests/test_add.py +++ b/tests/test_add.py @@ -6,8 +6,8 @@ import filecmp from dvc.main import main from dvc.utils import file_md5 from dvc.stage import Stage -from dvc.output import OutputOutsideOfRepoError, OutputAlreadyTrackedError -from dvc.output import OutputDoesNotExistError, OutputIsNotFileOrDirError +from dvc.output.base import OutputAlreadyTrackedError +from dvc.output.base import OutputDoesNotExistError, OutputIsNotFileOrDirError from dvc.project import StageNotFoundError from dvc.command.add import CmdAdd @@ -34,12 +34,6 @@ class TestAddNonExistentFile(TestDvc): self.dvc.add('non_existent_file') -class TestAddFileOutsideOfRepo(TestDvc): - def test(self): - with self.assertRaises(OutputOutsideOfRepoError) as cx: - self.dvc.add(os.path.join(os.path.dirname(self.dvc.root_dir), self.FOO)) - - class TestAddDirectory(TestDvc): def test(self): dname = 'directory' diff --git a/tests/test_checkout.py b/tests/test_checkout.py index 8732daf66..cd93baed8 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 import Output +from dvc.output.local import OutputLOCAL 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][Output.PARAM_MD5]) - del(d[Stage.PARAM_DEPS][0][Output.PARAM_MD5]) + del(d[Stage.PARAM_OUTS][0][OutputLOCAL.PARAM_MD5]) + del(d[Stage.PARAM_DEPS][0][OutputLOCAL.PARAM_MD5]) with open(self.file1_stage, 'w') as fd: yaml.dump(d, fd) diff --git a/tests/test_repro.py b/tests/test_repro.py index 9313faeb7..429c3b74f 100644 --- a/tests/test_repro.py +++ b/tests/test_repro.py @@ -4,14 +4,20 @@ import stat import shutil import filecmp +import boto3 +import uuid +from google.cloud import storage as gc + 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 import Output +from dvc.output.local import OutputLOCAL from dvc.stage import Stage from tests.basic_env import TestDvc +from tests.test_data_cloud import _should_test_aws, TEST_AWS_REPO_BUCKET +from tests.test_data_cloud import _should_test_gcp, TEST_GCP_REPO_BUCKET class TestRepro(TestDvc): @@ -159,8 +165,8 @@ class TestReproMissingMd5InStageFile(TestRepro): with open(self.file1_stage, 'r') as fd: d = yaml.load(fd) - del(d[Stage.PARAM_OUTS][0][Output.PARAM_MD5]) - del(d[Stage.PARAM_DEPS][0][Output.PARAM_MD5]) + del(d[Stage.PARAM_OUTS][0][OutputLOCAL.PARAM_MD5]) + del(d[Stage.PARAM_DEPS][0][OutputLOCAL.PARAM_MD5]) with open(self.file1_stage, 'w') as fd: yaml.dump(d, fd) @@ -178,3 +184,54 @@ class TestCmdRepro(TestRepro): ret = main(['repro', 'non-existing-file']) self.assertNotEqual(ret, 0) + + +class TestReproDependencyS3(TestDvc): + def test(self): + if not _should_test_aws(): + return + + key = str(uuid.uuid4()) + '/' + self.FOO + path = 's3://' + TEST_AWS_REPO_BUCKET + '/' + key + + s3 = boto3.resource('s3') + + with open(self.FOO, 'rb') as fd: + s3.Bucket(TEST_AWS_REPO_BUCKET).put_object(Key=key, Body=fd.read()) + + stage = self.dvc.run(outs=[self.FOO], + deps=[path], + cmd='aws s3 cp {} {}'.format(path, self.FOO)) + + with open(self.BAR, 'rb') as fd: + s3.Bucket(TEST_AWS_REPO_BUCKET).put_object(Key=key, Body=fd.read()) + + stages = self.dvc.reproduce(stage.path) + + self.assertEqual(len(stages), 1) + self.assertTrue(filecmp.cmp(self.FOO, self.BAR)) + + +class TestReproDependencyGS(TestDvc): + def test(self): + if not _should_test_gcp(): + return + + key = str(uuid.uuid4()) + '/' + self.FOO + path = 'gs://' + TEST_GCP_REPO_BUCKET + '/' + key + + client = gc.Client() + bucket = client.bucket(TEST_GCP_REPO_BUCKET) + + bucket.blob(key).upload_from_filename(self.FOO) + + stage = self.dvc.run(outs=[self.FOO], + deps=[path], + cmd='gsutil cp {} {}'.format(path, self.FOO)) + + bucket.blob(key).upload_from_filename(self.BAR) + + stages = self.dvc.reproduce(stage.path) + + self.assertEqual(len(stages), 1) + self.assertTrue(filecmp.cmp(self.FOO, self.BAR)) diff --git a/tests/test_stage.py b/tests/test_stage.py index fe4f0d046..dbef3d336 100644 --- a/tests/test_stage.py +++ b/tests/test_stage.py @@ -1,4 +1,4 @@ -from dvc.output import Output +from dvc.output.local import OutputLOCAL from dvc.stage import Stage, StageFileFormatError from tests.basic_env import TestDvc @@ -51,13 +51,13 @@ class TestSchemaDepsOuts(TestSchema): Stage.validate(d) def test_list(self): - l = [{Output.PARAM_PATH: 'foo', Output.PARAM_MD5: '123'}, - {Output.PARAM_PATH: 'bar', Output.PARAM_MD5: None}, - {Output.PARAM_PATH: 'baz'}] + l = [{OutputLOCAL.PARAM_PATH: 'foo', OutputLOCAL.PARAM_MD5: '123'}, + {OutputLOCAL.PARAM_PATH: 'bar', OutputLOCAL.PARAM_MD5: None}, + {OutputLOCAL.PARAM_PATH: 'baz'}] d = {Stage.PARAM_DEPS: l} Stage.validate(d) - l[0][Output.PARAM_CACHE] = True - l[1][Output.PARAM_CACHE] = False + l[0][OutputLOCAL.PARAM_CACHE] = True + l[1][OutputLOCAL.PARAM_CACHE] = False d = {Stage.PARAM_OUTS: l} Stage.validate(d)