From cb142f9f0e515b3db667caa1049d899646fca2d2 Mon Sep 17 00:00:00 2001 From: "Mr. Outis" Date: Thu, 13 Dec 2018 13:37:39 -0600 Subject: [PATCH 1/2] project: Refactor pipeline What's a pipeline if not the weekly connected component subgraphs of a directed acyclic graph? --- dvc/project.py | 20 +------------------- 1 file changed, 1 insertion(+), 19 deletions(-) diff --git a/dvc/project.py b/dvc/project.py index dce1ade24..720c10b58 100644 --- a/dvc/project.py +++ b/dvc/project.py @@ -1180,25 +1180,7 @@ class Project(object): G, G_active = self.graph() - if len(G.nodes()) == 0: - return [] - - # find pipeline ends aka "output stages" - ends = [node for node, in_degree in G.in_degree() if in_degree == 0] - - # filter out subgraphs that didn't exist in original G - pipelines = [] - for c in nx.weakly_connected_components(G_active): - H = G_active.subgraph(c) - found = False - for node in ends: - if node in H: - found = True - break - if found: - pipelines.append(H) - - return pipelines + return list(nx.weakly_connected_component_subgraphs(G_active)) def stages(self): """ From 4142f65531907caa872a14f465f3dfb90bde65d7 Mon Sep 17 00:00:00 2001 From: "Mr. Outis" Date: Sun, 16 Dec 2018 18:05:56 -0600 Subject: [PATCH 2/2] Add tests for pipelines --- tests/test_pipeline.py | 38 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 38 insertions(+) diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py index affdeb373..7350f7fce 100644 --- a/tests/test_pipeline.py +++ b/tests/test_pipeline.py @@ -187,3 +187,41 @@ class TestPipelineListSingle(TestPipelineShowDeep): def test(self): ret = main(['pipeline', 'list']) self.assertEqual(ret, 0) + + +class TestProjectPipeline(TestDvc): + def test_no_stages(self): + pipelines = self.dvc.pipelines() + self.assertEqual(len(pipelines), 0) + + def one_pipeline(self): + self.dvc.add('foo') + self.dvc.run(deps=['foo'], outs=['bar'], cmd='') + self.dvc.run(deps=['bar'], outs=['baz'], cmd='echo baz > baz') + pipelines = self.dvc.pipelines() + + self.assertEqual(len(pipelines), 1) + self.assertEqual(pipelines[0].nodes, 3) + self.assertEqual(pipelines[0].edges, 2) + + def two_pipelines(self): + self.dvc.add('foo') + self.dvc.run(deps=['foo'], outs=['bar'], cmd='') + self.dvc.run(deps=['bar'], outs=['baz'], cmd='echo baz > baz') + + self.dvc.add('code.py') + + pipelines = self.dvc.pipelines() + + self.assertEqual(len(pipelines), 2) + self.assertEqual(pipelines[0].nodes, 3) + self.assertEqual(pipelines[0].edges, 2) + self.assertEqual(pipelines[1].nodes, 1) + self.assertEqual(pipelines[1].edges, 0) + + def locked_stage(self): + self.dvc.add('foo') + self.dvc.lock_stage('foo.dvc') + + pipelines = self.dvc.pipelines() + self.assertEqual(len(pipelines), 0)