Stone_Red/dvc
1
0
Fork 0
mirror of https://github.com/Stone-Red-Code/dvc.git synced 2026-11-11 23:10:10 +01:00
Code Issues Packages Projects Releases Wiki Activity

test: test remotes in CLI

Browse Source 
Signed-off-by: Ruslan Kuprieiev <[email protected]>
This commit is contained in:
Ruslan Kuprieiev
2018-04-07 02:34:00 +03:00
parent 1ad63d1ba6
commit e8b5af6cd9
1 changed files with 145 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_data_cloud.py
+145 -54
View File
@@ -2,6 +2,7 @@ import os
import stat
import uuid
import shutil
import tempfile
from dvc.main import main
from dvc.config import Config, ConfigError
@@ -17,6 +18,58 @@ TEST_SECTION = 'remote "{}"'.format(TEST_REMOTE)
TEST_CONFIG = {Config.SECTION_CORE: {Config.SECTION_CORE_REMOTE: TEST_REMOTE},
TEST_SECTION: {Config.SECTION_REMOTE_URL: ''}}
TEST_AWS_REPO_BUCKET = 'dvc-test'
TEST_AWS_REPO_REGION = 'us-east-2'
TEST_GCP_REPO_BUCKET = 'dvc-test'
def should_test_aws():
dvc_test_aws = os.getenv("DVC_TEST_AWS")
if dvc_test_aws == "true":
return True
elif dvc_test_aws == "false":
return False
if os.getenv("AWS_ACCESS_KEY_ID") and os.getenv("AWS_SECRET_ACCESS_KEY"):
return True
return False
def should_test_gcp():
if os.getenv("DVC_TEST_GCP") == "true":
return True
if os.getenv("GOOGLE_APPLICATION_CREDENTIALS") and os.getenv("GCP_CREDS"):
return True
return False
def get_local_storagepath():
return tempfile.mkdtemp()
def get_local_url():
return get_local_storagepath()
def get_aws_storagepath():
return TEST_AWS_REPO_BUCKET + '/' + str(uuid.uuid4())
def get_aws_url():
return 's3://' + get_aws_storagepath()
def get_gcp_storagepath():
return TEST_GCP_REPO_BUCKET + '/' + str(uuid.uuid4())
def get_gcp_url():
return 'gs://' + get_gcp_storagepath()
class TestDataCloud(TestDvc):
def _test_cloud(self, config, cl):
@@ -106,59 +159,33 @@ class TestDataCloudBase(TestDvc):
class TestDataCloudAWS(TestDataCloudBase):
TEST_REPO_BUCKET = 'dvc-test'
TEST_REPO_REGION = 'us-east-2'
def _should_test(self):
dvc_test_aws = os.getenv("DVC_TEST_AWS")
if dvc_test_aws == "true":
return True
elif dvc_test_aws == "false":
return False
if os.getenv("AWS_ACCESS_KEY_ID") and os.getenv("AWS_SECRET_ACCESS_KEY"):
return True
return False
def _setup_cloud(self):
if not self._should_test():
if not should_test_aws():
return
repo = 's3://' + self.TEST_REPO_BUCKET + '/' + str(uuid.uuid4())
repo = get_aws_url()
# Setup cloud
config = TEST_CONFIG
config[TEST_SECTION][Config.SECTION_REMOTE_URL] = repo
config[TEST_SECTION][Config.SECTION_AWS_REGION] = self.TEST_REPO_REGION
config[TEST_SECTION][Config.SECTION_AWS_REGION] = TEST_AWS_REPO_REGION
cloud_settings = CloudSettings(self.dvc.cache, None, config[TEST_SECTION])
self.cloud = DataCloudAWS(cloud_settings)
def test(self):
if self._should_test():
if should_test_aws():
self._test_cloud()
class TestDataCloudGCP(TestDataCloudBase):
TEST_REPO_GCP_BUCKET = 'dvc-test'
def _should_test(self):
if os.getenv("DVC_TEST_GCP") == "true":
return True
if os.getenv("GOOGLE_APPLICATION_CREDENTIALS") and os.getenv("GCP_CREDS"):
return True
return False
def _setup_cloud(self):
if not self._should_test():
if not should_test_gcp():
return
if os.getenv("GOOGLE_APPLICATION_CREDENTIALS") and os.getenv("GCP_CREDS"):
shutil.copyfile(self.GCP_CREDS_FILE, os.getenv("GOOGLE_APPLICATION_CREDENTIALS"))
repo = 'gs://' + self.TEST_REPO_GCP_BUCKET + '/' + str(uuid.uuid4())
repo = get_gcp_url()
# Setup cloud
config = TEST_CONFIG
@@ -167,17 +194,13 @@ class TestDataCloudGCP(TestDataCloudBase):
self.cloud = DataCloudGCP(cloud_settings)
def test(self):
if self._should_test():
if should_test_gcp():
self._test_cloud()
class TestDataCloudLOCAL(TestDataCloudBase):
def _should_test(self):
return True
def _setup_cloud(self):
self.dname = 'cloud'
os.mkdir(self.dname)
self.dname = get_local_url()
config = TEST_CONFIG
config[TEST_SECTION][Config.SECTION_REMOTE_URL] = self.dname
@@ -185,22 +208,20 @@ class TestDataCloudLOCAL(TestDataCloudBase):
self.cloud = DataCloudLOCAL(cloud_settings)
def test(self):
if self._should_test():
self._test_cloud()
self.assertTrue(os.path.isdir(self.dname))
self._test_cloud()
self.assertTrue(os.path.isdir(self.dname))
class TestDataCloudLocalCli(TestDvc):
class TestDataCloudCLIBase(TestDvc):
def main(self, args):
ret = main(args)
self.assertEqual(ret, 0)
def test(self):
dname = 'cloud'
os.mkdir(dname)
self.main(['config', 'core.cloud', 'local'])
self.main(['config', 'local.storagepath', dname])
def _test_cloud(self, remote=None):
if remote:
remote_arg = ['-r', remote]
else:
remote_arg = []
stage = self.dvc.add(self.FOO)
cache = stage.outs[0].cache
@@ -209,27 +230,97 @@ class TestDataCloudLocalCli(TestDvc):
cache_dir = stage_dir.outs[0].cache
#FIXME check status output
self.main(['status'])
self.main(['status', '-c'] + remote_arg)
self.main(['push'])
self.main(['push'] + remote_arg)
self.assertTrue(os.path.exists(cache))
self.assertTrue(os.path.isfile(cache))
self.assertTrue(os.path.isfile(cache_dir))
self.main(['status'])
self.main(['status', '-c'] + remote_arg)
shutil.rmtree(self.dvc.cache.cache_dir)
self.main(['status'])
self.main(['status', '-c'] + remote_arg)
self.main(['pull'])
self.main(['pull'] + remote_arg)
self.assertTrue(os.path.exists(cache))
self.assertTrue(os.path.isfile(cache))
with open(cache, 'r') as fd:
self.assertEqual(fd.read(), self.FOO_CONTENTS)
self.assertTrue(os.path.isfile(cache_dir))
self.main(['status'])
self.main(['status', '-c'] + remote_arg)
def should_test(self):
return True
def _test(self):
pass
def _test_compat(self):
pass
def test(self):
if self.should_test():
self._test()
self._test_compat()
class TestDataCloudLOCALCLI(TestDataCloudCLIBase):
def _test_compat(self):
storagepath = get_local_storagepath()
self.main(['config', 'core.cloud', 'local'])
self.main(['config', 'local.storagepath', storagepath])
self._test_cloud()
def _test(self):
url = get_local_url()
self.main(['remote', 'add', TEST_REMOTE, url])
self._test_cloud(TEST_REMOTE)
class TestDataCloudAWSCLI(TestDataCloudCLIBase):
def should_test(self):
return should_test_aws()
def _test_compat(self):
storagepath = get_aws_storagepath()
self.main(['config', 'core.cloud', 'aws'])
self.main(['config', 'aws.storagepath', storagepath])
self.main(['config', 'aws.region', TEST_AWS_REPO_REGION])
self._test_cloud()
def _test(self):
url = get_aws_url()
self.main(['remote', 'add', TEST_REMOTE, url])
self.main(['remote', 'modify', TEST_REMOTE, Config.SECTION_AWS_REGION, TEST_AWS_REPO_REGION])
self._test_cloud(TEST_REMOTE)
class TestDataCloudGCPCLI(TestDataCloudCLIBase):
def should_test(self):
return should_test_gcp()
def _test_compat(self):
storagepath = get_gcp_storagepath()
self.main(['config', 'core.cloud', 'gcp'])
self.main(['config', 'gcp.storagepath', storagepath])
self._test_cloud()
def _test(self):
url = get_gcp_url()
self.main(['remote', 'add', TEST_REMOTE, url])
self._test_cloud(TEST_REMOTE)
class TestDataCloudErrorCLI(TestDvc):