diff --git a/.travis.yml b/.travis.yml index 0efc91eb5..10a0fbcc7 100644 --- a/.travis.yml +++ b/.travis.yml @@ -7,6 +7,29 @@ python: - "3.5" # - "3.6" Not supported by pyinstaller yet cache: pip +# Travis doesn't support testing python on Mac yet, so we need +# to workaround it by installing it directly with brew. +# +# Note that we specify language as ruby-2.4.0 to help us satisfy +# fpm dependencies, as travis runs el capitan with an old ruby +# version, that causes fpm to crash. +matrix: + include: + - os: osx + language: ruby + rvm: 2.4.0 + env: PYTHON_VER=2.7.10 + - os: osx + language: ruby + rvm: 2.4.0 + env: PYTHON_VER=3.4.3 + - os: osx + language: ruby + rvm: 2.4.0 + env: PYTHON_VER=3.5.0 +before_install: + - if [[ "$TRAVIS_OS_NAME" == "osx" ]]; then pyenv install $PYTHON_VER; fi + - if [[ "$TRAVIS_OS_NAME" == "osx" ]]; then pyenv global $PYTHON_VER; fi install: - pip install --upgrade pip - pip install -r requirements.txt @@ -16,7 +39,8 @@ install: - mkdir -p ~/.aws && printf "[default]\naws_access_key_id = ABCDEFG\naws_secret_access_key = 123456789" > ~/.aws/credentials script: - ./unittests.sh - - ./build_linux.sh + - ./functests.sh + - if [[ "$TRAVIS_OS_NAME" == "osx" ]]; then ./build_macos.sh; else ./build_linux.sh; fi - ./build_package.sh after_sucess: - CODECLIMATE_REPO_TOKEN=7350efa14a67c3911af96dadaabe811b43eead7b466309b47687499bf5d72287 codeclimate-test-reporter @@ -28,6 +52,7 @@ deploy: file: - dvc*.rpm - dvc*.deb + - dvc*.pkg skip_cleanup: true on: tags: true diff --git a/build_macos.sh b/build_macos.sh index c51eca7ac..f0c09ead8 100755 --- a/build_macos.sh +++ b/build_macos.sh @@ -1,5 +1,7 @@ #!/bin/sh +set -e + BUILD_DIR=build INSTALL_DIR=usr/local BIN_DIR=$BUILD_DIR/$INSTALL_DIR/bin @@ -40,16 +42,6 @@ install_dependencies() pip install -r requirements.txt print_info "Installing fpm..." - if command_exists dnf; then - sudo dnf install ruby-devel gcc make rpm-build - elif command_exists yum; then - sudo yum install ruby-devel gcc make rpm-build - elif command_exists apt-get; then - sudo apt-get install ruby ruby-dev rubygems build-essential - #else - # echo "Unable to install fpm dependencies" && exit 1 - fi - gem install --no-ri --no-rdoc fpm print_info "Installing pyinstaller..." @@ -65,8 +57,7 @@ build_dvc() cleanup install_dependencies build_dvc -#fpm_build rpm fpm_build osxpkg -#cleanup +cleanup -print_info "Successfully built dvc rpm and deb packages" +print_info "Successfully built dvc osx package" diff --git a/dvc/cli.py b/dvc/cli.py index f886bd704..7341d300b 100644 --- a/dvc/cli.py +++ b/dvc/cli.py @@ -7,12 +7,17 @@ from dvc.command.init import CmdInit from dvc.command.remove import CmdRemove from dvc.command.run import CmdRun from dvc.command.repro import CmdRepro -from dvc.command.data_sync import CmdDataSync +from dvc.command.data_sync import CmdDataSync, CmdDataPush, CmdDataPull, CmdDataStatus from dvc.command.lock import CmdLock from dvc.command.gc import CmdGC from dvc.command.import_file import CmdImportFile from dvc.command.target import CmdTarget +<<<<<<< HEAD from dvc.command.instance_create import CmdInstanceCreate +======= +from dvc.command.config import CmdConfig +from dvc.command.test import CmdTest +>>>>>>> f0b75172e7a0d4b14191354e8d0960a41a56ccb6 from dvc.config import Config from dvc import VERSION @@ -118,22 +123,48 @@ def parse_args(argv=None): help='Arguments of a command') run_parser.set_defaults(func=CmdRun) - # Sync - sync_parser = subparsers.add_parser( - 'sync', - parents=[parent_parser], - help='Synchronize data file with cloud (cloud settings already setup.') - sync_parser.add_argument( + # Parent parser used in sync/pull/push + parent_sync_parser = argparse.ArgumentParser( + add_help=False, + parents=[parent_parser]) + parent_sync_parser.add_argument( 'targets', nargs='+', help='File or directory to sync.') - sync_parser.add_argument('-j', + parent_sync_parser.add_argument('-j', '--jobs', type=int, default=cpu_count(), help='Number of jobs to run simultaneously.') + + # Sync + sync_parser = subparsers.add_parser( + 'sync', + parents=[parent_sync_parser], + help='Synchronize data file with cloud (cloud settings already setup.') sync_parser.set_defaults(func=CmdDataSync) + # Pull + pull_parser = subparsers.add_parser( + 'pull', + parents=[parent_sync_parser], + help='Pull data files from the cloud') + pull_parser.set_defaults(func=CmdDataPull) + + # Push + push_parser = subparsers.add_parser( + 'push', + parents=[parent_sync_parser], + help='Push data files to the cloud') + push_parser.set_defaults(func=CmdDataPush) + + # Status + status_parser = subparsers.add_parser( + 'status', + parents=[parent_sync_parser], + help='Show status for data files') + status_parser.set_defaults(func=CmdDataStatus) + # Repro repro_parser = subparsers.add_parser( 'repro', @@ -201,6 +232,12 @@ def parse_args(argv=None): type=int, default=cpu_count(), help='Number of jobs to run simultaneously.') + import_parser.add_argument('-c', + '--continue', + dest='cont', + action='store_true', + default=False, + help='Resume downloading file from url') import_parser.set_defaults(func=CmdImportFile) # Lock @@ -263,6 +300,7 @@ def parse_args(argv=None): help='Reset target.') target_parser.set_defaults(func=CmdTarget) +<<<<<<< HEAD # Cloud ex_parser = subparsers.add_parser( 'ex', @@ -340,6 +378,20 @@ def parse_args(argv=None): default=False, help='Detect all ephemeral disks and stripe together in raid-0') instance_create_parser.set_defaults(func=CmdInstanceCreate) +======= + # Config + config_parser = subparsers.add_parser( + 'config', + parents=[parent_parser], + help='Get or set repository options') + config_parser.add_argument('name', + help='Option name') + config_parser.add_argument('value', + nargs='?', + default=None, + help='Option value') + config_parser.set_defaults(func=CmdConfig) +>>>>>>> f0b75172e7a0d4b14191354e8d0960a41a56ccb6 if isinstance(argv, str): argv = argv.split() diff --git a/dvc/command/base.py b/dvc/command/base.py index 208939b6f..440a302e9 100644 --- a/dvc/command/base.py +++ b/dvc/command/base.py @@ -55,6 +55,10 @@ class CmdBase(object): def args(self): return self._settings.args + @property + def cloud(self): + return self._settings.cloud + @property def parsed_args(self): return self._settings._parsed_args diff --git a/dvc/command/config.py b/dvc/command/config.py new file mode 100644 index 000000000..340c40fbf --- /dev/null +++ b/dvc/command/config.py @@ -0,0 +1,51 @@ +import os +import configobj + +from dvc.command.base import CmdBase +from dvc.logger import Logger +from dvc.config import Config + +class CmdConfig(CmdBase): + def __init__(self, settings): + super(CmdConfig, self).__init__(settings) + + @property + def _config_path(self): + return os.path.join(self.git.git_dir, Config.CONFIG) + + def _find_key(self, l, name): + for k in l: + if k.lower() == name.lower(): + return k + + return None + + def run(self): + _section, _opt = self.parsed_args.name.strip().split('.') + + # Using configobj because it doesn't + # drop comments like configparser does. + config = configobj.ConfigObj(self._config_path) + + section = self._find_key(config.keys(), _section) + if not section: + return 1 + + opt = self._find_key(config[section].keys(), _opt) + if not opt: + return 1 + + if self.parsed_args.value == None: + Logger.info(config[section][opt]) + return 0 + + try: + config[section][opt] = self.parsed_args.value + config.write() + except Exception as exc: + Logger.error('Failed to set \'{}\' to \'{}\': {}'.format(self.parsed_args.name, + self.parsed_args.value, + exc)) + return 1 + + return 0 diff --git a/dvc/command/data_sync.py b/dvc/command/data_sync.py index fbe9ff211..9b377c7b2 100644 --- a/dvc/command/data_sync.py +++ b/dvc/command/data_sync.py @@ -1,17 +1,7 @@ -import os -from multiprocessing.pool import ThreadPool - from dvc.command.base import CmdBase, DvcLock -from dvc.exceptions import DvcException -from dvc.system import System -from dvc.data_cloud import DataCloud -from dvc.utils import map_progress - - -class DataSyncError(DvcException): - def __init__(self, msg): - super(DataSyncError, self).__init__('Data sync error: {}'.format(msg)) +from dvc.logger import Logger +import dvc.data_cloud as cloud class CmdDataSync(CmdBase): def __init__(self, settings): @@ -19,21 +9,44 @@ class CmdDataSync(CmdBase): def run(self): with DvcLock(self.is_locker, self.git): - cloud = DataCloud(self.settings) - targets = [] + self.cloud.sync(self.parsed_args.targets, self.parsed_args.jobs) - if len(self.parsed_args.targets) == 0: - raise DataSyncError('Sync target is not specified') - for target in self.parsed_args.targets: - if System.islink(target): - targets.append(target) - elif os.path.isdir(target): - for root, dirs, files in os.walk(target): - for f in files: - targets.append(os.path.join(root, f)) - else: - raise DataSyncError('File "{}" does not exit'.format(target)) +class CmdDataPull(CmdBase): + def __init__(self, settings): + super(CmdDataPull, self).__init__(settings) - map_progress(cloud.sync, targets, self.parsed_args.jobs) - pass + def run(self): + with DvcLock(self.is_locker, self.git): + self.cloud.pull(self.parsed_args.targets, self.parsed_args.jobs) + + +class CmdDataPush(CmdBase): + def __init__(self, settings): + super(CmdDataPush, self).__init__(settings) + + def run(self): + with DvcLock(self.is_locker, self.git): + self.cloud.push(self.parsed_args.targets, self.parsed_args.jobs) + +class CmdDataStatus(CmdBase): + def __init__(self, settings): + super(CmdDataStatus, self).__init__(settings) + + def run(self): + with DvcLock(self.is_locker, self.git): + status = self.cloud.status(self.parsed_args.targets, self.parsed_args.jobs) + + for s in status: + target, ret = s + + if ret == cloud.STATUS_UNKNOWN or ret == cloud.STATUS_OK: + continue + + prefix_map = { + cloud.STATUS_DELETED : 'deleted: ', + cloud.STATUS_MODIFIED : 'modified:', + cloud.STATUS_NEW : 'new file:', + } + + Logger.info('\t{}\t{}'.format(prefix_map[ret], target.resolved_cache.dvc)) diff --git a/dvc/command/gc.py b/dvc/command/gc.py index 6ff3a5051..c303fadaf 100644 --- a/dvc/command/gc.py +++ b/dvc/command/gc.py @@ -2,7 +2,6 @@ import os from dvc.command.traverse import Traverse from dvc.logger import Logger -from dvc.path.data_item import DataItemError class CmdGC(Traverse): @@ -12,11 +11,7 @@ class CmdGC(Traverse): def process_file(self, target): Logger.debug(u'[Cmd-GC] GC file {}.'.format(target)) - try: - data_item = self.settings.path_factory.existing_data_item(target) - except DataItemError: - Logger.warn(u'[Cmd-GC] Data item {} is not a valid symbolic link'.format(target)) - data_item = self.settings.path_factory.data_item(target) + data_item = self._get_data_item(target) for cache_data_item in data_item.get_all_caches(): if cache_data_item.cache.relative != data_item.resolved_cache.relative: @@ -27,10 +22,6 @@ class CmdGC(Traverse): Logger.debug(u'[Cmd-GC] GC data item {}. Success.'.format(data_item.data.relative)) pass - def _remove_cloud_cache(self, data_item): - if not self.parsed_args.keep_in_cloud: - self.cloud.remove_from_cloud(data_item) - def is_recursive(self): return self.parsed_args.recursive diff --git a/dvc/command/import_file.py b/dvc/command/import_file.py index 201f93cc7..e6530bf8b 100644 --- a/dvc/command/import_file.py +++ b/dvc/command/import_file.py @@ -1,15 +1,18 @@ import os import re -import requests + +try: + from urlparse import urlparse +except ImportError: + # Python 3 + from urllib.parse import urlparse from dvc.command.base import CmdBase, DvcLock -from dvc.data_cloud import sizeof_fmt +from dvc.data_cloud import sizeof_fmt, file_md5 from dvc.logger import Logger from dvc.exceptions import DvcException from dvc.state_file import StateFile from dvc.system import System -from dvc.progress import progress -from dvc.utils import copyfile, map_progress class ImportFileError(DvcException): def __init__(self, msg): @@ -106,85 +109,27 @@ class CmdImportFile(CmdBase): if not os.path.exists(cache_dir): os.makedirs(cache_dir) - data_targets.append((input, output, data_item)) + data_targets.append((input, data_item)) return data_targets def import_files(self, targets, lock=False): data_targets = self.collect_targets(targets) - self.download_targets(data_targets) - self.create_state_files(data_targets, lock) - - def download_target(self, target): - """ - Download single target from url or from local path. - """ - input = target[0] - output = target[2].cache.relative - - if self.is_url(input): - Logger.debug("Downloading {} -> {}.".format(input, output)) - self.download_file(input, output) - Logger.debug("Done downloading {} -> {}.".format(input, output)) - else: - Logger.debug("Copying {} -> {}".format(input, output)) - self.copy_file(input, output) - Logger.debug("Dony copying {} -> {}".format(input, output)) - - @staticmethod - def copy_file(input, output): - """ - Copy single file from local path. - """ - copyfile(input, output) - - def download_file(self, from_url, to_file): - """ - Download single file from url. - """ - r = requests.get(from_url, stream=True) - - name = os.path.basename(from_url) - chunk_size = 1024 * 100 - downloaded = 0 - last_reported = 0 - report_bucket = 100*1024*10 - total_length = r.headers.get('content-length') - - with open(to_file, 'wb') as f: - for chunk in r.iter_content(chunk_size=chunk_size): - if not chunk: # filter out keep-alive new chunks - continue - - downloaded += chunk_size - - last_reported += chunk_size - if last_reported >= report_bucket: - last_reported = 0 - Logger.debug('Downloaded {}'.format(sizeof_fmt(downloaded))) - - # update progress bar - progress.update_target(name, downloaded, total_length) - - f.write(chunk) - - # tell progress bar that this target is finished downloading - progress.finish_target(name) - - def download_targets(self, targets): - """ - Download targets in a number of threads. - """ - map_progress(self.download_target, targets, self.parsed_args.jobs) + imported_targets = self.cloud.import_data(data_targets, self.parsed_args.jobs) + self.create_state_files(imported_targets, lock) def create_state_files(self, targets, lock): """ Create state files for all targets. """ for t in targets: - input = t[0] - output = t[1] - data_item = t[2] + orig_target, processed_data_item = t + input, data_item = orig_target + output = data_item.data.relative + + if processed_data_item == None: + Logger.debug('Skipping creating state file for failed import {}'.format(data_item.state.relative)) + continue Logger.debug('Creating symlink {} --> {}'.format(data_item.symlink_file, data_item.data.relative)) System.symlink(data_item.symlink_file, data_item.data.relative) @@ -199,14 +144,6 @@ class CmdImportFile(CmdBase): state_file.save() Logger.debug('State file "{}" was created'.format(data_item.state.relative)) - URL_REGEX = re.compile( - r'^(?:http|ftp)s?://' # http:// or https:// - r'(?:(?:[A-Z0-9](?:[A-Z0-9-]{0,61}[A-Z0-9])?\.)+(?:[A-Z]{2,6}\.?|[A-Z0-9-]{2,}\.?)|' # domain... - r'localhost|' # localhost... - r'\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})' # ...or ip - r'(?::\d+)?' # optional port - r'(?:/?|[/?]\S+)$', re.IGNORECASE) - @staticmethod def is_url(url): - return CmdImportFile.URL_REGEX.match(url) is not None + return len(urlparse(url).scheme) != 0 diff --git a/dvc/command/remove.py b/dvc/command/remove.py index 21e2f4e94..24c365467 100644 --- a/dvc/command/remove.py +++ b/dvc/command/remove.py @@ -2,7 +2,6 @@ import os from dvc.command.traverse import Traverse from dvc.logger import Logger -from dvc.path.data_item import DataItemError class CmdRemove(Traverse): @@ -12,11 +11,7 @@ class CmdRemove(Traverse): def process_file(self, target): Logger.debug(u'[Cmd-Remove] Remove file {}.'.format(target)) - try: - data_item = self.settings.path_factory.existing_data_item(target) - except DataItemError: - Logger.warn(u'[Cmd-Remove] Data file {} is not valid symbolic link'.format(target)) - data_item = self.settings.path_factory.data_item(target) + data_item = self._get_data_item(target) self._remove_cache_file(data_item) self._remove_state_file(data_item) @@ -26,10 +21,6 @@ class CmdRemove(Traverse): Logger.debug(u'[Cmd-Remove] Remove data item {}. Success.'.format(data_item.data.relative)) pass - def _remove_cloud_cache(self, data_item): - if not self.parsed_args.keep_in_cloud: - self.cloud.remove_from_cloud(data_item) - def _remove_state_file(self, data_item): if os.path.isfile(data_item.state.relative): self._remove_dvc_path(data_item.state, 'state') diff --git a/dvc/command/traverse.py b/dvc/command/traverse.py index 8dc07069a..63e030521 100644 --- a/dvc/command/traverse.py +++ b/dvc/command/traverse.py @@ -3,8 +3,7 @@ import os from dvc.command.base import CmdBase, DvcLock from dvc.exceptions import DvcException from dvc.logger import Logger -from dvc.data_cloud import DataCloud - +from dvc.path.data_item import DataItemError class TraverseError(DvcException): def __init__(self, msg): @@ -15,7 +14,6 @@ class Traverse(CmdBase): def __init__(self, settings, cmd_name, do_not_start_from_root=True): super(Traverse, self).__init__(settings) self._cmd_name = cmd_name - self.cloud = DataCloud(self.settings) self._do_not_start_from_root = do_not_start_from_root def run(self): @@ -69,6 +67,20 @@ class Traverse(CmdBase): self.traverse_dir_finalize(target) pass + # Common helpers + def _get_data_item(self, target): + try: + data_item = self.settings.path_factory.existing_data_item(target) + except DataItemError: + Logger.warn(u'[TraverseFileTree] Data file {} is not valid symbolic link'.format(target)) + data_item = self.settings.path_factory.data_item(target) + + return data_item + + def _remove_cloud_cache(self, data_item): + if not self.parsed_args.keep_in_cloud: + self.settings.cloud.remove(data_item) + ## API: def is_recursive(self): return True diff --git a/dvc/data_cloud.py b/dvc/data_cloud.py index f290e36cb..b075de06b 100644 --- a/dvc/data_cloud.py +++ b/dvc/data_cloud.py @@ -4,10 +4,20 @@ import os import threading import configparser import tempfile +import requests +import configparser +import filecmp +import math from boto.s3.connection import S3Connection from google.cloud import storage as gc +try: + from urlparse import urlparse +except ImportError: + # Python 3 + from urllib.parse import urlparse + import dvc from dvc.cloud.instance_manager import CloudSettings from dvc.logger import Logger @@ -16,8 +26,25 @@ from dvc.config import ConfigError from dvc.progress import progress from dvc.utils import copyfile from dvc.cloud.credentials_aws import AWSCredentials +from dvc.utils import cached_property +from dvc.system import System +from dvc.utils import map_progress +STATUS_UNKNOWN = 0 +STATUS_OK = 1 +STATUS_MODIFIED = 2 +STATUS_NEW = 3 +STATUS_DELETED = 4 + +STATUS_MAP = { + # (local_exists, remote_exists, cmp) + (True, True, True) : STATUS_OK, + (True, True, False) : STATUS_MODIFIED, + (True, False, None) : STATUS_NEW, + (False, True, None) : STATUS_DELETED, +} + class DataCloudError(DvcException): def __init__(self, msg): super(DataCloudError, self).__init__('Data sync error: {}'.format(msg)) @@ -31,7 +58,10 @@ def sizeof_fmt(num, suffix='B'): return "%.1f%s%s" % (num, 'Y', suffix) -def percent_cb(name, complete, total): +def percent_cb(name, part_complete, part_total, offset=0, multipart_total=None): + complete = offset + part_complete + total = multipart_total if multipart_total != None else part_total + Logger.debug('{}: {} transferred out of {}'.format( name, sizeof_fmt(complete), @@ -39,17 +69,20 @@ def percent_cb(name, complete, total): progress.update_target(os.path.basename(name), complete, total) -def create_cb(name): - return (lambda cur,tot: percent_cb(name, cur, tot)) +def create_cb(name, offset=0, multipart_total=None): + return (lambda cur,tot: percent_cb(name, cur, tot, offset, multipart_total)) def file_md5(fname): """ get the (md5 hexdigest, md5 digest) of a file """ - hash_md5 = hashlib.md5() - with open(fname, "rb") as f: - for chunk in iter(lambda: f.read(1024*1000), b""): - hash_md5.update(chunk) - return (hash_md5.hexdigest(), hash_md5.digest()) + if os.path.exists(fname): + hash_md5 = hashlib.md5() + with open(fname, "rb") as f: + for chunk in iter(lambda: f.read(1024*1000), b""): + hash_md5.update(chunk) + return (hash_md5.hexdigest(), hash_md5.digest()) + else: + return (None, None) class DataCloudBase(object): @@ -92,24 +125,39 @@ class DataCloudBase(object): def cache_file_key(self, file): return '{}/{}'.format(self.storage_prefix, file).strip('/') + def tmp_file(self, fname): + return fname + '.part' + def sanity_check(self): pass - def sync_to_cloud(self, item): + def _import(self, bucket, path, fname, item): pass - def sync_from_cloud(self, item): + def push(self, item): pass + def pull(self, item): + fname = item.resolved_cache.dvc + key_name = self.cache_file_key(fname) + + return self._import(self.storage_bucket, key_name, fname, item) + + def import_data(self, url, item): + o = urlparse(url) + + return self._import(o.netloc, o.path, item.cache.relative, item) + def sync(self, fname): item = self._cloud_settings.path_factory.data_item(fname) - if os.path.isfile(item.resolved_cache.dvc): - self.sync_to_cloud(item) - else: - self.create_directory(fname, item) - self.sync_from_cloud(item) - def create_directory(self, fname, item): + if os.path.isfile(item.resolved_cache.dvc): + return self.push(item) + else: + self.create_directory(item) + return self.pull(item) + + def create_directory(self, item): self._lock.acquire() try: dir = os.path.dirname(item.cache.relative) @@ -121,27 +169,158 @@ class DataCloudBase(object): raise DataCloudError(u'Cannot create directory {}: {}'.format(dir, ex)) elif not os.path.isdir(dir): msg = u'File {} cannot be synced because {} is not a directory' - raise DataCloudError(msg.format(fname, dir)) + raise DataCloudError(msg.format(item.cache.relative, dir)) finally: self._lock.release() - def remove_from_cloud(self, item): + def remove(self, item): pass + def _status(self, item): + pass + + def status(self, item): + return STATUS_MAP.get(self._status(item), STATUS_UNKNOWN) class DataCloudLOCAL(DataCloudBase): - def sync_to_cloud(self, item): + def push(self, item): Logger.debug('sync to cloud ' + item.resolved_cache.dvc + " " + self.storage_path) copyfile(item.resolved_cache.dvc, self.storage_path) + return item - def sync_from_cloud(self, item): + def _import(self, i, out, item): + tmp_file = self.tmp_file(out) + try: + copyfile(i, tmp_file) + os.rename(tmp_file, out) + except Exception as exc: + Logger.error('Failed to copy "{}": {}'.format(i, exc)) + return None + + return item + + def pull(self, item): Logger.debug('sync from cloud ' + self.storage_path + " " + item.resolved_cache.dvc) - copyfile(self.storage_path, item.resolved_cache.dvc) + return self._import(self.storage_path, item.resolved_cache.dvc, item) - def remove_from_cloud(self, item): + def remove(self, item): Logger.debug('rm from cloud ' + item.resolved_cache.dvc) os.remove(item.resolved_cache.dvc) + def import_data(self, path, item): + Logger.debug('import from cloud ' + path + " " + item.cache.relative) + return self._import(path, item.cache.relative, item) + + def _status(self, data_item): + local = data_item.resolved_cache.relative + remote = '{}/{}'.format(self.storage_path, os.path.basename(local)) + + remote_exists = os.path.exists(remove) + local_exists = os.path.exists(local) + c = None + if local_exists and remote_exists: + c = filecmp.cmp(local, remote) + + return (local_exists, remote_exists, c) + +class DataCloudHTTP(DataCloudBase): + def push(self, item): + raise Exception('Not implemented yet') + + def pull(self, item): + raise Exception('Not implemented yet') + + def remove(self, item): + raise Exception('Not implemented yet') + + def status(self, item): + raise Exception('Not implemented yet') + + def _downloaded_size(self, fname): + if os.path.exists(fname) and self.parsed_args.cont: + downloaded = os.path.getsize(fname) + header = {'Range': 'bytes=%d-' % downloaded} + + Logger.debug('found existing {} file, resuming download'.format(tmp_file)) + + return (downloaded, header) + + return (0, None) + + def _get_header(self, r, name): + val = r.headers.get(name) + if val == None: + Logger.debug('\'{}\' not supported by the server'.format(name)) + + return val + + def _verify_downloaded_size(self, r, downloaded_size): + content_range = self._get_header(r, 'content-range') + if downloaded_size and content_range == None: + Logger.debug('Can\'t resume download') + return 0 + + return downloaded_size + + def _download(self, r, fname, downloaded): + mode = 'ab' if downloaded else 'wb' + name = os.path.basename(r.url) + total_length = self._get_header(r, 'content-length') + chunk_size = 1024 * 100 + + progress.update_target(name, downloaded, total_length) + + with open(fname, mode) as f: + for chunk in r.iter_content(chunk_size=chunk_size): + if not chunk: # filter out keep-alive new chunks + continue + + f.write(chunk) + downloaded += len(chunk) + progress.update_target(name, downloaded, total_length) + + + progress.finish_target(name) + + def _verify_md5(self, r, fname): + md5 = file_md5(fname)[0] + content_md5 = self._get_header(r, 'content-md5') + + if content_md5 == None: + return True + + if md5 != content_md5: + Logger.error('Checksum mismatch') + return False + + Logger.debug('Checksum matches') + return True + + def import_data(self, url, item): + """ + Download single file from url. + """ + + to_file = item.cache.relative + tmp_file = self.tmp_file(to_file) + + downloaded, header = self._downloaded_size(tmp_file) + r = requests.get(url, stream=True, headers=header) + downloaded = self._verify_downloaded_size(r, downloaded) + + try: + self._download(r, tmp_file, downloaded) + except Exception as exc: + Logger.error('Failed to download "{}": {}'.format(url, exc)) + return None + + if not self._verify_md5(r, tmp_file): + return None + + os.rename(tmp_file, to_file) + + return item + class DataCloudAWS(DataCloudBase): """ DataCloud class for Amazon Web Services """ @@ -163,7 +342,64 @@ class DataCloudAWS(DataCloudBase): return 's3.amazonaws.com' return 's3.%s.amazonaws.com' % region - def _get_bucket_aws(self): + def get_aws_credentials(self): + """ gets aws credentials, looking in various places + + Params: + + Searches: + 1 any override in dvc.conf [AWS] CredentialPath; + 2 ~/.aws/credentials + + + Returns: + if successfully found, (access_key_id, secret) + None otherwise + """ + + # FIX: It won't work in Windows. + default = os.path.expanduser('~/.aws/credentials') + + paths = self.credential_paths(default) + for path in paths: + try: + cc = configparser.SafeConfigParser() + + # use readfp(open( ... to aid mocking. + cc.readfp(open(path, 'r')) + + if 'default' in cc.keys(): + access_key = cc['default'].get('aws_access_key_id', None) + secret = cc['default'].get('aws_secret_access_key', None) + + if access_key is not None and secret is not None: + return (access_key, secret) + except Exception as e: + pass + + return None + + def credential_paths(self, default): + paths = [] + credpath = self._cloud_config.get('CredentialPath', None) + if credpath is not None and len(credpath) > 0: + credpath = os.path.expanduser(credpath) + if os.path.isfile(credpath): + paths.append(credpath) + else: + Logger.warn('AWS CredentialPath "%s" not found; falling back to default "%s"' % (credpath, default)) + paths.append(default) + else: + paths.append(default) + return paths + + def sanity_check(self): + creds = self.get_aws_credentials() + if creds is None: + Logger.info("can't find aws credetials, assuming envirment variables or iam role") + self._aws_creds = creds + + def _get_bucket_aws(self, bucket_name): """ get a bucket object, aws """ if all([self._aws_creds.access_key_id, self._aws_creds.secret_access_key, self.aws_region_host]): conn = S3Connection(self._aws_creds.access_key_id, @@ -171,69 +407,124 @@ class DataCloudAWS(DataCloudBase): host=self.aws_region_host) else: conn = S3Connection() - bucket_name = self.storage_bucket bucket = conn.lookup(bucket_name) if bucket is None: raise DataCloudError('Storage path {} is not setup correctly'.format(bucket_name)) return bucket - def sync_from_cloud(self, item): - """ sync from cloud, aws version """ + def _cmp_checksum(self, key, fname): + md5_cloud = key.metadata.get('dvc-md5', None) + md5_local = file_md5(fname)[0] - bucket = self._get_bucket_aws() + if md5_cloud == None: + md5_cloud = key.etag[1:-1] - key_name = self.cache_file_key(item.resolved_cache.dvc) + if md5_cloud == md5_local: + return True + + return False + + def _import(self, bucket_name, key_name, fname, data_item): + + bucket = self._get_bucket_aws(bucket_name) + + tmp_file = self.tmp_file(fname) + name = os.path.basename(fname) key = bucket.get_key(key_name) if not key: Logger.error('File "{}" does not exist in the cloud'.format(key_name)) - return + return None - Logger.info('Downloading cache file from S3 "{}/{}"'.format(bucket.name, key_name)) + if self._cmp_checksum(key, fname): + Logger.debug('File "{}" matches with "{}".'.format(fname, key_name)) + return data_item + + Logger.debug('Downloading cache file from S3 "{}/{}" to "{}"'.format(bucket.name, key_name, fname)) temp_file = None try: - temp_file = tempfile.NamedTemporaryFile(dir=item.resolved_cache.dirname, delete=False) - key.get_contents_to_filename(temp_file.name, - cb=create_cb(item.resolved_cache.relative)) - os.rename(temp_file.name, item.resolved_cache.relative) + key.get_contents_to_filename(tmp_file, cb=create_cb(name)) + os.rename(tmp_file, fname) except Exception as exc: Logger.error('Failed to download "{}": {}'.format(key_name, exc)) - if temp_file and os.path.exists(temp_file.name): - os.remove(temp_file.name) - return + return None - progress.finish_target(os.path.basename(item.resolved_cache.relative)) - Logger.info('Downloading completed') + progress.finish_target(name) + Logger.debug('Downloading completed') - def sync_to_cloud(self, data_item): - """ sync_to_cloud, aws version """ + return data_item + + def _push_multipart(self, key, fname): + # AWS doesn't provide easilly accessible md5 for multipart + # objects, so we have to store our own md5 sum to use later. + metadata = {'dvc-md5' : str(file_md5(fname)[0])} + + mp = key.bucket.initiate_multipart_upload(key.name, metadata=metadata) + + source_size = os.stat(fname).st_size + chunk_size = 50*1024*1024 + chunk_count = int(math.ceil(source_size / float(chunk_size))) + + with open(fname, 'rb') as fp: + for i in range(chunk_count): + offset = i * chunk_size + left = source_size - offset + size = min([chunk_size, left]) + part_num = i + 1 + + fp.seek(offset) + mp.upload_part_from_file(fp=fp, + size=size, + num_cb=100, + part_num=part_num, + cb=create_cb(fname, offset, source_size)) + + if len(mp.get_all_parts()) != chunk_count: + raise Exception("Couldn't upload all file parts") + + mp.complete_upload() + + def push(self, data_item): + """ push, aws version """ aws_key = self.cache_file_key(data_item.resolved_cache.dvc) - bucket = self._get_bucket_aws() + bucket = self._get_bucket_aws(self.storage_bucket) key = bucket.get_key(aws_key) if key: Logger.debug('File already uploaded to the cloud. Checksum validation...') - md5_cloud = key.etag[1:-1] - md5_local = file_md5(data_item.resolved_cache.dvc)[0] - if md5_cloud == md5_local: + if self._cmp_checksum(key, data_item.resolved_cache.dvc): Logger.debug('File checksum matches. No uploading is needed.') - return + return data_item Logger.debug('Checksum miss-match. Re-uploading is required.') key = bucket.new_key(aws_key) try: - key.set_contents_from_filename(data_item.resolved_cache.relative, - cb=create_cb(data_item.resolved_cache.relative)) + self._push_multipart(key, data_item.resolved_cache.relative) except Exception as exc: Logger.error('Failed to upload "{}": {}'.format(data_item.resolved_cache.relative, exc)) - return + return None progress.finish_target(os.path.basename(data_item.resolved_cache.relative)) - def remove_from_cloud(self, data_item): + return data_item + + def _status(self, data_item): + aws_key = self.cache_file_key(data_item.resolved_cache.dvc) + bucket = self._get_bucket_aws(self.storage_bucket) + key = bucket.get_key(aws_key) + + remote_exists = key is not None + local_exists = os.path.exists(data_item.resolved_cache.relative) + c = None + if remote_exists and local_exists: + c = self._cmp_checksum(key, data_item.resolved_cache.dvc) + + return (local_exists, remote_exists, c) + + def remove(self, data_item): aws_file_name = self.cache_file_key(data_item.cache.dvc) Logger.debug(u'[Cmd-Remove] Remove from cloud {}.'.format(aws_file_name)) @@ -265,50 +556,94 @@ class DataCloudGCP(DataCloudBase): if project is None or len(project) < 1: raise ConfigError('can\'t read google cloud project name. Please set ProjectName in section GC.') - def _get_bucket_gc(self): + def _get_bucket_gc(self, storage_bucket): """ get a bucket object, gc """ client = gc.Client(project=self.gc_project_name) - bucket = client.bucket(self.storage_bucket) + bucket = client.bucket(storage_bucket) if not bucket.exists(): raise DataCloudError('sync up: google cloud bucket {} doesn\'t exist'.format(self.storage_bucket)) return bucket - def sync_from_cloud(self, item): - """ sync from cloud, gcp version """ + def _cmp_checksum(self, blob, fname): + b64_encoded_md5 = base64.b64encode(file_md5(fname)[1]) - bucket = self._get_bucket_gc() - key = self.cache_file_key(item.resolved_cache.dvc) + if blob.md5_hash == b64_encoded_md5: + return True + + return False + + def _import(self, bucket_name, key, fname, data_item): + + bucket = self._get_bucket_gc(bucket_name) + + name = os.path.basename(fname) + tmp_file = self.tmp_file(fname) blob = bucket.get_blob(key) if not blob: Logger.error('File "{}" does not exist in the cloud'.format(key)) - return + return None Logger.info('Downloading cache file from gc "{}/{}"'.format(bucket.name, key)) - blob.download_to_filename(item.resolved_cache.dvc) + # percent_cb is not available for download_to_filename, so + # lets at least update progress at keypoints(start, finish) + progress.update_target(name, 0, None) + + try: + blob.download_to_filename(tmp_file) + os.rename(tmp_file, fname) + except Exception as exc: + Logger.error('Failed to download "{}": {}'.format(key, exc)) + return None + + progress.finish_target(name) + Logger.info('Downloading completed') - def sync_to_cloud(self, data_item): - """ sync_to_cloud, gcp version """ + return data_item - bucket = self._get_bucket_gc() + def push(self, data_item): + """ push, gcp version """ + + bucket = self._get_bucket_gc(self.storage_bucket) blob_name = self.cache_file_key(data_item.resolved_cache.dvc) + name = os.path.basename(data_item.resolved_cache.dvc) blob = bucket.get_blob(blob_name) if blob is not None and blob.exists(): - b64_encoded_md5 = base64.b64encode(file_md5(data_item.resolved_cache.dvc)[1]) - - if blob.md5_hash == b64_encoded_md5: + if self._cmp_checksum(blob, data_item.resolved_cache.dvc): Logger.debug('checksum %s matches. Skipping upload' % data_item.cache.relative) - return + return data_item Logger.debug('checksum %s mismatch. re-uploading' % data_item.cache.relative) + # same as in _import + progress.update_target(name, 0, None) + blob = bucket.blob(blob_name) blob.upload_from_filename(data_item.resolved_cache.relative) + + progress.finish_target(name) Logger.info('uploading %s completed' % data_item.resolved_cache.relative) - def remove_from_cloud(self, item): + return data_item + + def _status(self, data_item): + """ status, gcp version """ + + bucket = self._get_bucket_gc(self.storage_bucket) + blob_name = self.cache_file_key(data_item.resolved_cache.dvc) + blob = bucket.get_blob(blob_name) + + remote_exists = blob is not None and blob.exists() + local_exists = os.path.exists(data_item.resolved_cache.relative) + c = None + if remote_exists and local_exists: + c = self._cmp_checksum(blob, data_item.resolved_cache.dvc) + + return (local_exists, remote_exists, c) + + def remove(self, item): raise Exception('NOT IMPLEMENTED YET') @@ -316,9 +651,19 @@ class DataCloud(object): """ Generic class to do initial config parsing and redirect to proper DataCloud methods """ CLOUD_MAP = { - 'AWS': DataCloudAWS, - 'GCP': DataCloudGCP, - 'LOCAL': DataCloudLOCAL, + 'AWS' : DataCloudAWS, + 'GCP' : DataCloudGCP, + 'HTTP' : DataCloudHTTP, + 'LOCAL' : DataCloudLOCAL, + } + + SCHEME_MAP = { + 's3' : 'AWS', + 'http' : 'HTTP', + 'https' : 'HTTP', + 'ftp' : 'HTTP', + 'gs' : 'GCP', + '' : 'LOCAL', } def __init__(self, settings): @@ -376,8 +721,72 @@ class DataCloud(object): self._cloud.sanity_check() - def sync(self, fname): - return self._cloud.sync(fname) + def _collect_dir(self, d): + targets = [] - def remove_from_cloud(self, item): - return self._cloud.remove_from_cloud(item) + for root, dirs, files in os.walk(d): + for f in files: + path = os.path.join(root, f) + item = self._settings.path_factory.data_item(path) + targets.append(item) + + return targets + + def _collect_target(self, target): + if System.islink(target): + item = self._settings.path_factory.data_item(target) + return [item] + elif os.path.isdir(target): + return self._collect_dir(target) + + Logger.warn('Target "{}" does not exist'.format(target)) + + return [] + + def _collect_targets(self, targets): + collected = [] + + for t in targets: + collected += self._collect_target(t) + + return collected + + def _map_targets(self, f, targets, jobs): + collected = self._collect_targets(targets) + + return map_progress(f, collected, jobs) + + def _import(self, target): + url, item = target + o = urlparse(url) + + typ = self.SCHEME_MAP.get(o.scheme, None) + if typ == None: + Logger.error('Not supported scheme \'{}\''.format(o.scheme)) + return None + + cloud_config = None + if self._config.has_section(typ): + cloud_config = self._config[typ] + + cloud = self.CLOUD_MAP[typ](self._settings, self._config, cloud_config) + + return cloud.import_data(url, item) + + def sync(self, targets, jobs=1): + return self._map_targets(self._cloud.sync, targets, jobs) + + def push(self, targets, jobs=1): + return self._map_targets(self._cloud.push, targets, jobs) + + def pull(self, targets, jobs=1): + return self._map_targets(self._cloud.pull, targets, jobs) + + def import_data(self, targets, jobs=1): + return map_progress(self._import, targets, jobs) + + def remove(self, item): + return self._cloud.remove(item) + + def status(self, targets, jobs=1): + return self._map_targets(self._cloud.status, targets, jobs) diff --git a/dvc/git_wrapper.py b/dvc/git_wrapper.py index f466db785..4cc5d5e32 100644 --- a/dvc/git_wrapper.py +++ b/dvc/git_wrapper.py @@ -46,6 +46,15 @@ class GitWrapperI(object): return GitWrapper.parse_porcelain_files(out) + @staticmethod + def git_config_get(name): + code, out, err = Executor.exec_cmd(['git', 'config', '--get', name]) + Logger.debug('[dvc-git] "git config --get {}": code({}), out({}), err({})'.format( + name, code, out, err)) + if code != 0: + return None + return out + @staticmethod def git_path_to_system_path(path): if os.name == 'nt': @@ -86,6 +95,17 @@ class GitWrapper(GitWrapperI): Logger.error("{} {}".format(status, file)) return False + # Sanity check to make sure we will be able to commit + name = self.git_config_get('user.name') + if name == None: + Logger.error('[dvc-git] Please setup user.name in git config') + return False + + email = self.git_config_get('user.email') + if email == None: + Logger.error('[dvc-git] Please setup user.email in git config') + return False + return True @property diff --git a/dvc/path/data_item.py b/dvc/path/data_item.py index 04fb341a1..90e2fbff4 100644 --- a/dvc/path/data_item.py +++ b/dvc/path/data_item.py @@ -62,6 +62,9 @@ class DataItem(object): return self.data.dvc.__hash__() def __eq__(self, other): + if other == None: + return False + return self.data.dvc == other.data.dvc @property diff --git a/dvc/settings.py b/dvc/settings.py index a20a4784a..7ac62581b 100644 --- a/dvc/settings.py +++ b/dvc/settings.py @@ -5,6 +5,7 @@ from dvc.config import ConfigI, Config from dvc.exceptions import DvcException from dvc.path.factory import PathFactory from dvc.cli import parse_args +from dvc.data_cloud import DataCloud class SettingsError(DvcException): def __init__(self, msg): @@ -16,15 +17,15 @@ class Settings(object): self._args = None args = None - if argv != None and len(argv) != 0: + if argv is not None and len(argv) != 0: args = parse_args(argv) self._args = argv[2:] - if git == None: + if git is None: git = GitWrapper() - if config == None: - if args != None and args.cmd != 'init': + if config is None: + if args is not None and args.cmd != 'init': config = Config() else: config = ConfigI() @@ -33,6 +34,7 @@ class Settings(object): self._config = config self._path_factory = PathFactory(git, config) self._parsed_args = args + self._cloud = DataCloud(self) @property def args(self): @@ -63,3 +65,7 @@ class Settings(object): @property def path_factory(self): return self._path_factory + + @property + def cloud(self): + return self._cloud diff --git a/dvc/utils.py b/dvc/utils.py index bb4e8e11f..f2045f76b 100644 --- a/dvc/utils.py +++ b/dvc/utils.py @@ -68,8 +68,10 @@ def map_progress(func, targets, n_threads): p = ThreadPool(processes=n_threads) try: - p.map(func, targets) + ret = p.map(func, targets) except Exception as exc: Logger.error('Unexpected exception while processing targets: {}'.format(exc)) finally: progress.finish() + + return zip(targets, ret) diff --git a/functest/clean.sh b/functest/clean.sh index f75d4f11e..8d39fe6d4 100755 --- a/functest/clean.sh +++ b/functest/clean.sh @@ -1,5 +1,5 @@ +#!/bin/bash -source common.rc - -rm -rf $RAW_DATA_LOCAL $CODE_LOCAL $TEST_REPO +source common.sh +rm -rf $CACHE_DIR $TEST_REPO diff --git a/functest/common.rc b/functest/common.rc deleted file mode 100644 index dda00a5f8..000000000 --- a/functest/common.rc +++ /dev/null @@ -1,35 +0,0 @@ - -BASE_DIR_S3=https://s3-us-west-2.amazonaws.com/dataversioncontrol/functests -RAW_DATA_S3=$BASE_DIR_S3/stackoverflow_raw_small - -RAW_DATA_LOCAL=sync_raw_data -CODE_LOCAL=sync_code -TEST_REPO=repo_test - -function dvc_info() { - echo "=============== DVC test: $1 ===============" -} - -function _init_repo() { - dvc_info 'init reporitory' - git init - dvc init - - dvc_info 'import xml files' - mkdir data/xml - dvc data-import ../$RAW_DATA_LOCAL/* data/xml - - dvc_info 'copy code' - mkdir code - cp ../$CODE_LOCAL/* code/ - git add code/ - git commit -m 'Add code' -} - -function create_repo() { - dvc_info 'create reporitory' - rm -rf $TEST_REPO - mkdir $TEST_REPO - (cd $TEST_REPO && _init_repo) -} - diff --git a/functest/common.sh b/functest/common.sh new file mode 100644 index 000000000..094296428 --- /dev/null +++ b/functest/common.sh @@ -0,0 +1,120 @@ +#!/bin/bash + +set -e + +SDIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" && pwd )" +CACHE_DIR=$SDIR/cache +TEST_REPO=$SDIR/myrepo +CODE_CACHE=$CACHE_DIR/code +DATA_CACHE=$CACHE_DIR/data + +DATA="Badges.xml Tags.xml PostLinks.xml" +CODE="xmltotsv.py" + +REGION='us-west-2' +BASE_DIR_S3=https://s3-$REGION.amazonaws.com/dataversioncontrol/functests +DATA_S3=$BASE_DIR_S3/stackoverflow_raw_small +CODE_S3=$BASE_DIR_S3/code + +function dvc_print() { + echo -e "$1" +} + +function dvc_info() { + dvc_print "\e[34m$1\e[0m" +} + +function dvc_header() { + dvc_info "=============== DVC test: $1 ===============" +} + +function dvc_error() { + dvc_print "\e[31m$1\e[0m" +} + +function dvc_fail() { + dvc_error "FAIL" + exit 1 +} + +function dvc_fatal() { + dvc_error "$1" + dvc_fail +} + +function dvc_pass() { + dvc_print "\e[32mPASS\e[0m" + exit 0 +} + +trap 'dvc_fail;' ERR + +function dvc_prepare_cache() { + if [ -d $CACHE_DIR ]; then + return + fi + + mkdir -p $DATA_CACHE + pushd $DATA_CACHE + + for f in $DATA; do + wget $DATA_S3/$f + done + + popd + + mkdir -p $CODE_CACHE + pushd $CODE_CACHE + + for f in $CODE; do + wget $CODE_S3/$f + done + + popd +} + +function dvc_create_git_repo() { + dvc_info "Creating git repo" + + dvc_prepare_cache + + mkdir $TEST_REPO + cd $TEST_REPO + + git init + + git config user.name "DVC tester" + git config user.email "dvctester@dataversioncontrol.com" + + mkdir code + cp $CODE_CACHE/* code/ + git add code/ + git commit -m 'Add code' +} + +function dvc_create_repo() { + dvc_info "Creating dvc repo" + + dvc_create_git_repo + + dvc init + + mkdir data/xml + dvc import $DATA_CACHE/* data/xml +} + +function dvc_check_dirs() { + for d in $1; do + if [ ! -d "$d" ]; then + dvc_fatal "Directory $d doesn't exist" + fi + done +} + +function dvc_check_files() { + for f in $1; do + if [ ! -f "$f" ]; then + dvc_fatal "File $f doesn't exist" + fi + done +} diff --git a/functest/init.sh b/functest/init.sh deleted file mode 100755 index 7ef46b650..000000000 --- a/functest/init.sh +++ /dev/null @@ -1,16 +0,0 @@ - -source common.rc - -./clean.sh - -rm -rf $RAW_DATA_LOCAL -mkdir $RAW_DATA_LOCAL -for file in "Badges.xml" "Tags.xml" "PostLinks.xml" -do - (cd $RAW_DATA_LOCAL; wget $RAW_DATA_S3/$file) -done - -rm -rf $CODE_LOCAL -mkdir $CODE_LOCAL -(cd $CODE_LOCAL; wget "$BASE_DIR_S3/code/xmltotsv.py") - diff --git a/functest/run.sh b/functest/run.sh new file mode 100755 index 000000000..e6d3ef6db --- /dev/null +++ b/functest/run.sh @@ -0,0 +1,17 @@ +#!/bin/bash + +set -e + +source common.sh + +TESTS+=" test-init.sh" +TESTS+=" test-import.sh" +TESTS+=" test-repro-code.sh" + +for t in $TESTS; do + rm -rf $TEST_REPO + dvc_header "$t" + ./$t +done + +rm -rf $TEST_REPO diff --git a/functest/test-import.sh b/functest/test-import.sh new file mode 100755 index 000000000..d57e1294a --- /dev/null +++ b/functest/test-import.sh @@ -0,0 +1,26 @@ +#!/bin/bash + +set -e + +source common.sh + +dvc_create_repo + +dvc_info "Import from local file" +dvc import $DATA_CACHE/Tags.xml data/local +dvc_check_files "data/local" + +dvc_info "Import from http url" +dvc import $DATA_S3/Tags.xml data/http +dvc_check_files "data/http" + +# Temporarily disabled for travis builds, as +# they don't have aws credentials setup yet. +if [ -z $TRAVIS ]; then + dvc_info "Import from s3 url" + dvc config aws.region $REGION + dvc import s3://dataversioncontrol/functests/stackoverflow_raw_small/Tags.xml data/s3 + dvc_check_files "data/s3" +fi + +dvc_pass diff --git a/functest/test-init.sh b/functest/test-init.sh new file mode 100755 index 000000000..e291597db --- /dev/null +++ b/functest/test-init.sh @@ -0,0 +1,17 @@ +#!/bin/bash + +set -e + +source common.sh + +dvc_create_git_repo + +dvc init + +DIRS="data .cache .state" +FILES="dvc.conf" + +dvc_check_dirs $DIRS +dvc_check_files $FILES + +dvc_pass diff --git a/functest/test-repro-code.sh b/functest/test-repro-code.sh index 479131f0a..58d071f3d 100755 --- a/functest/test-repro-code.sh +++ b/functest/test-repro-code.sh @@ -1,20 +1,22 @@ #!/bin/bash -source common.rc +set -e -function run_test() { - mkdir data/tsv +source common.sh - dvc_info 'convert Budget.xml to tsv' - dvc run --code code/xmltotsv.py python code/xmltotsv.py --extract row/@Id,row/@UserId,row/@Name,row/@Class,row/@TagBased,row/@Date data/xml/Badges.xml data/tsv/Badges.tsv +dvc_create_repo + +mkdir data/tsv + +dvc_info 'Convert Budget.xml to tsv' +pip2 install xmltodict +dvc run --code code/xmltotsv.py python2 code/xmltotsv.py --extract row/@Id,row/@UserId,row/@Name,row/@Class,row/@TagBased,row/@Date data/xml/Badges.xml data/tsv/Badges.tsv - dvc_info 'modify code' - echo " " >> code/xmltotsv.py - git commit -am 'Change code' +dvc_info 'Modify code' +echo " " >> code/xmltotsv.py +git commit -am 'Change code' - dvc_info 'reproduce tsv convertion code' - dvc repro data/tsv/Badges.tsv -} +dvc_info 'Reproduce tsv convertion code' +dvc repro data/tsv/Badges.tsv -create_repo -(cd $TEST_REPO; run_test) +dvc_pass diff --git a/functest/tutorial_end_to_end.sh b/functest/tutorial_end_to_end.sh index 26ffdb58c..e209023d6 100755 --- a/functest/tutorial_end_to_end.sh +++ b/functest/tutorial_end_to_end.sh @@ -1,3 +1,6 @@ +#!/bin/bash + +set -e # 1. First ML model diff --git a/functests.sh b/functests.sh new file mode 100755 index 000000000..4c55eee8c --- /dev/null +++ b/functests.sh @@ -0,0 +1,11 @@ +#!/bin/bash + +DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" && pwd )" +BIN=$DIR/bin +DVC_HOME=$DIR + +export PATH=$BIN:$PATH +export DVC_HOME=$DIR + +cd functest +./run.sh diff --git a/requirements.txt b/requirements.txt index 253e3144e..8e3d4ad16 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,86 +1,89 @@ -altgraph -appdirs -backports.shutil-get-terminal-size -boto -cachetools -configparser -decorator -dill -enum34 -fasteners -funcsigs -future -futures -gapic-google-cloud-datastore-v1 -gapic-google-cloud-error-reporting-v1beta1 -gapic-google-cloud-logging-v2 -gapic-google-cloud-pubsub-v1 -gapic-google-cloud-spanner-admin-database-v1 -gapic-google-cloud-spanner-admin-instance-v1 -gapic-google-cloud-spanner-v1 -gapic-google-cloud-speech-v1beta1 -gapic-google-cloud-vision-v1 -google-auth -google-auth-httplib2 -google-cloud -google-cloud-bigquery -google-cloud-bigtable -google-cloud-core -google-cloud-datastore -google-cloud-dns -google-cloud-error-reporting -google-cloud-language -google-cloud-logging -google-cloud-monitoring -google-cloud-pubsub -google-cloud-resource-manager -google-cloud-runtimeconfig -google-cloud-spanner -google-cloud-speech -google-cloud-storage -google-cloud-translate -google-cloud-vision -google-gax -googleapis-common-protos -grpc-google-iam-v1 -grpcio -httplib2 -ipython -ipython-genutils -macholib -mock -modulegraph -monotonic -nose -oauth2client -packaging -pbr -pexpect -pickleshare -ply -prompt-toolkit -proto-google-cloud-datastore-v1 -proto-google-cloud-error-reporting-v1beta1 -proto-google-cloud-logging-v2 -proto-google-cloud-pubsub-v1 -proto-google-cloud-spanner-admin-database-v1 -proto-google-cloud-spanner-admin-instance-v1 -proto-google-cloud-spanner-v1 -proto-google-cloud-speech-v1beta1 -proto-google-cloud-vision-v1 -protobuf -ptyprocess -pyasn1 -pyasn1-modules -Pygments -PyInstaller -pyparsing -requests -rsa -scandir -simplegeneric -six -traitlets -wcwidth -colorama -google-compute-engine +altgraph==0.13 +appdirs==1.4.3 +backports.shutil-get-terminal-size==1.0.0 +boto==2.46.1 +cachetools==2.0.0 +configparser==3.5.0 +decorator==4.0.11 +dill==0.2.6 +enum34==1.1.6 +fasteners==0.14.1 +funcsigs==1.0.2 +future==0.16.0 +futures==3.0.5 +gapic-google-cloud-datastore-v1==0.15.3 +gapic-google-cloud-error-reporting-v1beta1==0.15.3 +gapic-google-cloud-logging-v2==0.91.3 +gapic-google-cloud-pubsub-v1==0.15.3 +gapic-google-cloud-spanner-admin-database-v1==0.15.3 +gapic-google-cloud-spanner-admin-instance-v1==0.15.3 +gapic-google-cloud-spanner-v1==0.15.3 +gapic-google-cloud-speech-v1beta1==0.15.3 +gapic-google-cloud-vision-v1==0.90.3 +google-auth==0.10.0 +google-auth-httplib2==0.0.2 +google-cloud==0.24.0 +google-cloud-bigquery==0.24.0 +google-cloud-bigtable==0.24.0 +google-cloud-core==0.24.0 +google-cloud-datastore==1.0.0 +google-cloud-dns==0.24.0 +google-cloud-error-reporting==0.24.0 +google-cloud-language==0.24.0 +google-cloud-logging==0.24.0 +google-cloud-monitoring==0.24.0 +google-cloud-pubsub==0.24.0 +google-cloud-resource-manager==0.24.0 +google-cloud-runtimeconfig==0.24.0 +google-cloud-spanner==0.24.0 +google-cloud-speech==0.24.0 +google-cloud-storage==1.0.0 +google-cloud-translate==0.24.0 +google-cloud-vision==0.24.0 +google-gax==0.15.8 +googleapis-common-protos==1.5.2 +grpc-google-iam-v1==0.11.1 +grpcio==1.2.1 +httplib2==0.10.3 +ipython==5.3.0 +ipython-genutils==0.2.0 +macholib==1.8 +mock==2.0.0 +modulegraph==0.14 +monotonic==1.3 +nose==1.3.7 +oauth2client==3.0.0 +packaging==16.8 +pathlib==1.0.1 +pathlib2==2.2.1 +pbr==3.0.0 +pexpect==4.2.1 +pickleshare==0.7.4 +ply==3.8 +prompt-toolkit==1.0.14 +proto-google-cloud-datastore-v1==0.90.3 +proto-google-cloud-error-reporting-v1beta1==0.15.3 +proto-google-cloud-logging-v2==0.91.3 +proto-google-cloud-pubsub-v1==0.15.3 +proto-google-cloud-spanner-admin-database-v1==0.15.3 +proto-google-cloud-spanner-admin-instance-v1==0.15.3 +proto-google-cloud-spanner-v1==0.15.3 +proto-google-cloud-speech-v1beta1==0.15.3 +proto-google-cloud-vision-v1==0.90.3 +protobuf==3.2.0 +ptyprocess==0.5.1 +pyasn1==0.2.3 +pyasn1-modules==0.0.8 +Pygments==2.2.0 +PyInstaller==3.2.1 +pyparsing==2.2.0 +requests==2.13.0 +rsa==3.4.2 +scandir==1.5 +simplegeneric==0.8.1 +six==1.10.0 +traitlets==4.3.2 +wcwidth==0.1.7 +colorama==0.3.9 +google-compute-engine==2.4.1 +configobj==5.0.6 diff --git a/setup.py b/setup.py index 0821df687..d47341ba2 100644 --- a/setup.py +++ b/setup.py @@ -4,97 +4,100 @@ from distutils.errors import DistutilsPlatformError from dvc import VERSION install_requires = [ - 'altgraph', - 'appdirs', - 'backports.shutil-get-terminal-size', - 'boto', - 'cachetools', - 'configparser', - 'decorator', - 'dill', - 'enum34', - 'fasteners', - 'funcsigs', - 'future', - 'futures', - 'gapic-google-cloud-datastore-v1', - 'gapic-google-cloud-error-reporting-v1beta1', - 'gapic-google-cloud-logging-v2', - 'gapic-google-cloud-pubsub-v1', - 'gapic-google-cloud-spanner-admin-database-v1', - 'gapic-google-cloud-spanner-admin-instance-v1', - 'gapic-google-cloud-spanner-v1', - 'gapic-google-cloud-speech-v1beta1', - 'gapic-google-cloud-vision-v1', - 'google-auth', - 'google-auth-httplib2', - 'google-cloud', - 'google-cloud-bigquery', - 'google-cloud-bigtable', - 'google-cloud-core', - 'google-cloud-datastore', - 'google-cloud-dns', - 'google-cloud-error-reporting', - 'google-cloud-language', - 'google-cloud-logging', - 'google-cloud-monitoring', - 'google-cloud-pubsub', - 'google-cloud-resource-manager', - 'google-cloud-runtimeconfig', - 'google-cloud-spanner', - 'google-cloud-speech', - 'google-cloud-storage', - 'google-cloud-translate', - 'google-cloud-vision', - 'google-gax', - 'googleapis-common-protos', - 'grpc-google-iam-v1', - 'grpcio', - 'httplib2', - 'ipython', - 'ipython-genutils', - 'macholib', - 'mock', - 'modulegraph', - 'monotonic', - 'nose', - 'oauth2client', - 'packaging', - 'pbr', - 'pexpect', - 'pickleshare', - 'ply', - 'prompt-toolkit', - 'proto-google-cloud-datastore-v1', - 'proto-google-cloud-error-reporting-v1beta1', - 'proto-google-cloud-logging-v2', - 'proto-google-cloud-pubsub-v1', - 'proto-google-cloud-spanner-admin-database-v1', - 'proto-google-cloud-spanner-admin-instance-v1', - 'proto-google-cloud-spanner-v1', - 'proto-google-cloud-speech-v1beta1', - 'proto-google-cloud-vision-v1', - 'protobuf', - 'ptyprocess', - 'pyasn1', - 'pyasn1-modules', - 'Pygments', - 'PyInstaller', - 'pyparsing', - 'requests', - 'rsa', - 'scandir', - 'simplegeneric', - 'six', - 'traitlets', - 'wcwidth', - 'colorama', - 'google-compute-engine' + 'altgraph==0.13', + 'appdirs==1.4.3', + 'backports.shutil-get-terminal-size==1.0.0', + 'boto==2.46.1', + 'cachetools==2.0.0', + 'configparser==3.5.0', + 'decorator==4.0.11', + 'dill==0.2.6', + 'enum34==1.1.6', + 'fasteners==0.14.1', + 'funcsigs==1.0.2', + 'future==0.16.0', + 'futures==3.0.5', + 'gapic-google-cloud-datastore-v1==0.15.3', + 'gapic-google-cloud-error-reporting-v1beta1==0.15.3', + 'gapic-google-cloud-logging-v2==0.91.3', + 'gapic-google-cloud-pubsub-v1==0.15.3', + 'gapic-google-cloud-spanner-admin-database-v1==0.15.3', + 'gapic-google-cloud-spanner-admin-instance-v1==0.15.3', + 'gapic-google-cloud-spanner-v1==0.15.3', + 'gapic-google-cloud-speech-v1beta1==0.15.3', + 'gapic-google-cloud-vision-v1==0.90.3', + 'google-auth==0.10.0', + 'google-auth-httplib2==0.0.2', + 'google-cloud==0.24.0', + 'google-cloud-bigquery==0.24.0', + 'google-cloud-bigtable==0.24.0', + 'google-cloud-core==0.24.0', + 'google-cloud-datastore==1.0.0', + 'google-cloud-dns==0.24.0', + 'google-cloud-error-reporting==0.24.0', + 'google-cloud-language==0.24.0', + 'google-cloud-logging==0.24.0', + 'google-cloud-monitoring==0.24.0', + 'google-cloud-pubsub==0.24.0', + 'google-cloud-resource-manager==0.24.0', + 'google-cloud-runtimeconfig==0.24.0', + 'google-cloud-spanner==0.24.0', + 'google-cloud-speech==0.24.0', + 'google-cloud-storage==1.0.0', + 'google-cloud-translate==0.24.0', + 'google-cloud-vision==0.24.0', + 'google-gax==0.15.8', + 'googleapis-common-protos==1.5.2', + 'grpc-google-iam-v1==0.11.1', + 'grpcio==1.2.1', + 'httplib2==0.10.3', + 'ipython==5.3.0', + 'ipython-genutils==0.2.0', + 'macholib==1.8', + 'mock==2.0.0', + 'modulegraph==0.14', + 'monotonic==1.3', + 'nose==1.3.7', + 'oauth2client==3.0.0', + 'packaging==16.8', + 'pathlib==1.0.1', + 'pathlib2==2.2.1', + 'pbr==3.0.0', + 'pexpect==4.2.1', + 'pickleshare==0.7.4', + 'ply==3.8', + 'prompt-toolkit==1.0.14', + 'proto-google-cloud-datastore-v1==0.90.3', + 'proto-google-cloud-error-reporting-v1beta1==0.15.3', + 'proto-google-cloud-logging-v2==0.91.3', + 'proto-google-cloud-pubsub-v1==0.15.3', + 'proto-google-cloud-spanner-admin-database-v1==0.15.3', + 'proto-google-cloud-spanner-admin-instance-v1==0.15.3', + 'proto-google-cloud-spanner-v1==0.15.3', + 'proto-google-cloud-speech-v1beta1==0.15.3', + 'proto-google-cloud-vision-v1==0.90.3', + 'protobuf==3.2.0', + 'ptyprocess==0.5.1', + 'pyasn1==0.2.3', + 'pyasn1-modules==0.0.8', + 'Pygments==2.2.0', + 'PyInstaller==3.2.1', + 'pyparsing==2.2.0', + 'requests==2.13.0', + 'rsa==3.4.2', + 'scandir==1.5', + 'simplegeneric==0.8.1', + 'six==1.10.0', + 'traitlets==4.3.2', + 'wcwidth==0.1.7', + 'colorama==0.3.9', + 'google-compute-engine==2.4.1', + 'configobj==5.0.6', ] if platform.system() == 'Darwin': - install_requires.append('appnope') - install_requires.append('py2app') + install_requires.append('appnope==0.1.0') + install_requires.append('py2app==0.12') setup( name='dvc', diff --git a/test-requirements.txt b/test-requirements.txt index 4ebc8aea5..21329b8d1 100644 --- a/test-requirements.txt +++ b/test-requirements.txt @@ -1 +1,2 @@ coverage +xmltodict diff --git a/tests/test_utils.py b/tests/test_utils.py new file mode 100644 index 000000000..ba2862a34 --- /dev/null +++ b/tests/test_utils.py @@ -0,0 +1,60 @@ +import os +import shutil +import filecmp +from random import randint +from time import sleep + +from dvc import utils +from tests.basic_env import DirHierarchyEnvironment + +class TestExecutor(DirHierarchyEnvironment): + def setUp(self): + DirHierarchyEnvironment.init_environment(self) + + def tearDown(self): + pass + + def test_rmtree(self): + root = 'testdir' + + os.makedirs(root + '/subdir') + with open(root + '/file1', 'w+') as f: + f.write('file1contents') + with open(root + '/subdir/file2', 'w+') as f: + f.write('file2contents') + + utils.rmtree(root) + self.assertFalse(os.path.exists(root)) + + def test_copyfile(self): + src = 'file1' + dest = 'file2' + dest_dir = 'testdir' + + with open(src, 'w+') as f: + f.write('file1contents') + + os.mkdir(dest_dir) + + utils.copyfile(src, dest) + self.assertTrue(filecmp.cmp(src, dest)) + + utils.copyfile(src, dest_dir) + self.assertTrue(filecmp.cmp(src, '{}/{}'.format(dest_dir, src))) + + shutil.rmtree(dest_dir) + os.remove(src) + os.remove(dest) + + def test_map_progress(self): + def f(target): + sleep(randint(1,2)) + with open(target, 'w+') as o: + o.write(target) + sleep(randint(1,2)) + + targets = ['map{}'.format(i) for i in range(1, 10)] + n_threads = [1, 10, 20] + + for n in n_threads: + utils.map_progress(f, targets, n)