from dvc.main import main from tests.basic_env import TestDvc from tests.test_repro import TestRepro, TestReproChangedDeepData import os class TestPipelineShowSingle(TestDvc): def setUp(self): super(TestPipelineShowSingle, self).setUp() self.stage = "foo.dvc" self.dotFile = "graph.dot" ret = main(["add", self.FOO]) self.assertEqual(ret, 0) def test(self): ret = main(["pipeline", "show", self.stage]) self.assertEqual(ret, 0) def test_commands(self): ret = main(["pipeline", "show", self.stage, "--commands"]) self.assertEqual(ret, 0) def test_outs(self): ret = main(["pipeline", "show", self.stage, "--outs"]) self.assertEqual(ret, 0) def test_ascii(self): ret = main(["pipeline", "show", "--ascii", self.stage]) self.assertEqual(ret, 0) def test_dot(self): ret = main(["pipeline", "show", "--dot", self.dotFile, self.stage]) self.assertEqual(ret, 0) self.assertTrue(os.path.isfile(self.dotFile)) def test_ascii_commands(self): ret = main(["pipeline", "show", "--ascii", self.stage, "--commands"]) self.assertEqual(ret, 0) def test_ascii_outs(self): ret = main(["pipeline", "show", "--ascii", self.stage, "--outs"]) self.assertEqual(ret, 0) def test_dot_commands(self): ret = main( ["pipeline", "show", "--dot", self.dotFile, self.stage, "--commands"] ) self.assertEqual(ret, 0) self.assertTrue(os.path.isfile(self.dotFile)) def test_dot_outs(self): ret = main(["pipeline", "show", "--dot", self.dotFile, self.stage, "--outs"]) self.assertEqual(ret, 0) self.assertTrue(os.path.isfile(self.dotFile)) def test_not_dvc_file(self): ret = main(["pipeline", "show", self.FOO]) self.assertNotEqual(ret, 0) def test_non_existing(self): ret = main(["pipeline", "show", "non-existing"]) self.assertNotEqual(ret, 0) class TestPipelineShow(TestRepro): def setUp(self): super(TestPipelineShow, self).setUp() self.dotFile = "graph.dot" def test(self): ret = main(["pipeline", "show", self.file1_stage]) self.assertEqual(ret, 0) def test_commands(self): ret = main(["pipeline", "show", self.file1_stage, "--commands"]) self.assertEqual(ret, 0) def test_outs(self): ret = main(["pipeline", "show", self.file1_stage, "--outs"]) self.assertEqual(ret, 0) def test_ascii(self): ret = main(["pipeline", "show", "--ascii", self.file1_stage]) self.assertEqual(ret, 0) def test_dot(self): ret = main(["pipeline", "show", "--dot", self.dotFile, self.file1_stage]) self.assertEqual(ret, 0) self.assertTrue(os.path.isfile(self.dotFile)) def test_ascii_commands(self): ret = main(["pipeline", "show", "--ascii", self.file1_stage, "--commands"]) self.assertEqual(ret, 0) def test_ascii_outs(self): ret = main(["pipeline", "show", "--ascii", self.file1_stage, "--outs"]) self.assertEqual(ret, 0) def test_dot_commands(self): ret = main( ["pipeline", "show", "--dot", self.dotFile, self.file1_stage, "--commands"] ) self.assertEqual(ret, 0) self.assertTrue(os.path.isfile(self.dotFile)) def test_dot_outs(self): ret = main( ["pipeline", "show", "--dot", self.dotFile, self.file1_stage, "--outs"] ) self.assertEqual(ret, 0) self.assertTrue(os.path.isfile(self.dotFile)) def test_not_dvc_file(self): ret = main(["pipeline", "show", self.file1]) self.assertNotEqual(ret, 0) def test_non_existing(self): ret = main(["pipeline", "show", "non-existing"]) self.assertNotEqual(ret, 0) class TestPipelineShowDeep(TestReproChangedDeepData): def setUp(self): super(TestPipelineShowDeep, self).setUp() self.dotFile = "graph.dot" def test(self): ret = main(["pipeline", "show", self.file1_stage]) self.assertEqual(ret, 0) def test_commands(self): ret = main(["pipeline", "show", self.file1_stage, "--commands"]) self.assertEqual(ret, 0) def test_outs(self): ret = main(["pipeline", "show", self.file1_stage, "--outs"]) self.assertEqual(ret, 0) def test_ascii(self): ret = main(["pipeline", "show", "--ascii", self.file1_stage]) self.assertEqual(ret, 0) def test_dot(self): ret = main(["pipeline", "show", "--dot", self.dotFile, self.file1_stage]) self.assertEqual(ret, 0) self.assertTrue(os.path.isfile(self.dotFile)) def test_ascii_commands(self): ret = main(["pipeline", "show", "--ascii", self.file1_stage, "--commands"]) self.assertEqual(ret, 0) def test_ascii_outs(self): ret = main(["pipeline", "show", "--ascii", self.file1_stage, "--outs"]) self.assertEqual(ret, 0) def test_dot_commands(self): ret = main( ["pipeline", "show", "--dot", self.dotFile, self.file1_stage, "--commands"] ) self.assertEqual(ret, 0) self.assertTrue(os.path.isfile(self.dotFile)) def test_dot_outs(self): ret = main( ["pipeline", "show", "--dot", self.dotFile, self.file1_stage, "--outs"] ) self.assertEqual(ret, 0) self.assertTrue(os.path.isfile(self.dotFile)) def test_not_dvc_file(self): ret = main(["pipeline", "show", self.file1]) self.assertNotEqual(ret, 0) def test_non_existing(self): ret = main(["pipeline", "show", "non-existing"]) self.assertNotEqual(ret, 0) class TestPipelineListEmpty(TestDvc): def test(self): ret = main(["pipeline", "list"]) self.assertEqual(ret, 0) 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)