Repository: iterative/dvc

Issue:
graph: check for circular dependencies when parsing DAG
E.g. same way we check output duplication in project.py:graph(). We should be able to just search for cycles in the networkx graph and print the trace.

Discussion:

Patch:
diff --git a/dvc/exceptions.py b/dvc/exceptions.py
index b26481fc6..565ce4023 100644
--- a/dvc/exceptions.py
+++ b/dvc/exceptions.py
@@ -117,3 +117,17 @@ class DvcParserError(DvcException):
     """Base class for CLI parser errors."""
     def __init__(self):
         super(DvcParserError, self).__init__("parser error")
+
+
+class CyclicGraphError(DvcException):
+    def __init__(self, stages):
+        assert isinstance(stages, list)
+        stages = '\n'.join('\t- {}'.format(stage) for stage in stages)
+        msg = (
+            "you've introduced a cycle in your pipeline that involves"
+            ' the following stages:'
+            '\n'
+            '{stages}'
+            .format(stages=stages)
+        )
+        super(CyclicGraphError, self).__init__(msg)
diff --git a/dvc/project.py b/dvc/project.py
index ed1affbe4..74403171a 100644
--- a/dvc/project.py
+++ b/dvc/project.py
@@ -176,25 +176,15 @@ class Project(object):
     def install(self):
         self.scm.install()
 
-    def _check_cwd_specified_as_output(self, cwd, stages):
-        from dvc.exceptions import WorkingDirectoryAsOutputError
+    @staticmethod
+    def _check_cyclic_graph(graph):
+        import networkx as nx
+        from dvc.exceptions import CyclicGraphError
 
-        cwd_path = os.path.abspath(os.path.normpath(cwd))
+        cycles = list(nx.simple_cycles(graph))
 
-        for stage in stages:
-            for output in stage.outs:
-                if os.path.isdir(output.path) and output.path == cwd_path:
-                    raise WorkingDirectoryAsOutputError(cwd, stage.relpath)
-
-    def _check_output_duplication(self, outs, stages):
-        from dvc.exceptions import OutputDuplicationError
-
-        for stage in stages:
-            for o in stage.outs:
-                for out in outs:
-                    if o.path == out.path and o.stage.path != out.stage.path:
-                        stages = [o.stage.relpath, out.stage.relpath]
-                        raise OutputDuplicationError(o.path, stages)
+        if cycles:
+            raise CyclicGraphError(cycles[0])
 
     def add(self, fname, recursive=False):
         from dvc.stage import Stage
@@ -215,7 +205,6 @@ class Project(object):
         else:
             fnames = [fname]
 
-        all_stages = self.stages()
         stages = []
         self._files_to_git_add = []
         with self.state:
@@ -228,16 +217,22 @@ class Project(object):
                     stages.append(stage)
                     continue
 
-                self._check_output_duplication(stage.outs, all_stages)
-
                 stage.save()
-                stage.dump()
                 stages.append(stage)
 
+        self._check_dag(self.stages() + stages)
+
+        for stage in stages:
+            stage.dump()
+
         self._remind_to_git_add()
 
         return stages
 
+    def _check_dag(self, stages):
+        """Generate graph including the new stage to check for errors"""
+        self.graph(stages=stages)
+
     def remove(self, target, outs_only=False):
         from dvc.stage import Stage
 
@@ -389,10 +384,7 @@ class Project(object):
         if stage is None:
             return None
 
-        all_stages = self.stages()
-
-        self._check_cwd_specified_as_output(cwd, all_stages)
-        self._check_output_duplication(stage.outs, all_stages)
+        self._check_dag(self.stages() + [stage])
 
         self._files_to_git_add = []
         with self.state:
@@ -416,7 +408,7 @@ class Project(object):
         if stage is None:
             return None
 
-        self._check_output_duplication(stage.outs, self.stages())
+        self._check_dag(self.stages() + [stage])
 
         self._files_to_git_add = []
         with self.state:
@@ -1174,24 +1166,35 @@ class Project(object):
     def metrics_remove(self, path):
         self._metrics_modify(path, delete=True)
 
-    def graph(self, from_directory=None):
+    def graph(self, stages=None, from_directory=None):
         import networkx as nx
-        from dvc.exceptions import OutputDuplicationError
+        from dvc.exceptions import (OutputDuplicationError,
+                                    WorkingDirectoryAsOutputError)
 
         G = nx.DiGraph()
         G_active = nx.DiGraph()
-        stages = self.stages(from_directory)
-
+        stages = stages or self.stages(from_directory)
+        stages = [stage for stage in stages if stage]
         outs = []
