diff --git a/dvc/cli.py b/dvc/cli.py index 4f3ceb90b..7522557f0 100644 --- a/dvc/cli.py +++ b/dvc/cli.py @@ -268,9 +268,28 @@ def parse_args(argv=None): '--yes', action='store_true', default=False, - help="Automatic 'yes' answer to all prompts. E.g. " + help="(OBSOLETED, use --overwrite-dvcfile instead) " + "Automatic 'yes' answer to all prompts. E.g. " "when '.dvc' file exists and dvc asks if you " "want to overwrite it.") + run_parser.add_argument( + '--overwrite-dvcfile', + action='store_true', + default=False, + help="Overwrite existing dvc file without asking " + "for confirmation.") + run_parser.add_argument( + '--ignore-build-cache', + action='store_true', + default=False, + help="Run this stage even if it has been already " + "ran with the same command/dependencies/outputs/etc " + "before.") + run_parser.add_argument( + '--remove-outs', + action='store_true', + default=False, + help="Remove outputs before running the command.") run_parser.add_argument( 'command', nargs=argparse.REMAINDER, diff --git a/dvc/command/run.py b/dvc/command/run.py index 0fa2c2036..79857919d 100644 --- a/dvc/command/run.py +++ b/dvc/command/run.py @@ -20,10 +20,8 @@ class CmdRun(CmdBase): return cmd def run(self): + overwrite = (self.args.yes or self.args.overwrite_dvcfile) try: - if self.args.yes: - self.project.prompt.default = True - self.project.run(cmd=self._joined_cmd(), outs=self.args.outs, outs_no_cache=self.args.outs_no_cache, @@ -31,7 +29,10 @@ class CmdRun(CmdBase): deps=self.args.deps, fname=self.args.file, cwd=self.args.cwd, - no_exec=self.args.no_exec) + no_exec=self.args.no_exec, + overwrite=overwrite, + ignore_build_cache=self.args.ignore_build_cache, + remove_outs=self.args.remove_outs) except DvcException as ex: self.project.logger.error('Failed to run command', ex) return 1 diff --git a/dvc/project.py b/dvc/project.py index 7703dea5c..0ce183565 100644 --- a/dvc/project.py +++ b/dvc/project.py @@ -325,16 +325,21 @@ class Project(object): def _unprotect_file(self, path): import stat import uuid + from dvc.system import System from dvc.utils import copyfile, move, remove - self.logger.debug("Unprotecting '{}'".format(path)) + if System.is_symlink(path) or System.is_hardlink(path): + self.logger.debug("Unprotecting '{}'".format(path)) - tmp = os.path.join(os.path.dirname(path), '.' + str(uuid.uuid4())) - move(path, tmp) + tmp = os.path.join(os.path.dirname(path), '.' + str(uuid.uuid4())) + move(path, tmp) - copyfile(tmp, path) + copyfile(tmp, path) - remove(tmp) + remove(tmp) + else: + self.logger.debug("Skipping copying for '{}', since it is not " + "a symlink or a hardlink.".format(path)) os.chmod(path, os.stat(path).st_mode | stat.S_IWRITE) @@ -363,18 +368,26 @@ class Project(object): fname=None, cwd=os.curdir, no_exec=False, - overwrite=False): + overwrite=False, + ignore_build_cache=False, + remove_outs=False): from dvc.stage import Stage - stage = Stage.loads(project=self, - fname=fname, - cmd=cmd, - cwd=cwd, - outs=outs, - outs_no_cache=outs_no_cache, - metrics_no_cache=metrics_no_cache, - deps=deps, - overwrite=overwrite) + with self.state: + stage = Stage.loads(project=self, + fname=fname, + cmd=cmd, + cwd=cwd, + outs=outs, + outs_no_cache=outs_no_cache, + metrics_no_cache=metrics_no_cache, + deps=deps, + overwrite=overwrite, + ignore_build_cache=ignore_build_cache, + remove_outs=remove_outs) + + if stage is None: + return None all_stages = self.stages() diff --git a/dvc/stage.py b/dvc/stage.py index 30cd5b085..952197872 100644 --- a/dvc/stage.py +++ b/dvc/stage.py @@ -29,6 +29,12 @@ class StageFileDoesNotExistError(DvcException): super(StageFileDoesNotExistError, self).__init__(msg) +class StageFileAlreadyExistsError(DvcException): + def __init__(self, relpath): + msg = "Stage '{}' already exists".format(relpath) + super(StageFileAlreadyExistsError, self).__init__(msg) + + class StageFileIsNotDvcFileError(DvcException): def __init__(self, fname): msg = "'{}' is not a dvc file".format(fname) @@ -149,7 +155,10 @@ class Stage(object): len(self.deps) == 1 and \ len(self.outs) == 1 - def _changed_deps(self, print_info, log): + def _changed_deps(self, log): + if self.locked: + return False + if self.is_callback: msg = "Dvc file '{}' is a 'callback' stage (has a command and " \ "no dependencies) and thus always considered as changed." @@ -159,37 +168,35 @@ class Stage(object): for dep in self.deps: if not dep.changed(): continue - if print_info: - msg = "Dependency '{}' of '{}' changed." - log(msg.format(dep, self.relpath)) + log("Dependency '{}' of '{}' changed.".format(dep, self.relpath)) return True return False - def changed(self, print_info=False): - ret = False + def _changed_outs(self, log): + for out in self.outs: + if not out.changed(): + continue + log("Output '{}' of '{}' changed.".format(out, self.relpath)) + return True + return False + + def _changed_md5(self, log): + if self.changed_md5(): + log("Dvc file '{}' changed.".format(self.relpath)) + return True + return False + + def changed(self, print_info=False): if print_info: log = self.project.logger.info else: log = self.project.logger.debug - if not self.locked: - ret = self._changed_deps(print_info, log) - - for out in self.outs: - if not out.changed(): - continue - if print_info: - msg = "Output '{}' of '{}' changed." - log(msg.format(out, self.relpath)) - ret = True - - if self.changed_md5(): - if print_info: - msg = "Dvc file '{}' changed." - log(msg.format(self.relpath)) - ret = True + ret = any([self._changed_deps(log), + self._changed_outs(log), + self._changed_md5(log)]) if ret: msg = "Stage '{}' changed.".format(self.relpath) @@ -206,6 +213,12 @@ class Stage(object): for out in self.outs: out.remove(ignore_remove=ignore_remove) + def unprotect_outs(self): + for out in self.outs: + if out.path_info['scheme'] != 'local' or not out.exists: + continue + self.project.unprotect(out.path) + def remove(self): self.remove_outs(ignore_remove=True) os.unlink(self.path) @@ -290,6 +303,37 @@ class Stage(object): if not os.path.realpath(cwd).startswith(proj_dir): raise StageBadCwdError(cwd) + def is_cached(self): + """ + Checks if this stage has been already ran and saved to the same + dvc file. + """ + from dvc.remote.local import RemoteLOCAL + from dvc.remote.s3 import RemoteS3 + + old = Stage.load(self.project, self.path) + if old._changed_outs(log=self.project.logger.debug): + return False + + # NOTE: need to save checksums for deps in order to compare them + # with what is written in the old stage. + for dep in self.deps: + dep.save() + + old_d = old.dumpd() + new_d = self.dumpd() + + # NOTE: need to remove checksums from old dict in order to compare + # it to the new one, since the new one doesn't have checksums yet. + old_d.pop(self.PARAM_MD5, None) + new_d.pop(self.PARAM_MD5, None) + outs = old_d.get(self.PARAM_OUTS, []) + for out in outs: + out.pop(RemoteLOCAL.PARAM_MD5, None) + out.pop(RemoteS3.PARAM_ETAG, None) + + return old_d == new_d + @staticmethod def loads(project=None, cmd=None, @@ -301,7 +345,10 @@ class Stage(object): cwd=os.curdir, locked=False, add=False, - overwrite=True): + overwrite=True, + ignore_build_cache=False, + remove_outs=False): + stage = Stage(project=project, cwd=cwd, cmd=cmd, @@ -325,17 +372,28 @@ class Stage(object): cwd = os.path.abspath(cwd) path = os.path.join(cwd, fname) - if os.path.exists(path): - relpath = os.path.relpath(path) - msg = "'{}' already exists. " \ - "Do you wish to run the command and overwrite it?" - if not overwrite \ - and not project.prompt.prompt(msg.format(relpath), False): - raise DvcException("'{}' already exists".format(relpath)) - stage.cwd = cwd stage.path = path + # NOTE: remove outs before we check build cache + if remove_outs: + stage.remove_outs(ignore_remove=False) + project.logger.warn("Build cache is ignored when using " + "--remove-outs.") + ignore_build_cache = True + else: + stage.unprotect_outs() + + if os.path.exists(path): + if not ignore_build_cache and stage.is_cached(): + Logger.info('Stage is cached, skipping.') + return None + + msg = "'{}' already exists. Do you wish to run the command and " \ + "overwrite it?".format(stage.relpath) + if not overwrite and not project.prompt.prompt(msg, False): + raise StageFileAlreadyExistsError(stage.relpath) + return stage @staticmethod diff --git a/dvc/system.py b/dvc/system.py index bba11fa1e..767b12607 100644 --- a/dvc/system.py +++ b/dvc/system.py @@ -127,6 +127,11 @@ class System(object): from ctypes import c_void_p, c_wchar_p, Structure, WinError, POINTER from ctypes.wintypes import DWORD, HANDLE, BOOL + # NOTE: use this flag to open symlink itself and not the target + # See https://docs.microsoft.com/en-us/windows/desktop/api/ + # fileapi/nf-fileapi-createfilew#symbolic-link-behavior + FILE_FLAG_OPEN_REPARSE_POINT = 0x00200000 + FILE_FLAG_BACKUP_SEMANTICS = 0x02000000 FILE_SHARE_READ = 0x00000001 OPEN_EXISTING = 3 @@ -147,7 +152,7 @@ class System(object): ("nFileIndexHigh", DWORD), ("nFileIndexLow", DWORD)] - flags = FILE_FLAG_BACKUP_SEMANTICS + flags = FILE_FLAG_BACKUP_SEMANTICS | FILE_FLAG_OPEN_REPARSE_POINT func = ctypes.windll.kernel32.CreateFileW func.argtypes = [c_wchar_p, @@ -240,3 +245,22 @@ class System(object): return System._wait_for_input_posix(timeout) else: return System._wait_for_input_windows(timeout) + + @staticmethod + def is_symlink(path): + if System.is_unix(): + return os.path.islink(path) + + # https://docs.microsoft.com/en-us/windows/desktop/fileio/ + # file-attribute-constants + FILE_ATTRIBUTE_REPARSE_POINT = 0x400 + info = System.getdirinfo(path) + return info.dwFileAttributes & FILE_ATTRIBUTE_REPARSE_POINT + + @staticmethod + def is_hardlink(path): + if System.is_unix(): + return os.stat(path).st_nlink > 1 + + info = System.getdirinfo(path) + return info.nNumberOfLinks > 1 diff --git a/tests/test_run.py b/tests/test_run.py index 2e5f0a932..445264920 100644 --- a/tests/test_run.py +++ b/tests/test_run.py @@ -1,12 +1,16 @@ import os import mock +import time +import shutil import filecmp import subprocess +from dvc.project import Project from dvc.main import main from dvc.utils import file_md5 +from dvc.system import System from dvc.stage import Stage, StageFileBadNameError, MissingDep -from dvc.stage import StageBadCwdError +from dvc.stage import StageBadCwdError, StageFileAlreadyExistsError from dvc.command.run import CmdRun from dvc.exceptions import (OutputDuplicationError, CircularDependencyError, @@ -180,3 +184,260 @@ class TestCmdRun(TestDvc): def test_keyboard_interrupt(self, mock_popen): ret = main(['run', 'mycmd']) self.assertEqual(ret, 252) + + +class TestRunRemoveOuts(TestDvc): + def test(self): + with open(self.CODE, 'w+') as fobj: + fobj.write("import sys\n") + fobj.write("import os\n") + fobj.write("if os.path.exists(sys.argv[1]):\n") + fobj.write(" sys.exit(1)\n") + fobj.write("open(sys.argv[1], 'w+').close()\n") + + ret = main(['run', + '--remove-outs', + '-d', self.CODE, + '-o', self.FOO, + 'python', self.CODE, self.FOO]) + self.assertEqual(ret, 0) + + +class TestRunUnprotectOutsCopy(TestDvc): + def test(self): + with open(self.CODE, 'w+') as fobj: + fobj.write("import sys\n") + fobj.write("with open(sys.argv[1], 'a+') as fobj:\n") + fobj.write(" fobj.write('foo')\n") + + ret = main(['config', 'cache.protected', 'true']) + self.assertEqual(ret, 0) + + ret = main(['config', 'cache.type', 'copy']) + self.assertEqual(ret, 0) + + ret = main(['run', + '-d', self.CODE, + '-o', self.FOO, + 'python', self.CODE, self.FOO]) + self.assertEqual(ret, 0) + self.assertFalse(os.access(self.FOO, os.W_OK)) + self.assertEqual(open(self.FOO, 'r').read(), 'foofoo') + + ret = main(['run', + '--overwrite-dvcfile', + '--ignore-build-cache', + '-d', self.CODE, + '-o', self.FOO, + 'python', self.CODE, self.FOO]) + self.assertEqual(ret, 0) + self.assertFalse(os.access(self.FOO, os.W_OK)) + self.assertEqual(open(self.FOO, 'r').read(), 'foofoofoo') + + +class TestRunUnprotectOutsSymlink(TestDvc): + def test(self): + with open(self.CODE, 'w+') as fobj: + fobj.write("import sys\n") + fobj.write("import os\n") + fobj.write("assert os.path.exists(sys.argv[1])\n") + fobj.write("with open(sys.argv[1], 'a+') as fobj:\n") + fobj.write(" fobj.write('foo')\n") + + ret = main(['config', 'cache.protected', 'true']) + self.assertEqual(ret, 0) + + ret = main(['config', 'cache.type', 'symlink']) + self.assertEqual(ret, 0) + + self.assertEqual(ret, 0) + ret = main(['run', + '-d', self.CODE, + '-o', self.FOO, + 'python', self.CODE, self.FOO]) + self.assertEqual(ret, 0) + self.assertFalse(os.access(self.FOO, os.W_OK)) + self.assertTrue(System.is_symlink(self.FOO)) + self.assertEqual(open(self.FOO, 'r').read(), 'foofoo') + + ret = main(['run', + '--overwrite-dvcfile', + '--ignore-build-cache', + '-d', self.CODE, + '-o', self.FOO, + 'python', self.CODE, self.FOO]) + self.assertEqual(ret, 0) + self.assertFalse(os.access(self.FOO, os.W_OK)) + self.assertTrue(System.is_symlink(self.FOO)) + self.assertEqual(open(self.FOO, 'r').read(), 'foofoofoo') + + +class TestRunUnprotectOutsHardlink(TestDvc): + def test(self): + with open(self.CODE, 'w+') as fobj: + fobj.write("import sys\n") + fobj.write("import os\n") + fobj.write("assert os.path.exists(sys.argv[1])\n") + fobj.write("with open(sys.argv[1], 'a+') as fobj:\n") + fobj.write(" fobj.write('foo')\n") + + ret = main(['config', 'cache.protected', 'true']) + self.assertEqual(ret, 0) + + ret = main(['config', 'cache.type', 'hardlink']) + self.assertEqual(ret, 0) + + self.assertEqual(ret, 0) + ret = main(['run', + '-d', self.CODE, + '-o', self.FOO, + 'python', self.CODE, self.FOO]) + self.assertEqual(ret, 0) + self.assertFalse(os.access(self.FOO, os.W_OK)) + self.assertTrue(System.is_hardlink(self.FOO)) + self.assertEqual(open(self.FOO, 'r').read(), 'foofoo') + + ret = main(['run', + '--overwrite-dvcfile', + '--ignore-build-cache', + '-d', self.CODE, + '-o', self.FOO, + 'python', self.CODE, self.FOO]) + self.assertEqual(ret, 0) + self.assertFalse(os.access(self.FOO, os.W_OK)) + self.assertTrue(System.is_hardlink(self.FOO)) + self.assertEqual(open(self.FOO, 'r').read(), 'foofoofoo') + + +class TestCmdRunOverwrite(TestDvc): + def test(self): + ret = main(['run', + '-d', self.FOO, + '-d', self.CODE, + '-o', 'out', + '-f', 'out.dvc', + 'python', self.CODE, self.FOO, 'out']) + self.assertEqual(ret, 0) + + stage_mtime = os.path.getmtime('out.dvc') + + ret = main(['run', + '-d', self.FOO, + '-d', self.CODE, + '-o', 'out', + '-f', 'out.dvc', + 'python', self.CODE, self.FOO, 'out']) + self.assertEqual(ret, 0) + + # NOTE: check that dvcfile was NOT overwritten + self.assertEqual(stage_mtime, os.path.getmtime('out.dvc')) + stage_mtime = os.path.getmtime('out.dvc') + + ret = main(['run', + '-d', self.FOO, + '-d', self.CODE, + '--overwrite-dvcfile', + '--ignore-build-cache', + '-o', 'out', + '-f', 'out.dvc', + 'python', self.CODE, self.FOO, 'out']) + self.assertEqual(ret, 0) + + # NOTE: check that dvcfile was overwritten + self.assertNotEqual(stage_mtime, os.path.getmtime('out.dvc')) + stage_mtime = os.path.getmtime('out.dvc') + + ret = main(['run', + '--overwrite-dvcfile', + '-f', 'out.dvc', + '-d', self.BAR]) + self.assertEqual(ret, 0) + + # NOTE: check that dvcfile was overwritten + self.assertNotEqual(stage_mtime, os.path.getmtime('out.dvc')) + + +class TestRunDeterministicBase(TestDvc): + def setUp(self): + super(TestRunDeterministicBase, self).setUp() + self.out_file = 'out' + self.stage_file = self.out_file + '.dvc' + self.cmd = 'python {} {} {}'.format(self.CODE, self.FOO, self.out_file) + self.deps = [self.FOO, self.CODE] + self.outs = [self.out_file] + self.overwrite = False + self.ignore_build_cache = False + + self._run() + + def _run(self): + self.stage = self.dvc.run(cmd=self.cmd, + fname=self.stage_file, + overwrite=self.overwrite, + ignore_build_cache=self.ignore_build_cache, + deps=self.deps, + outs=self.outs) + + +class TestRunDeterministic(TestRunDeterministicBase): + def test(self): + self._run() + + +class TestRunDeterministicOverwrite(TestRunDeterministicBase): + def test(self): + self.overwrite = True + self.ignore_build_cache = True + self._run() + + +class TestRunDeterministicCallback(TestRunDeterministicBase): + def test(self): + self.stage.remove() + self.deps = [] + self._run() + self._run() + + +class TestRunDeterministicChangedDep(TestRunDeterministicBase): + def test(self): + os.unlink(self.FOO) + shutil.copy(self.BAR, self.FOO) + with self.assertRaises(StageFileAlreadyExistsError): + self._run() + + +class TestRunDeterministicChangedDepsList(TestRunDeterministicBase): + def test(self): + self.deps = [self.BAR, self.CODE] + with self.assertRaises(StageFileAlreadyExistsError): + self._run() + + +class TestRunDeterministicNewDep(TestRunDeterministicBase): + def test(self): + self.deps = [self.FOO, self.BAR, self.CODE] + with self.assertRaises(StageFileAlreadyExistsError): + self._run() + + +class TestRunDeterministicRemoveDep(TestRunDeterministicBase): + def test(self): + self.deps = [self.CODE] + with self.assertRaises(StageFileAlreadyExistsError): + self._run() + + +class TestRunDeterministicChangedOut(TestRunDeterministicBase): + def test(self): + os.unlink(self.out_file) + self.out_file_mtime = None + with self.assertRaises(StageFileAlreadyExistsError): + self._run() + + +class TestRunDeterministicChangedCmd(TestRunDeterministicBase): + def test(self): + self.cmd += ' arg' + with self.assertRaises(StageFileAlreadyExistsError): + self._run()