def testDependencies(self):
     my_task = Task("t2", [""], file_dep=['d2.txt'])
     output = StringIO()
     cmd_list = CmdFactory(List, outstream=output, task_list=[my_task])
     cmd_list._execute(list_deps=True)
     got = output.getvalue()
     assert "d2.txt" in got

 def test_no_children(self):
     my_task = Task("t2", [""], file_dep=['d2.txt'])
     output = StringIO()
     cmd = CmdFactory(Graphx, outstream=output, task_list=[my_task])
     cmd._execute(graph_type='json', no_children=True)
     got = output.getvalue()
     self.assertNotIn("d2.txt", got)

 def testCustomReporter(self, depfile_name):
     output = StringIO()
     cmd_run = CmdFactory(Run, backend='dbm', dep_file=depfile_name,
                          task_list=[tasks_sample()[0]])
     cmd_run._execute(output, reporter=MyReporter)
     got = output.getvalue().split("\n")[:-1]
     assert 'MyReporter.start t1' == got[0]

 def testFilterAll(self):
     output = StringIO()
     cmd_list = CmdFactory(List, outstream=output, task_list=tasks_sample())
     cmd_list._execute(subtasks=True, pos_args=['g1'])
     got = [line.strip() for line in output.getvalue().split('\n') if line]
     expected = ['g1', 'g1.a', 'g1.b']
     assert expected == got

 def testProcessRunEmptyFilter(self, depfile_name):
     output = StringIO()
     cmd_run = CmdFactory(Run, backend='dbm', dep_file=depfile_name,
                          task_list=tasks_sample(), sel_tasks=[])
     cmd_run._execute(output)
     got = output.getvalue().split("\n")[:-1]
     assert [] == got

 def testProcessRunSingle(self, depfile_name):
     output = StringIO()
     cmd_run = CmdFactory(Run, backend="dbm", dep_file=depfile_name, task_list=tasks_sample(), sel_tasks=["t3"])
     cmd_run._execute(output, single=True)
     got = output.getvalue().split("\n")[:-1]
     # t1 is a depenendency of t3 but not included
     assert [".  t3"] == got

 def testProcessRunMThread(self, dependency1, depfile_name):
     output = StringIO()
     cmd_run = CmdFactory(Run, backend="dbm", dep_file=depfile_name, task_list=tasks_sample())
     result = cmd_run._execute(output, num_process=1, par_type="thread")
     assert 0 == result
     got = output.getvalue().split("\n")[:-1]
     assert [".  t1", ".  t2", ".  g1.a", ".  g1.b", ".  t3"] == got

 def test_filter(self, depfile, dependency1):
     output = StringIO()
     tasks = tasks_sample()
     cmd_reset = CmdFactory(ResetDep, outstream=output, task_list=tasks,
                            dep_manager=depfile)
     cmd_reset._execute(pos_args=['t2'])
     got = output.getvalue()
     assert "processed t2\n" == got

 def testCustomTemplate(self):
     output = StringIO()
     tasks = tasks_sample()
     cmd_list = CmdFactory(List, outstream=output, task_list=tasks)
     cmd_list._execute(template='xxx {name} xxx {doc}')
     got = [line.strip() for line in output.getvalue().split('\n') if line]
     assert 'xxx g1 xxx g1 doc string' == got[0]
     assert 'xxx t3 xxx t3 doc string' == got[3]

 def test_children(self):
     # TODO: Implement no-child option.
     my_task = Task("t2", [""], file_dep=['d2.txt'])
     output = StringIO()
     cmd = CmdFactory(Graphx, outstream=output, task_list=[my_task])
     cmd._execute(graph_type='json', no_children=False)
     got = output.getvalue()
     self.assertIn("d2.txt", got)

 def testQuiet(self):
     output = StringIO()
     tasks = tasks_sample()
     cmd_list = CmdFactory(List, outstream=output, task_list=tasks)
     cmd_list._execute()
     got = [line.strip() for line in output.getvalue().split('\n') if line]
     expected = [t.name for t in tasks if not t.is_subtask]
     assert sorted(expected) == got

 def test_unicode_name(self, depfile):
     task_list = [Task(six.u("t做"), [""], doc=six.u("t1 doc string 做")),]
     output = StringIO()
     cmd_list = CmdFactory(List, outstream=output, dep_file=depfile.name,
                           task_list=task_list)
     cmd_list._execute()
     got = [line.strip() for line in output.getvalue().split('\n') if line]
     assert six.u('t做') == got[0]

 def test_store_json_stdout(self):
     output = StringIO()
     cmd = CmdFactory(Graphx, outstream=output, task_list=_sample_tasks())
     cmd._execute(graph_type='json')
     got = output.getvalue()
     self.assertIn("read", got)
     self.assertIn("t3", got)
     self.assertIn("join_files", got)

 def testSubTask(self):
     output = StringIO()
     tasks = tasks_sample()
     cmd_list = CmdFactory(List, outstream=output, task_list=tasks)
     cmd_list._execute(subtasks=True)
     got = [line.strip() for line in output.getvalue().split('\n') if line]
     expected = [t.name for t in sorted(tasks)]
     assert expected == got

 def testSortByName(self):
     # by default, the task list should be ordered by name
     task_list = list(tasks_sample())
     output = StringIO()
     cmd_list = CmdFactory(List, outstream=output, task_list=task_list)
     cmd_list._execute()
     got = [line.strip() for line in output.getvalue().split('\n') if line]
     expected = ['g1', 't1', 't2', 't3']
     assert expected == got

 def testWithPrivate(self):
     task_list = list(tasks_sample())
     task_list.append(Task("_s3", [""]))
     output = StringIO()
     cmd_list = CmdFactory(List, outstream=output, task_list=task_list)
     cmd_list._execute(private=True, pos_args=['_s3'])
     got = [line.strip() for line in output.getvalue().split('\n') if line]
     expected = ['_s3']
     assert expected == got

 def testSortByDefinition(self):
     # test sorting task list by order of definition
     task_list = list(tasks_sample())
     output = StringIO()
     cmd_list = CmdFactory(List, outstream=output, task_list=task_list)
     cmd_list._execute(sort='definition')
     got = [line.strip() for line in output.getvalue().split('\n') if line]
     expected = ['t1', 't2', 'g1', 't3']
     assert expected == got

 def test_missing_file_dep(self, depfile):
     my_task = Task("t2", [""], file_dep=['tests/data/missing'])
     output = StringIO()
     cmd_reset = CmdFactory(ResetDep, outstream=output, task_list=[my_task],
                            dep_manager=depfile)
     cmd_reset._execute()
     got = output.getvalue()
     assert ("failed t2 (Dependent file 'tests/data/missing' does not "
             "exist.)\n") == got

 def test_file_dep_up_to_date(self, depfile, dependency1):
     my_task = Task("t2", [""], file_dep=['tests/data/dependency1'])
     depfile.save_success(my_task)
     output = StringIO()
     cmd_reset = CmdFactory(ResetDep, outstream=output, task_list=[my_task],
                            dep_manager=depfile)
     cmd_reset._execute()
     got = output.getvalue()
     assert "skip t2\n" == got

 def test_execute(self, depfile, dependency1):
     output = StringIO()
     tasks = tasks_sample()
     cmd_reset = CmdFactory(ResetDep, outstream=output, task_list=tasks,
                            dep_manager=depfile)
     cmd_reset._execute()
     got = [line.strip() for line in output.getvalue().split('\n') if line]
     expected = ["processed %s" % t.name for t in tasks]
     assert sorted(expected) == sorted(got)