-        outs_by_path = {}
+
+        for stage in stages:
+            for out in stage.outs:
+                existing = [o.stage for o in outs if o.path == out.path]
+
+                if existing:
+                    stages = [stage.relpath, existing[0].relpath]
+                    raise OutputDuplicationError(out.path, stages)
+
+                outs.append(out)
+
         for stage in stages:
-            for o in stage.outs:
-                existing = outs_by_path.get(o.path, None)
-                if existing is not None:
-                    stages = [o.stage.relpath, existing.stage.relpath]
-                    raise OutputDuplicationError(o.path, stages)
-                outs.append(o)
-                outs_by_path[o.path] = o
+            for out in outs:
+                overlaps = (stage.cwd == out.path
+                            or stage.cwd.startswith(out.path + os.sep))
+
+                if overlaps:
+                    raise WorkingDirectoryAsOutputError(stage.cwd,
+                                                        stage.relpath)
 
         # collect the whole DAG
         for stage in stages:
@@ -1215,12 +1218,14 @@ class Project(object):
                         G_active.add_node(dep_node, stage=dep_stage)
                         G_active.add_edge(node, dep_node)
 
+        self._check_cyclic_graph(G)
+
         return G, G_active
 
     def pipelines(self, from_directory=None):
         import networkx as nx
 
