Merge pull request #730 from efiop/master

gs/s3 as dependency
This commit is contained in:
Ruslan Kuprieiev
2018-05-24 18:31:09 +03:00
committed by GitHub
16 changed files with 513 additions and 255 deletions
+28 -116
View File
@@ -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
+71
View File
@@ -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]
+23
View File
@@ -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
+105
View File
@@ -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<path>(/+|.:\\+)?.*)$'
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]
+53
View File
@@ -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]
+26 -108
View File
@@ -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
+21
View File
@@ -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)')
+5
View File
@@ -0,0 +1,5 @@
from dvc.dependency.gs import DependencyGS
class OutputGS(DependencyGS):
def __init__(self, stage, path, etag=None):
raise NotImplemented
+94
View File
@@ -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)
+6
View File
@@ -0,0 +1,6 @@
from dvc.dependency.s3 import DependencyS3
class OutputS3(DependencyS3):
def __init__(self, stage, path, etag=None):
raise NotImplemented
+9 -9
View File
@@ -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
+1 -2
View File
@@ -1,6 +1,5 @@
import os
import time
import json
import requests
from dvc import VERSION_BASE
@@ -36,7 +35,7 @@ class Updater(object):
try:
r = requests.get(self.URL)
j = json.loads(r.content)
j = r.json()
latest = j['version']
open(self.updater_file, 'w+').close()
except Exception as exc:
+2 -8
View File
@@ -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'
+3 -3
View File
@@ -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)
+60 -3
View File
@@ -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))
+6 -6
View File
@@ -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)