From 73039f00f23b30a9caa671d979f1bd2cd8876385 Mon Sep 17 00:00:00 2001 From: Dmitry Petrov Date: Sun, 2 Apr 2017 19:35:15 -0700 Subject: [PATCH] Run cmd refactoring --- dvc/command/repro.py | 8 ++- dvc/command/run.py | 113 ++++++++++++++------------------ dvc/path/data_item.py | 16 +++++ dvc/path/factory.py | 4 ++ dvc/path/stated_data_item.py | 39 +++++++++++ dvc/repository_change.py | 105 ++++++++++------------------- tests/test_repository_change.py | 100 ++++++++++++++++++---------- 7 files changed, 214 insertions(+), 171 deletions(-) create mode 100644 dvc/path/stated_data_item.py diff --git a/dvc/command/repro.py b/dvc/command/repro.py index 89493cfde..99175d4f8 100644 --- a/dvc/command/repro.py +++ b/dvc/command/repro.py @@ -37,11 +37,11 @@ class CmdRepro(CmdRun): # Overridden methods: @property - def declaration_input_files(self): + def declaration_input_data_items(self): return [] @property - def declaration_output_files(self): + def declaration_output_data_items(self): return [] def run(self): @@ -129,7 +129,9 @@ class ReproChange(object): Logger.debug('Reproducing data file "{}". Re-runs cmd: {}'.format( self._data_item.data.relative, ' '.join(self._repro_argv))) - return self._cmd_obj.run_command(self._repro_argv) + + data_items_from_args = self._cmd_obj._data_items_from_args(self._repro_argv) + return self._cmd_obj.run_command(self._repro_argv, data_items_from_args) def reproduce(self, force=False): were_input_files_changed = False diff --git a/dvc/command/run.py b/dvc/command/run.py index 6f2e1ccd8..6a2f6ac93 100644 --- a/dvc/command/run.py +++ b/dvc/command/run.py @@ -11,6 +11,7 @@ from dvc.exceptions import DvcException from dvc.repository_change import RepositoryChange from dvc.state_file import StateFile from dvc.utils import run +from dvc.utils import cached_property class RunError(DvcException): @@ -29,10 +30,10 @@ class CmdRun(CmdBase): parser.add_argument('--random', help='not reproducible, output is random', action='store_true') parser.add_argument('--stdout', help='output std output to a file') parser.add_argument('--stderr', help='output std error to a file') - parser.add_argument('--input-file', '-i', action='append', - help='Declare input file for reproducible cmd') - parser.add_argument('--output-file', '-o', action='append', - help='Declare output file for reproducible cmd') + parser.add_argument('--input', '-i', action='append', + help='Declare input data items for reproducible cmd') + parser.add_argument('--output', '-o', action='append', + help='Declare output data items for reproducible cmd') parser.add_argument('--code', '-c', action='append', help='Code dependencies which produce the output') pass @@ -41,17 +42,13 @@ class CmdRun(CmdBase): def code_dependencies(self): return self.args.code - @property - def declaration_input_files(self): - if self.args.input_file: - return self.args.input_file - return [] + @cached_property + def declaration_input_data_items(self): + return self._data_items_from_params(self.args.input, 'Input') - @property - def declaration_output_files(self): - if self.args.output_file: - return self.args.output_file - return [] + @cached_property + def declaration_output_data_items(self): + return self._data_items_from_params(self.args.stdout, 'Output') def run(self): lock = fasteners.InterProcessLock(self.git.lock_file) @@ -64,54 +61,43 @@ class CmdRun(CmdBase): if not self.skip_git_actions and not self.git.is_ready_to_go(): return 1 - if not self.run_command(self._args_unkn, self.args.stdout, self.args.stderr): + if not self.run_command(self._args_unkn, + self._data_items_from_args(self._args_unkn), + self.args.stdout, + self.args.stderr): return 1 - message = 'DVC run: {}'.format(' '.join(sys.argv)) - self.commit_if_needed(message) + self.commit_if_needed('DVC run: {}'.format(' '.join(sys.argv))) finally: lock.release() return 0 - def run_command(self, argv, stdout=None, stderr=None): + def run_command(self, argv, data_items_from_args, stdout=None, stderr=None): repo_change = RepositoryChange(argv, stdout, stderr, self.git, self.config, self.path_factory) - # print('===== new_files={}'.format(repo_change.new_files)) - # print('===== modified_files={}'.format(repo_change.modified_files)) - # print('===== removed_files={}'.format(repo_change.removed_files)) - # print('===== externally_created_files={}'.format(repo_change.externally_created_files)) - # # raise Exception() - if not self.skip_git_actions and not self.validate_file_states(repo_change): self.remove_new_files(repo_change) return False - changed_files_nlx = self.git.abs_paths_to_dvc(repo_change.changed_files) - output_files = changed_files_nlx + self.git.abs_paths_to_dvc(self.declaration_output_files) - args_files_nlx = self.git.abs_paths_to_dvc(self.get_data_files_from_args(argv)) + output_set = set(self.declaration_output_data_items + repo_change.changed_data_items) + output_files_dvc = map(lambda x: x.data.dvc, output_set) - input_files_from_args = list(set(args_files_nlx) - set(changed_files_nlx)) - input_files = self.git.abs_paths_to_dvc(input_files_from_args + self.declaration_input_files) + input_set = set(data_items_from_args + self.declaration_input_data_items) + input_files_dvc = map(lambda x: x.data.dvc, input_set) - for data_item in repo_change.data_items_for_changed_files: - dirname = os.path.dirname(data_item.cache.relative) - if not os.path.isdir(dirname): - os.makedirs(dirname) + code_dependencies_dvc = self.git.abs_paths_to_dvc(self.code_dependencies) + for data_item in repo_change.changed_data_items: Logger.debug('Move output file "{}" to cache dir "{}" and create a symlink'.format( data_item.data.relative, data_item.cache.relative)) - shutil.move(data_item.data.relative, data_item.cache.relative) - - data_item.create_symlink() - - dvc_code_dependencies = map(lambda x: self.git.abs_paths_to_dvc([x])[0], self.code_dependencies) + data_item.move_data_to_cache() Logger.debug('Create state file "{}"'.format(data_item.state.relative)) state_file = StateFile(data_item.state.relative, self.git, - input_files, - output_files, - dvc_code_dependencies, + input_files_dvc, + output_files_dvc, + code_dependencies_dvc, argv=argv) state_file.save() pass @@ -120,28 +106,18 @@ class CmdRun(CmdBase): @staticmethod def remove_new_files(repo_change): - for file in repo_change.new_files: - rel_path = GitWrapper.abs_paths_to_relative([file])[0] - Logger.error('Removing created file: {}'.format(rel_path)) - os.remove(file) + for data_item in repo_change.new_data_items: + Logger.error('Removing created file: {}'.format(data_item.data.relative)) + os.remove(data_item.data.relative) pass - def validate_file_states(self, files_states): + def validate_file_states(self, repo_change): error = False - for file in GitWrapper.abs_paths_to_relative(files_states.removed_files): - Logger.error('Error: file "{}" was removed'.format(file)) + for data_item in repo_change.removed_data_items: + Logger.error('Error: file "{}" was removed'.format(data_item.data.relative)) error = True - # for file in GitWrapper.abs_paths_to_relative(files_states.modified_files): - # Logger.error('Error: file "{}" was modified'.format(file)) - # error = True - - # # This code doesn't cover repro case - # for file in GitWrapper.abs_paths_to_relative(files_states.unusual_state_files): - # Logger.error('Error: file "{}" is in not acceptable state'.format(file)) - # error = True - - for file in GitWrapper.abs_paths_to_relative(files_states.externally_created_files): + for file in GitWrapper.abs_paths_to_relative(repo_change.externally_created_files): Logger.error('Error: file "{}" was created outside of the data directory'.format(file)) error = True @@ -151,25 +127,32 @@ class CmdRun(CmdBase): format(self.config.data_dir)) return False - # if not files_states.new_files: - # Logger.error('Errors occurred. No files were changed in run cmd.') - # return False - return True - def get_data_files_from_args(self, argv): + def _data_items_from_args(self, argv): result = [] for arg in argv: try: if os.path.isfile(arg): - self.path_factory.data_item(arg) - result.append(arg) + data_item = self.path_factory.data_item(arg) + result.append(data_item) except NotInDataDirError: pass return result + def _data_items_from_params(self, files, param_text): + if not files: + return [] + + data_items, external = self.path_factory.to_data_items(files) + if external: + raise RunError('{} should point to data items from data dir: {}'.format( + param_text, ', '.join(external)) + ) + return data_items + if __name__ == '__main__': run(CmdRun()) diff --git a/dvc/path/data_item.py b/dvc/path/data_item.py index cf369c3e4..d43482eb7 100644 --- a/dvc/path/data_item.py +++ b/dvc/path/data_item.py @@ -1,5 +1,7 @@ import os +import shutil + from dvc.path.path import Path from dvc.exceptions import DvcException from dvc.utils import cached_property @@ -34,6 +36,12 @@ class DataItem(object): raise NotInDataDirError(data_file, self._config.data_dir) pass + def __hash__(self): + return self.data.dvc.__hash__() + + def __eq__(self, other): + return self.data.dvc == other.data.dvc + @property def data(self): return self._data @@ -69,3 +77,11 @@ class DataItem(object): def create_symlink(self): os.symlink(self._symlink_file, self.data.relative) + + def move_data_to_cache(self): + cache_dir = os.path.dirname(self.cache.relative) + if not os.path.isdir(cache_dir): + os.makedirs(cache_dir) + + shutil.move(self.data.relative, self.cache.relative) + os.symlink(self._symlink_file, self.data.relative) diff --git a/dvc/path/factory.py b/dvc/path/factory.py index d39e4b99e..a9615b7d9 100644 --- a/dvc/path/factory.py +++ b/dvc/path/factory.py @@ -2,6 +2,7 @@ import os from dvc.path.data_item import DataItem, DataItemError, NotInDataDirError from dvc.path.path import Path +from dvc.path.stated_data_item import StatedDataItem class PathFactory(object): @@ -16,6 +17,9 @@ class PathFactory(object): def data_item(self, data_file): return DataItem(data_file, self._git, self._config) + def stated_data_item(self, state, data_file): + return StatedDataItem(state, data_file, self._git, self._config) + def existing_data_item(self, file): if not os.path.islink(file): raise DataItemError(u'Data file "%s" must be a symbolic link' % file) diff --git a/dvc/path/stated_data_item.py b/dvc/path/stated_data_item.py new file mode 100644 index 000000000..01fef1532 --- /dev/null +++ b/dvc/path/stated_data_item.py @@ -0,0 +1,39 @@ +from dvc.path.data_item import DataItem + + +class StatedDataItem(DataItem): + STATUS_UNTRACKED = '?' + STATUS_DELETE = 'D' + STATUS_MODIFIED = 'M' + STATUS_TYPE_CHANGED = 'T' + + def __init__(self, state, data_file, git, config, cache_file=None): + super(StatedDataItem, self).__init__(data_file, git, config, cache_file) + self._status = state + + @property + def status(self): + return self._status + + def _check_status(self, status): + return self._status.find(status) >= 0 + + @property + def is_removed(self): + return self._check_status(self.STATUS_DELETE) + + @property + def is_modified(self): + return self._check_status(self.STATUS_MODIFIED) \ + or self._check_status(self.STATUS_TYPE_CHANGED) + + @property + def is_new(self): + return self._check_status(self.STATUS_UNTRACKED) + + @property + def is_unusual(self): + return self.is_new or self.is_modified or self.is_removed + + def __repr__(self): + return u'({}, {})'.format(self.state, self.data_dvc_short) diff --git a/dvc/repository_change.py b/dvc/repository_change.py index 6799bc688..d688ee25e 100644 --- a/dvc/repository_change.py +++ b/dvc/repository_change.py @@ -3,6 +3,9 @@ import os from dvc.exceptions import DvcException from dvc.git_wrapper import GitWrapper from dvc.executor import Executor +from dvc.path.data_item import NotInDataDirError +from dvc.path.stated_data_item import StatedDataItem +from dvc.utils import cached_property class RepositoryChangeError(DvcException): @@ -10,40 +13,6 @@ class RepositoryChangeError(DvcException): DvcException.__init__(self, 'Repository change error: {}'.format(msg)) -class FileModificationState(object): - STATUS_UNTRACKED = '?' - STATUS_DELETE = 'D' - STATUS_MODIFIED = 'M' - STATUS_TYPE_CHANGED = 'T' - - def __init__(self, state, file): - self.state = state - self.file = file - - def _check_status(self, status): - return self.state.find(status) >= 0 - - @property - def is_removed(self): - return self._check_status(self.STATUS_DELETE) - - @property - def is_modified(self): - return self._check_status(self.STATUS_MODIFIED) \ - or self._check_status(self.STATUS_TYPE_CHANGED) - - @property - def is_new(self): - return self._check_status(self.STATUS_UNTRACKED) - - @property - def is_unusual(self): - return self.is_new or self.is_modified or self.is_removed - - def __repr__(self): - return u'({}, {})'.format(self.state, self.file) - - class RepositoryChange(object): """Pre-condition: git repository has no changes""" @@ -53,60 +22,60 @@ class RepositoryChange(object): self.path_factory = path_factory Executor.exec_cmd_only_success(args, stdout, stderr) - self._file_states = self._get_file_states() - self._changed_data_items, self._externally_created_files = path_factory.to_data_items(self.changed_files) + self._stated_data_items = [] + self._externally_created_files = [] + self._init_file_states() @staticmethod def exec_cmd(args, stdout, stderr, git, path_factory): return RepositoryChange(args, stdout, stderr, git) - def get_filenames(self, func): - target = filter(func, self._file_states) - return list(map(lambda x: x.file, target)) + @cached_property + def removed_data_items(self): + return [x for x in self._stated_data_items if x.is_removed] + + @cached_property + def modified_data_items(self): + return [x for x in self._stated_data_items if x.is_modified] + + @cached_property + def new_data_items(self): + return [x for x in self._stated_data_items if x.is_new] + + @cached_property + def unusual_data_items(self): + return [x for x in self._stated_data_items if x.is_unusual] @property - def removed_files(self): - return self.get_filenames(lambda x: x.is_removed) + def changed_data_items(self): + return self.new_data_items + self.modified_data_items - @property - def modified_files(self): - return self.get_filenames(lambda x: x.is_modified) + def _add_stated_data_item(self, state, file): + try: + item = self.path_factory.stated_data_item(state, file) + self._stated_data_items.append(item) + except NotInDataDirError: + self._externally_created_files.append(file) + pass - @property - def new_files(self): - return self.get_filenames(lambda x: x.is_new) - - @property - def unusual_state_files(self): - return self.get_filenames(lambda x: not x.is_unusual) - - @property - def changed_files(self): - return self.new_files + self.modified_files - - def _get_file_states(self): + def _init_file_states(self): statuses = GitWrapper.git_file_statuses() - result = [] for status, file in statuses: file_path = os.path.join(self.git.git_dir_abs, file) if not os.path.isdir(file_path): - result.append(FileModificationState(status, file_path)) + self._add_stated_data_item(status, file_path) else: files = [] self.get_all_files_from_dir(file_path, files) - state = FileModificationState.STATUS_UNTRACKED + FileModificationState.STATUS_UNTRACKED + state = StatedDataItem.STATUS_UNTRACKED + StatedDataItem.STATUS_UNTRACKED for f in files: - result.append(FileModificationState(state, f)) - - return result + self._add_stated_data_item(state, f) + pass def get_all_files_from_dir(self, dir, result): - if not os.path.isdir(dir): - raise RepositoryChangeError('Changed path {} is not directory'.format(dir)) - files = os.listdir(dir) for f in files: path = os.path.join(dir, f) @@ -116,10 +85,6 @@ class RepositoryChange(object): self.get_all_files_from_dir(path, result) pass - @property - def data_items_for_changed_files(self): - return self._changed_data_items - @property def externally_created_files(self): return self._externally_created_files diff --git a/tests/test_repository_change.py b/tests/test_repository_change.py index 2bcafb2b1..96f11f985 100644 --- a/tests/test_repository_change.py +++ b/tests/test_repository_change.py @@ -1,6 +1,7 @@ import os import shutil import subprocess +import tempfile from unittest import TestCase from dvc.config import ConfigI @@ -9,32 +10,17 @@ from dvc.git_wrapper import GitWrapperI from dvc.repository_change import RepositoryChange -class TestRepositoryChange(TestCase): +class BasicTestRepositoryChange(TestCase): def setUp(self): - self.test_dir = os.path.realpath(os.path.join('/', 'tmp', 'dvc_unit_test_repo_change')) + self.test_dir = tempfile.mkdtemp() self.tearDown() os.mkdir(self.test_dir) os.chdir(self.test_dir) os.mkdir('data') - devnull = open(os.devnull, 'w') - subprocess.Popen(['git', 'init'], stdout=devnull, stderr=None).wait() - - self.file_to_modify = 'file1.txt' - self.file_created_before_run = 'file2.txt' - self.file_created_by_run = os.path.join('data', 'out.txt') - self.file_to_remove = 'file3.txt' - - self.create_file(self.file_to_modify) - self.create_file(self.file_created_before_run) - self.create_file(self.file_to_remove) - - subprocess.Popen(['git', 'add', self.file_to_modify, self.file_to_remove], - stdout=devnull, stderr=None).wait() - subprocess.Popen(['git', 'commit', '-m', '"Adding one file3"'], - stdout=devnull, stderr=None).wait() - os.remove(self.file_to_remove) + self._devnull = open(os.devnull, 'w') + subprocess.Popen(['git', 'init'], stdout=self._devnull, stderr=None).wait() self.git = GitWrapperI(self.test_dir) self.config = ConfigI('data', 'cache', 'state') @@ -42,7 +28,6 @@ class TestRepositoryChange(TestCase): def tearDown(self): shutil.rmtree(self.test_dir, ignore_errors=True) - pass @staticmethod def create_file(file2): @@ -50,26 +35,75 @@ class TestRepositoryChange(TestCase): fd.write('random text') fd.close() + +class TestRepositoryChange(BasicTestRepositoryChange): def test_all(self): + file_to_modify = os.path.join('data', 'file1.txt') + file_created_before_run = os.path.join('data', 'file2.txt') + file_created_by_run = os.path.join('data', 'out.txt') + file_to_remove = os.path.join('data', 'file3.txt') + + self.create_file(file_to_modify) + self.create_file(file_created_before_run) + self.create_file(file_to_remove) + + subprocess.Popen(['git', 'add', file_to_modify, file_to_remove], + stdout=self._devnull, stderr=None).wait() + subprocess.Popen(['git', 'commit', '-m', '"Adding one file3"'], + stdout=self._devnull, stderr=None).wait() + os.remove(file_to_remove) + change = RepositoryChange(['ls', '-la'], - self.file_created_by_run, - self.file_to_modify, + file_created_by_run, + file_to_modify, self.git, self.config, self.path_factory) - expected_new = [self.file_created_by_run, self.file_created_before_run] - self.assertEqual(set(change.new_files), set(map(os.path.realpath, expected_new))) + expected_new = [file_created_by_run, file_created_before_run] + new_file_abs = [x.data.dvc for x in change.new_data_items] + self.assertEqual(set(new_file_abs), set(expected_new)) - self.assertEqual(change.removed_files, [os.path.realpath(self.file_to_remove)]) - self.assertEqual(change.modified_files, [os.path.realpath(self.file_to_modify)]) + self.assertEqual(len(change.removed_data_items), 1) + self.assertEqual(change.removed_data_items[0].data.dvc, file_to_remove) + self.assertEqual(len(change.modified_data_items), 1) + self.assertEqual(change.modified_data_items[0].data.dvc, file_to_modify) - expected_to_change_set = set(map(os.path.realpath, expected_new + [self.file_to_modify])) - self.assertEqual(set(change.changed_files), expected_to_change_set) + expected_to_change_set = set(expected_new + [file_to_modify]) + changed_file_names = [x.data.dvc for x in change.changed_data_items] + self.assertEqual(set(changed_file_names), expected_to_change_set) - expected_to_change_set.remove(os.path.realpath(self.file_created_by_run)) - self.assertEqual(expected_to_change_set, set(change.externally_created_files)) - - self.assertEqual(1, len(change.data_items_for_changed_files)) - self.assertEqual(self.file_created_by_run, change.data_items_for_changed_files[0].data.dvc) + self.assertEqual([], change.externally_created_files) pass + + +class TestRepositoryChangeDeepInDirectory(BasicTestRepositoryChange): + def test(self): + os.mkdir(os.path.join('data', 'dir32')) + deep_file = os.path.join('data', 'dir32', 'file1.txt') + self.create_file(deep_file) + + change = RepositoryChange(['ls', '-la'], + deep_file, + None, + self.git, + self.config, + self.path_factory) + + self.assertEqual([x.data.dvc for x in change.new_data_items], [deep_file]) + + +class TestRepositoryChangeExternallyCreated(BasicTestRepositoryChange): + def test(self): + not_in_data_dir_file = 'file1.txt' + self.create_file(not_in_data_dir_file) + + change = RepositoryChange(['ls', '-la'], + not_in_data_dir_file, + None, + self.git, + self.config, + self.path_factory) + + self.assertEqual([os.path.realpath(not_in_data_dir_file)], + change.externally_created_files)