-        G, G_active = self.graph(from_directory)
+        G, G_active = self.graph(from_directory=from_directory)
 
         return [
             G.subgraph(c).copy()
diff --git a/dvc/stage.py b/dvc/stage.py
index 7bb718f7e..3512a1a9c 100644
--- a/dvc/stage.py
+++ b/dvc/stage.py
@@ -391,6 +391,8 @@ class Stage(object):
             if not overwrite and not prompt.confirm(msg):
                 raise StageFileAlreadyExistsError(stage.relpath)
 
+            os.unlink(path)
+
         return stage
 
     @staticmethod


Test patch:
diff --git a/tests/test_repro.py b/tests/test_repro.py
index a943657e2..998b7eb96 100644
--- a/tests/test_repro.py
+++ b/tests/test_repro.py
@@ -17,6 +17,7 @@ from dvc.utils.compat import urljoin
 from dvc.remote.local import RemoteLOCAL
 from dvc.stage import Stage, StageFileDoesNotExistError
 from dvc.system import System
+from dvc.exceptions import CyclicGraphError, WorkingDirectoryAsOutputError
 
 from tests.basic_env import TestDvc
 from tests.test_data_cloud import _should_test_aws, TEST_AWS_REPO_BUCKET
@@ -53,6 +54,145 @@ class TestReproFail(TestRepro):
         self.assertNotEqual(ret, 0)
 
 
+class TestReproCyclicGraph(TestDvc):
+    def test(self):
+        self.dvc.run(deps=[self.FOO],
+                     outs=['bar.txt'],
+                     cmd='echo bar > bar.txt')
+
+        self.dvc.run(deps=['bar.txt'],
+                     outs=['baz.txt'],
+                     cmd='echo baz > baz.txt')
+
+        with open('cycle.dvc', 'w') as fd:
+            stage_dump = {
+                'cmd': 'echo baz > foo',
+                'deps': [{'path': 'baz.txt'}],
+                'outs': [{'path': self.FOO}],
+            }
+
+            yaml.safe_dump(stage_dump, fd, default_flow_style=False)
+
+        with self.assertRaises(CyclicGraphError):
+            self.dvc.reproduce('cycle.dvc')
+
+
+class TestReproWorkingDirectoryAsOutput(TestDvc):
+    """
+    |  stage.cwd  |  out.path | cwd as output |
+    |:-----------:|:---------:|:-------------:|
+    |     dir     |    dir    |      True     |
+    | dir/subdir/ |    dir    |      True     |
+    |     dir     |   dir-1   |     False     |
+    |      .      | something |     False     |
+    """
+    def test(self):
+        # File structure:
+        #       .
+        #       |-- dir1
+        #       |  |__ dir2.dvc         (out.path == ../dir2)
+        #       |__ dir2
+        #           |__ something.dvc    (stage.cwd == ./dir2)
+
+        os.mkdir(os.path.join(self.dvc.root_dir, 'dir1'))
+
+        self.dvc.run(
+            cwd='dir1',
+            outs=['../dir2'],
+            cmd='mkdir {path}'.format(path=os.path.join('..', 'dir2'))
+        )
+
+        faulty_stage_path = os.path.join('dir2', 'something.dvc')
+
+        with open(faulty_stage_path, 'w') as fd:
+            stage_dump = {
+                'cmd': 'echo something > something',
+                'outs': [{'path': 'something'}],
+            }
+
+            yaml.safe_dump(stage_dump, fd, default_flow_style=False)
+
+        with self.assertRaises(WorkingDirectoryAsOutputError):
+            self.dvc.reproduce(faulty_stage_path)
+
+    def test_nested(self):
+        from dvc.stage import Stage
+        #
+        #       .
+        #       |-- a
+        #       |  |__ nested
+        #       |     |__ dir
+        #       |       |__ error.dvc     (stage.cwd == 'a/nested/dir')
+        #       |__ b
+        #          |__ nested.dvc         (stage.out == 'a/nested')
+        dir1 = 'b'
+        dir2 = 'a'
+
+        os.mkdir(dir1)
+        os.mkdir(dir2)
+
+        nested_dir = os.path.join(dir2, 'nested')
+        out_dir = os.path.relpath(nested_dir, dir1)
+
+        nested_stage = self.dvc.run(
+            cwd=dir1,          # b
+            outs=[out_dir],    # ../a/nested
+            cmd='mkdir {path}'.format(path=out_dir)
+        )
+
+        os.mkdir(os.path.join(nested_dir, 'dir'))
+
+        error_stage_path = os.path.join(nested_dir, 'dir', 'error.dvc')
+
+        with open(error_stage_path, 'w') as fd:
+            stage_dump = {
+                'cmd': 'echo something > something',
+                'outs': [{'path': 'something'}],
+            }
+
+            yaml.safe_dump(stage_dump, fd, default_flow_style=False)
+
+        # NOTE: os.walk() walks in a sorted order and we need dir2 subdirs to
+        # be processed before dir1 to load error.dvc first.
+        with patch.object(Project, 'stages') as mock_stages:
+            mock_stages.return_value = [
+                nested_stage,
+                Stage.load(self.dvc, error_stage_path)
+            ]
+
+            with self.assertRaises(WorkingDirectoryAsOutputError):
+                self.dvc.reproduce(error_stage_path)
+
+    def test_similar_paths(self):
+        # File structure:
+        #
+        #       .
+        #       |-- something.dvc   (out.path == something)
+        #       |-- something
+        #       |__ something-1
+        #          |-- a
+        #          |__ a.dvc        (stage.cwd == something-1)
+
+        self.dvc.run(outs=['something'], cmd='mkdir something')
+
+        os.mkdir('something-1')
+
+        stage = os.path.join('something-1', 'a.dvc')
+
+        with open(stage, 'w') as fd:
+            stage_dump = {
+                'cmd': 'echo a > a',
+                'outs': [{'path': 'a'}],
+            }
+
+            yaml.safe_dump(stage_dump, fd, default_flow_style=False)
+
+        try:
+            self.dvc.reproduce(stage)
+        except WorkingDirectoryAsOutputError:
+            self.fail('should not raise WorkingDirectoryAsOutputError')
+
+
 class TestReproDepUnderDir(TestDvc):
     def test(self):
         stages = self.dvc.add(self.DATA_DIR)
diff --git a/tests/test_run.py b/tests/test_run.py
index 71f64ce45..3ccbb0f0f 100644
--- a/tests/test_run.py
+++ b/tests/test_run.py
@@ -12,6 +12,7 @@ from dvc.stage import StageFileBadNameError, MissingDep
 from dvc.stage import StageBadCwdError, StageFileAlreadyExistsError
 from dvc.exceptions import (OutputDuplicationError,
                             CircularDependencyError,
+                            CyclicGraphError,
                             ArgumentDuplicationError,
                             WorkingDirectoryAsOutputError)
 
@@ -122,6 +123,20 @@ class TestRunCircularDependency(TestDvc):
                          outs=['foo'],
                          fname='circular-dependency.dvc')
 
+    def test_graph(self):
+        self.dvc.run(deps=[self.FOO],
+                     outs=['bar.txt'],
+                     cmd='echo bar > bar.txt')
+
+        self.dvc.run(deps=['bar.txt'],
+                     outs=['baz.txt'],
+                     cmd='echo baz > baz.txt')
+
+        with self.assertRaises(CyclicGraphError):
+            self.dvc.run(deps=['baz.txt'],
+                         outs=[self.FOO],
+                         cmd='echo baz > foo')
+
 
 class TestRunDuplicatedArguments(TestDvc):
     def test(self):