Run cmd refactoring

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

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())

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)

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)

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)

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

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)