import os
import sys
import pickle
from multiprocessing import Queue
import platform
from unittest.mock import Mock

import pytest

from doit.exceptions import BaseFail, InvalidTask
from doit.dependency import DbmDB, Dependency
from doit.reporter import ConsoleReporter
from doit.task import Task, DelayedLoader
from doit.control import TaskDispatcher, ExecNode
from doit import runner


PLAT_IMPL = platform.python_implementation()

# sample actions
def my_print(*args):
    pass
def _fail():
    return False
def _error():
    raise Exception("I am the exception.\n")
def _exit():
    raise SystemExit()
def simple_result():
    print("success output")
    print("success error", file=sys.stderr)
    return 'my-result'
def simple_fail():
    print("simple output")
    print("simple error", file=sys.stderr)
    raise Exception('this task failed')

class FakeReporter(object):
    """Just log everything in internal attribute - used on tests"""
    def __init__(self, with_exceptions=False, outstream=None, options=None):
        # include Exception object of log failures
        self.with_exceptions = with_exceptions
        self.log = []

    def get_status(self, task):
        self.log.append(('start', task))

    def execute_task(self, task):
        self.log.append(('execute', task))

    def add_failure(self, task, exception):
        if self.with_exceptions:
            self.log.append(('fail', task, exception))
        else:
            self.log.append(('fail', task))

    def add_success(self, task):
        self.log.append(('success', task))

    def skip_uptodate(self, task):
        self.log.append(('up-to-date', task))

    def skip_ignore(self, task):
        self.log.append(('ignore', task))

    def cleanup_error(self, exception):
        self.log.append(('cleanup_error',))

    def runtime_error(self, msg):
        self.log.append(('runtime_error',))

    def teardown_task(self, task):
        self.log.append(('teardown', task))

    def complete_run(self):
        pass


@pytest.fixture
def reporter(request):
    return FakeReporter()


class TestRunner(object):
    def testInit(self, reporter, dep_manager):
        my_runner = runner.Runner(dep_manager, reporter)
        assert False == my_runner._stop_running
        assert runner.SUCCESS == my_runner.final_result


class TestRunner_SelectTask(object):
    def test_ready(self, reporter, dep_manager):
        t1 = Task("taskX", [(my_print, ["out a"] )])
        my_runner = runner.Runner(dep_manager, reporter)
        assert True == my_runner.select_task(ExecNode(t1, None), {})
        assert ('start', t1) == reporter.log.pop(0)
        assert not reporter.log

    def test_DependencyError(self, reporter, dep_manager):
        t1 = Task("taskX", [(my_print, ["out a"] )],
                  file_dep=["i_dont_exist"])
        my_runner = runner.Runner(dep_manager, reporter)
        assert False == my_runner.select_task(ExecNode(t1, None), {})
        assert ('start', t1) == reporter.log.pop(0)
        assert ('fail', t1) == reporter.log.pop(0)
        assert not reporter.log

    def test_upToDate(self, reporter, dep_manager):
        t1 = Task("taskX", [(my_print, ["out a"] )], file_dep=[__file__])
        my_runner = runner.Runner(dep_manager, reporter)
        my_runner.dep_manager.save_success(t1)
        assert False == my_runner.select_task(ExecNode(t1, None), {})
        assert ('start', t1) == reporter.log.pop(0)
        assert ('up-to-date', t1) == reporter.log.pop(0)
        assert not reporter.log

    def test_ignore(self, reporter, dep_manager):
        t1 = Task("taskX", [(my_print, ["out a"] )])
        my_runner = runner.Runner(dep_manager, reporter)
        my_runner.dep_manager.ignore(t1)
        assert False == my_runner.select_task(ExecNode(t1, None), {})
        assert ('start', t1) == reporter.log.pop(0)
        assert ('ignore', t1) == reporter.log.pop(0)
        assert not reporter.log

    def test_alwaysExecute(self, reporter, dep_manager):
        t1 = Task("taskX", [(my_print, ["out a"] )], uptodate=[True])
        my_runner = runner.Runner(dep_manager, reporter, always_execute=True)
        my_runner.dep_manager.save_success(t1)
        n1 = ExecNode(t1, None)
        assert True == my_runner.select_task(n1, {})
        # run_status is set to run even if task is up-to-date
        assert n1.run_status == 'run'
        assert ('start', t1) == reporter.log.pop(0)
        assert not reporter.log

    def test_noSetup_ok(self, reporter, dep_manager):
        t1 = Task("taskX", [(my_print, ["out a"] )])
        my_runner = runner.Runner(dep_manager, reporter)
        assert True == my_runner.select_task(ExecNode(t1, None), {})
        assert ('start', t1) == reporter.log.pop(0)
        assert not reporter.log

    def test_withSetup(self, reporter, dep_manager):
        t1 = Task("taskX", [(my_print, ["out a"] )], setup=["taskY"])
        my_runner = runner.Runner(dep_manager, reporter)
        # defer execution
        n1 = ExecNode(t1, None)
        assert False == my_runner.select_task(n1, {})
        assert ('start', t1) == reporter.log.pop(0)
        assert not reporter.log
        # trying to select again
        assert True == my_runner.select_task(n1, {})
        assert not reporter.log


    def test_getargs_ok(self, reporter, dep_manager):
        def ok(): return {'x':1}
        def check_x(my_x): return my_x == 1
        t1 = Task('t1', [(ok,)])
        n1 = ExecNode(t1, None)
        t2 = Task('t2', [(check_x,)], getargs={'my_x':('t1','x')})
        n2 = ExecNode(t2, None)
        tasks_dict = {'t1': t1, 't2':t2}
        my_runner = runner.Runner(dep_manager, reporter)

        # t2 gives chance for setup tasks to be executed
        assert False == my_runner.select_task(n2, tasks_dict)
        assert ('start', t2) == reporter.log.pop(0)

        # execute task t1 to calculate value
        assert True == my_runner.select_task(n1, tasks_dict)
        assert ('start', t1) == reporter.log.pop(0)
        t1_result = my_runner.execute_task(t1)
        assert ('execute', t1) == reporter.log.pop(0)
        my_runner.process_task_result(n1, t1_result)
        assert ('success', t1) == reporter.log.pop(0)

        # t2.options are set on select_task
        assert True == my_runner.select_task(n2, tasks_dict)
        assert not reporter.log
        assert {'my_x': 1} == t2.options

    def test_getargs_fail(self, reporter, dep_manager):
        # invalid getargs. Exception wil be raised and task will fail
        def check_x(my_x): return True
        t1 = Task('t1', [lambda :True])
        n1 = ExecNode(t1, None)
        t2 = Task('t2', [(check_x,)], getargs={'my_x':('t1','x')})
        n2 = ExecNode(t2, None)
        tasks_dict = {'t1': t1, 't2':t2}
        my_runner = runner.Runner(dep_manager, reporter)

        # t2 gives chance for setup tasks to be executed
        assert False == my_runner.select_task(n2, tasks_dict)
        assert ('start', t2) == reporter.log.pop(0)

        # execute task t1 to calculate value
        assert True == my_runner.select_task(n1, tasks_dict)
        assert ('start', t1) == reporter.log.pop(0)
        t1_result = my_runner.execute_task(t1)
        assert ('execute', t1) == reporter.log.pop(0)
        my_runner.process_task_result(n1, t1_result)
        assert ('success', t1) == reporter.log.pop(0)

        # select_task t2 fails
        assert False == my_runner.select_task(n2, tasks_dict)
        assert ('fail', t2) == reporter.log.pop(0)
        assert not reporter.log


    def test_getargs_dict(self, reporter, dep_manager):
        def ok(): return {'x':1}
        t1 = Task('t1', [(ok,)])
        n1 = ExecNode(t1, None)
        t2 = Task('t2', None, getargs={'my_x':('t1', None)})
        tasks_dict = {'t1': t1, 't2':t2}
        my_runner = runner.Runner(dep_manager, reporter)
        t1_result = my_runner.execute_task(t1)
        my_runner.process_task_result(n1, t1_result)

        # t2.options are set on _get_task_args
        my_runner._get_task_args(t2, tasks_dict)
        assert {'my_x': {'x':1}} == t2.options


    def test_getargs_group(self, reporter, dep_manager):
        def ok(): return {'x':1}
        t1 = Task('t1', None, task_dep=['t1:a'], has_subtask=True)
        t1a = Task('t1:a', [(ok,)], subtask_of='t1')
        t2 = Task('t2', None, getargs={'my_x':('t1', None)})
        tasks_dict = {'t1': t1, 't1a':t1a, 't2':t2}
        my_runner = runner.Runner(dep_manager, reporter)
        t1a_result = my_runner.execute_task(t1a)
        my_runner.process_task_result(ExecNode(t1a, None), t1a_result)

        # t2.options are set on _get_task_args
        my_runner._get_task_args(t2, tasks_dict)
        assert {'my_x': {'a':{'x':1}} } == t2.options



    def test_getargs_group_value(self, reporter, dep_manager):
        def ok(): return {'x':1}
        t1 = Task('t1', None, task_dep=['t1:a'], has_subtask=True)
        t1a = Task('t1:a', [(ok,)], subtask_of='t1')
        t2 = Task('t2', None, getargs={'my_x':('t1', 'x')})
        tasks_dict = {'t1': t1, 't1a':t1a, 't2':t2}
        my_runner = runner.Runner(dep_manager, reporter)
        t1a_result = my_runner.execute_task(t1a)
        my_runner.process_task_result(ExecNode(t1a, None), t1a_result)

        # t2.options are set on _get_task_args
        my_runner._get_task_args(t2, tasks_dict)
        assert {'my_x': {'a':1} } == t2.options



class TestTask_Teardown(object):
    def test_ok(self, reporter, dep_manager):
        touched = []
        def touch():
            touched.append(1)
        t1 = Task('t1', [], teardown=[(touch,)])
        my_runner = runner.Runner(dep_manager, reporter)
        my_runner.teardown_list = [t1]
        t1.execute(my_runner.stream)
        my_runner.teardown()
        assert 1 == len(touched)
        assert ('teardown', t1) == reporter.log.pop(0)
        assert not reporter.log

    def test_reverse_order(self, reporter, dep_manager):
        def do_nothing():pass
        t1 = Task('t1', [], teardown=[do_nothing])
        t2 = Task('t2', [], teardown=[do_nothing])
        my_runner = runner.Runner(dep_manager, reporter)
        my_runner.teardown_list = [t1, t2]
        t1.execute(my_runner.stream)
        t2.execute(my_runner.stream)
        my_runner.teardown()
        assert ('teardown', t2) == reporter.log.pop(0)
        assert ('teardown', t1) == reporter.log.pop(0)
        assert not reporter.log

    def test_errors(self, reporter, dep_manager):
        def raise_something(x):
            raise Exception(x)
        t1 = Task('t1', [], teardown=[(raise_something,['t1 blow'])])
        t2 = Task('t2', [], teardown=[(raise_something,['t2 blow'])])
        my_runner = runner.Runner(dep_manager, reporter)
        my_runner.teardown_list = [t1, t2]
        t1.execute(my_runner.stream)
        t2.execute(my_runner.stream)
        my_runner.teardown()
        assert ('teardown', t2) == reporter.log.pop(0)
        assert ('cleanup_error',) == reporter.log.pop(0)
        assert ('teardown', t1) == reporter.log.pop(0)
        assert ('cleanup_error',) == reporter.log.pop(0)
        assert not reporter.log


class TestTask_RunAll(object):
    def test_reporter_runtime_error(self, reporter, dep_manager):
        t1 = Task('t1', [], calc_dep=['t2'])
        t2 = Task('t2', [lambda: {'file_dep':[1]}])
        my_runner = runner.Runner(dep_manager, reporter)
        my_runner.run_all(TaskDispatcher({'t1':t1, 't2':t2}, [], ['t1', 't2']))
        assert runner.ERROR == my_runner.final_result
        assert ('start', t2) == reporter.log.pop(0)
        assert ('execute', t2) == reporter.log.pop(0)
        assert ('success', t2) == reporter.log.pop(0)
        assert ('runtime_error',) == reporter.log.pop(0)
        assert not reporter.log


# run tests in both single process runner and multi-process runner
RUNNERS = [runner.Runner, runner.MThreadRunner]
# TODO: test should be added and skipped!
if runner.MRunner.available():
    RUNNERS.append(runner.MRunner)
@pytest.fixture(params=RUNNERS)
def RunnerClass(request):
    return request.param


# function used on actions, define here to make sure they are pickable
def ok(): return "ok"
def ok2(): return "different"
def my_action():
    import sys
    sys.stdout.write('out here')
    sys.stderr.write('err here')
    return {'bb': 5}
def use_args(arg1):
    print(arg1)
def make_args():
    return {'myarg':1}
def action_add_filedep(task, extra_dep):
    task.file_dep.add(extra_dep)


class TestRunner_run_tasks(object):

    def test_teardown(self, reporter, RunnerClass, dep_manager):
        t1 = Task('t1', [], teardown=[ok])
        t2 = Task('t2', [])
        my_runner = RunnerClass(dep_manager, reporter)
        assert [] == my_runner.teardown_list
        my_runner.run_tasks(TaskDispatcher({'t1':t1, 't2':t2}, [], ['t1', 't2']))
        my_runner.finish()
        assert ('teardown', t1) == reporter.log[-1]

    # testing whole process/API
    def test_success(self, reporter, RunnerClass, dep_manager):
        t1 = Task("t1", [(my_print, ["out a"] )] )
        t2 = Task("t2", [(my_print, ["out a"] )] )
        my_runner = RunnerClass(dep_manager, reporter)
        my_runner.run_tasks(TaskDispatcher({'t1':t1, 't2':t2}, [], ['t1', 't2']))
        assert runner.SUCCESS == my_runner.finish()
        assert ('start', t1) == reporter.log.pop(0), reporter.log
        assert ('execute', t1) == reporter.log.pop(0)
        assert ('success', t1) == reporter.log.pop(0)
        assert ('start', t2) == reporter.log.pop(0)
        assert ('execute', t2) == reporter.log.pop(0)
        assert ('success', t2) == reporter.log.pop(0)

    # test result, value, out, err are saved into task
    def test_result(self, reporter, RunnerClass, dep_manager):
        task = Task("taskY", [my_action] )
        my_runner = RunnerClass(dep_manager, reporter)
        assert None == task.result
        assert {} == task.values
        assert [None] == [a.out for a in task.actions]
        assert [None] == [a.err for a in task.actions]
        my_runner.run_tasks(TaskDispatcher({'taskY':task}, [], ['taskY']))
        assert runner.SUCCESS == my_runner.finish()
        assert {'bb': 5} == task.result
        assert {'bb': 5} == task.values
        assert ['out here'] == [a.out for a in task.actions]
        assert ['err here'] == [a.err for a in task.actions]

    # whenever a task fails remaining task are not executed
    def test_failureOutput(self, reporter, RunnerClass, dep_manager):
        t1 = Task("t1", [_fail])
        t2 = Task("t2", [_fail])
        my_runner = RunnerClass(dep_manager, reporter)
        my_runner.run_tasks(TaskDispatcher({'t1':t1, 't2':t2}, [], ['t1', 't2']))
        assert runner.FAILURE == my_runner.finish()
        assert ('start', t1) == reporter.log.pop(0)
        assert ('execute', t1) == reporter.log.pop(0)
        assert ('fail', t1) == reporter.log.pop(0)
        # second task is not executed
        assert 0 == len(reporter.log)


    def test_error(self, reporter, RunnerClass, dep_manager):
        t1 = Task("t1", [_error])
        t2 = Task("t2", [_error])
        my_runner = RunnerClass(dep_manager, reporter)
        my_runner.run_tasks(TaskDispatcher({'t1':t1, 't2':t2}, [], ['t1', 't2']))
        assert runner.ERROR == my_runner.finish()
        assert ('start', t1) == reporter.log.pop(0)
        assert ('execute', t1) == reporter.log.pop(0)
        assert ('fail', t1) == reporter.log.pop(0)
        # second task is not executed
        assert 0 == len(reporter.log)

    def test_dependency_error_after_execution(self, dep_manager):
        t1 = Task("t1", [(my_print, ["out a"] )],
                  file_dep=["i_dont_exist"], targets=['not_there'])
        reporter = FakeReporter(with_exceptions=True)
        my_runner = runner.Runner(dep_manager, reporter)
        # Missing file_dep is not caught because check is short-circuited by
        # missing target.
        my_runner.run_tasks(TaskDispatcher({'t1':t1}, [], ['t1']))
        assert runner.ERROR == my_runner.finish()
        print(reporter.log)
        assert ('start', t1) == reporter.log.pop(0)
        assert ('execute', t1) == reporter.log.pop(0)
        fail_log = reporter.log.pop(0)
        assert ('fail', t1) == fail_log[:2]
        assert "Dependent file 'i_dont_exist' does not exist" in str(fail_log[2])
        assert not reporter.log


    # when successful dependencies are updated
    def test_updateDependencies(self, reporter, RunnerClass, depfile_name):
        depPath = os.path.join(os.path.dirname(__file__), "data", "dependency1")
        ff = open(depPath,"a")
        ff.write("xxx")
        ff.close()
        dependencies = [depPath]

        filePath = os.path.join(os.path.dirname(__file__), "data", "target")
        ff = open(filePath,"a")
        ff.write("xxx")
        ff.close()
        targets = [filePath]

        t1 = Task("t1", [my_print], dependencies, targets)
        dep_manager = Dependency(DbmDB, depfile_name)
        my_runner = RunnerClass(dep_manager, reporter)
        my_runner.run_tasks(TaskDispatcher({'t1':t1}, [], ['t1']))
        assert runner.SUCCESS == my_runner.finish()
        d = Dependency(DbmDB, depfile_name)
        assert d._get("t1", os.path.abspath(depPath))


    def test_continue(self, reporter, RunnerClass, dep_manager):
        t1 = Task("t1", [(_fail,)] )
        t2 = Task("t2", [(_error,)] )
        t3 = Task("t3", [(ok,)])
        my_runner = RunnerClass(dep_manager, reporter, continue_=True)
        disp = TaskDispatcher({'t1':t1, 't2':t2, 't3':t3}, [], ['t1', 't2', 't3'])
        my_runner.run_tasks(disp)
        assert runner.ERROR == my_runner.finish()
        assert ('start', t1) == reporter.log.pop(0)
        assert ('execute', t1) == reporter.log.pop(0)
        assert ('fail', t1) == reporter.log.pop(0)
        assert ('start', t2) == reporter.log.pop(0)
        assert ('execute', t2) == reporter.log.pop(0)
        assert ('fail', t2) == reporter.log.pop(0)
        assert ('start', t3) == reporter.log.pop(0)
        assert ('execute', t3) == reporter.log.pop(0)
        assert ('success', t3) == reporter.log.pop(0)
        assert 0 == len(reporter.log)


    def test_continue_dont_execute_parent_of_failed_task(self, reporter,
                                                         RunnerClass, dep_manager):
        t1 = Task("t1", [(_error,)] )
        t2 = Task("t2", [(ok,)], task_dep=['t1'])
        t3 = Task("t3", [(ok,)])
        my_runner = RunnerClass(dep_manager, reporter, continue_=True)
        disp = TaskDispatcher({'t1':t1, 't2':t2, 't3':t3}, [], ['t1', 't2', 't3'])
        my_runner.run_tasks(disp)
        assert runner.ERROR == my_runner.finish()
        assert ('start', t1) == reporter.log.pop(0)
        assert ('execute', t1) == reporter.log.pop(0)
        assert ('fail', t1) == reporter.log.pop(0)
        assert ('start', t2) == reporter.log.pop(0)
        assert ('fail', t2) == reporter.log.pop(0)
        assert ('start', t3) == reporter.log.pop(0)
        assert ('execute', t3) == reporter.log.pop(0)
        assert ('success', t3) == reporter.log.pop(0)
        assert 0 == len(reporter.log)


    def test_continue_dep_error(self, reporter, RunnerClass, dep_manager):
        t1 = Task("t1", [(ok,)], file_dep=['i_dont_exist'] )
        t2 = Task("t2", [(ok,)], task_dep=['t1'])
        my_runner = RunnerClass(dep_manager, reporter, continue_=True)
        disp = TaskDispatcher({'t1':t1, 't2':t2}, [], ['t1', 't2'])
        my_runner.run_tasks(disp)
        assert runner.ERROR == my_runner.finish()
        assert ('start', t1) == reporter.log.pop(0)
        assert ('fail', t1) == reporter.log.pop(0)
        assert ('start', t2) == reporter.log.pop(0)
        assert ('fail', t2) == reporter.log.pop(0)
        assert 0 == len(reporter.log)


    def test_continue_ignored_dep(self, reporter, RunnerClass, dep_manager):
        t1 = Task("t1", [(ok,)], )
        t2 = Task("t2", [(ok,)], task_dep=['t1'])
        my_runner = RunnerClass(dep_manager, reporter, continue_=True)
        my_runner.dep_manager.ignore(t1)
        disp = TaskDispatcher({'t1':t1, 't2':t2}, [], ['t1', 't2'])
        my_runner.run_tasks(disp)
        assert runner.SUCCESS == my_runner.finish()
        assert ('start', t1) == reporter.log.pop(0)
        assert ('ignore', t1) == reporter.log.pop(0)
        assert ('start', t2) == reporter.log.pop(0)
        assert ('ignore', t2) == reporter.log.pop(0)
        assert 0 == len(reporter.log)


    def test_getargs(self, reporter, RunnerClass, dep_manager):
        t1 = Task("t1", [(use_args,)], getargs=dict(arg1=('t2','myarg')) )
        t2 = Task("t2", [(make_args,)])
        my_runner = RunnerClass(dep_manager, reporter)
        my_runner.run_tasks(TaskDispatcher({'t1':t1, 't2':t2}, [], ['t1', 't2']))
        assert runner.SUCCESS == my_runner.finish()
        assert ('start', t1) == reporter.log.pop(0)
        assert ('start', t2) == reporter.log.pop(0)
        assert ('execute', t2) == reporter.log.pop(0)
        assert ('success', t2) == reporter.log.pop(0)
        assert ('execute', t1) == reporter.log.pop(0)
        assert ('success', t1) == reporter.log.pop(0)
        assert 0 == len(reporter.log)


    def testActionModifiesFiledep(self, reporter, RunnerClass, dep_manager):
        extra_dep = os.path.join(os.path.dirname(__file__), 'sample_md5.txt')
        t1 = Task("t1", [(my_print, ["out a"] ),
                         (action_add_filedep, (), {'extra_dep': extra_dep})
                     ] )
        my_runner = RunnerClass(dep_manager, reporter)
        my_runner.run_tasks(TaskDispatcher({'t1':t1}, [], ['t1']))
        assert runner.SUCCESS == my_runner.finish()
        assert ('start', t1) == reporter.log.pop(0), reporter.log
        assert ('execute', t1) == reporter.log.pop(0)
        assert ('success', t1) == reporter.log.pop(0)
        assert t1.file_dep == set([extra_dep])

    # SystemExit runner should not interfere with SystemExit
    def testSystemExitRaises(self, reporter, RunnerClass, dep_manager):
        t1 = Task("t1", [_exit])
        my_runner = RunnerClass(dep_manager, reporter)
        disp = TaskDispatcher({'t1':t1}, [], ['t1'])
        pytest.raises(SystemExit, my_runner.run_tasks, disp)
        my_runner.finish()


@pytest.mark.skipif('not runner.MRunner.available()')
class TestMReporter(object):
    class MyRunner(object):
        def __init__(self):
            self.result_q = Queue()

    def testReporterMethod(self, reporter):
        fake_runner = self.MyRunner()
        mp_reporter = runner.MReporter(fake_runner, reporter)
        my_task = Task("task x", [])
        mp_reporter.add_success(my_task)
        # note limit is 2 seconds because of http://bugs.python.org/issue17707
        got = fake_runner.result_q.get(True, 2)
        assert {'name': "task x", "reporter": 'add_success'} == got

    def testNonReporterMethod(self, reporter):
        fake_runner = self.MyRunner()
        mp_reporter = runner.MReporter(fake_runner, reporter)
        assert hasattr(mp_reporter, 'add_success')
        assert not hasattr(mp_reporter, 'no_existent_method')


class TestJobTask(object):
    def test_closure_is_picklable(self):
        # can pickle because we use cloudpickle
        def non_top_function(): return 4
        t1 = Task('t1', [non_top_function])
        t1p = runner.JobTask(t1).task_pickle
        t2 = pickle.loads(t1p)
        assert 4 == t2.actions[0].py_callable()

    def test_not_picklable_raises_InvalidTask(self):
        def non_top_function(): pass
        class Unpicklable:
            def __getstate__(self):
                raise pickle.PicklingError("DO NOT PICKLE")
        d1 = Unpicklable()
        t1 = Task('t1', [non_top_function, (d1,)])
        pytest.raises(InvalidTask, runner.JobTask, t1)


# multiprocessing on Windows requires the whole object to be pickable
def test_MRunner_pickable(dep_manager):
    t1 = Task('t1', [])
    import sys
    reporter = ConsoleReporter(sys.stdout, {})
    run = runner.MRunner(dep_manager, reporter)
    run._run_tasks_init(TaskDispatcher({'t1':t1}, [], ['t1']))
    # assert nothing is raised
    pickle.dumps(run)


@pytest.mark.skipif('not runner.MRunner.available()')
class TestMRunner_get_next_job(object):
    # simple normal case
    def test_run_task(self, reporter, dep_manager):
        t1 = Task('t1', [])
        t2 = Task('t2', [])
        run = runner.MRunner(dep_manager, reporter)
        run._run_tasks_init(TaskDispatcher({'t1':t1, 't2':t2}, [], ['t1', 't2']))
        assert t1.name == run.get_next_job(None).name
        assert t2.name == run.get_next_job(None).name
        assert None == run.get_next_job(None)

    def test_stop_running(self, reporter, dep_manager):
        t1 = Task('t1', [])
        t2 = Task('t2', [])
        run = runner.MRunner(dep_manager, reporter)
        run._run_tasks_init(TaskDispatcher({'t1':t1, 't2':t2}, [], ['t1', 't2']))
        assert t1.name == run.get_next_job(None).name
        run._stop_running = True
        assert None == run.get_next_job(None)

    def test_waiting(self, reporter, dep_manager):
        t1 = Task('t1', [])
        t2 = Task('t2', [], setup=('t1',))
        run = runner.MRunner(dep_manager, reporter)
        dispatcher = TaskDispatcher({'t1':t1, 't2':t2}, [], ['t2'])
        run._run_tasks_init(dispatcher)

        # first start task 1
        j1 = run.get_next_job(None)
        assert t1.name == j1.name

        # hold until t1 is done
        assert isinstance(run.get_next_job(None), runner.JobHold)
        assert isinstance(run.get_next_job(None), runner.JobHold)

        n1 = dispatcher.nodes[j1.name]
        n1.run_status = 'done'
        j2 = run.get_next_job(n1)
        assert t2.name == j2.name
        assert None == run.get_next_job(dispatcher.nodes[j2.name])


    def test_waiting_controller(self, reporter, dep_manager):
        t1 = Task('t1', [])
        t2 = Task('t2', [], calc_dep=('t1',))
        run = runner.MRunner(dep_manager, reporter)
        run._run_tasks_init(TaskDispatcher({'t1':t1, 't2':t2}, [], ['t1', 't2']))

        # first task ok
        assert t1.name == run.get_next_job(None).name

        # hold until t1 finishes
        assert 0 == run.free_proc
        assert isinstance(run.get_next_job(None), runner.JobHold)
        assert 1 == run.free_proc


    def test_delayed_loaded(self, reporter, dep_manager):
        def create():
            return {'basename':'t1', 'actions': None}
        t1 = Task('t1', [], loader=DelayedLoader(create, executed='t2'))
        t2 = Task('t2', [])
        run = runner.MRunner(dep_manager, reporter)
        dispatcher = TaskDispatcher({'t1':t1, 't2':t2}, [], ['t1', 't2'])
        run._run_tasks_init(dispatcher)
        assert t2.name == run.get_next_job(None).name
        assert runner.JobHold.type == run.get_next_job(None).type

        # after t2 is done t1 can be dispatched
        n2 = dispatcher.nodes[t2.name]
        n2.run_status = 'done'
        j1 = run.get_next_job(n2)
        assert t1.name == j1.name
        # the job for t1 contains the whole task since sub-process dont
        # have it
        assert j1.type == runner.JobTask.type



@pytest.mark.skipif('not runner.MRunner.available()')
class TestMRunner_start_process(object):
    # 2 process, 3 tasks
    def test_all_processes(self, reporter, monkeypatch, dep_manager):
        mock_process = Mock()
        monkeypatch.setattr(runner.MRunner, 'Child', mock_process)
        t1 = Task('t1', [])
        t2 = Task('t2', [])
        td = TaskDispatcher({'t1':t1, 't2':t2}, [], ['t1', 't2'])
        run = runner.MRunner(dep_manager, reporter, num_process=2)
        run._run_tasks_init(td)
        result_q = Queue()
        task_q = Queue()

        proc_list = run._run_start_processes(task_q, result_q)
        run.finish()
        assert 2 == len(proc_list)
        assert t1.name == task_q.get().name
        assert t2.name == task_q.get().name


    # 2 process, 1 task
    def test_less_processes(self, reporter, monkeypatch, dep_manager):
        mock_process = Mock()
        monkeypatch.setattr(runner.MRunner, 'Child', mock_process)
        t1 = Task('t1', [])
        td = TaskDispatcher({'t1':t1}, [], ['t1'])
        run = runner.MRunner(dep_manager, reporter, num_process=2)
        run._run_tasks_init(td)
        result_q = Queue()
        task_q = Queue()

        proc_list = run._run_start_processes(task_q, result_q)
        run.finish()
        assert 1 == len(proc_list)
        assert t1.name == task_q.get().name


    # 2 process, 2 tasks (but only one task can be started)
    def test_waiting_process(self, reporter, monkeypatch, dep_manager):
        mock_process = Mock()
        monkeypatch.setattr(runner.MRunner, 'Child', mock_process)
        t1 = Task('t1', [])
        t2 = Task('t2', [], task_dep=['t1'])
        td = TaskDispatcher({'t1':t1, 't2':t2}, [], ['t1', 't2'])
        run = runner.MRunner(dep_manager, reporter, num_process=2)
        run._run_tasks_init(td)
        result_q = Queue()
        task_q = Queue()

        proc_list = run._run_start_processes(task_q, result_q)
        run.finish()
        assert 2 == len(proc_list)
        assert t1.name == task_q.get().name
        assert isinstance(task_q.get(), runner.JobHold)


def non_pickable_creator():
    return {'basename': 't2', 'actions': [lambda: True]}

class TestMRunner_parallel_run_tasks(object):

    @pytest.mark.skipif('not runner.MRunner.available()')
    def test_task_cloudpicklabe_multiprocess(self, reporter, dep_manager):
        t1 = Task("t1", [(my_print, ["out a"] )] )
        t2 = Task("t2", None, loader=DelayedLoader(
            non_pickable_creator, executed='t1'))
        my_runner = runner.MRunner(dep_manager, reporter)
        dispatcher = TaskDispatcher({'t1':t1, 't2':t2}, [], ['t1', 't2'])
        my_runner.run_tasks(dispatcher)
        assert runner.SUCCESS == my_runner.finish()

    def test_task_not_picklabe_thread(self, reporter, dep_manager):
        t1 = Task("t1", [(my_print, ["out a"] )] )
        t2 = Task("t2", None, loader=DelayedLoader(
            non_pickable_creator, executed='t1'))
        my_runner = runner.MThreadRunner(dep_manager, reporter)
        dispatcher = TaskDispatcher({'t1':t1, 't2':t2}, [], ['t1', 't2'])
        # threaded code have no problems with closures
        my_runner.run_tasks(dispatcher)
        assert runner.SUCCESS == my_runner.finish()
        assert ('start', t1) == reporter.log.pop(0), reporter.log
        assert ('execute', t1) == reporter.log.pop(0)
        assert ('success', t1) == reporter.log.pop(0)
        assert ('start', t2) == reporter.log.pop(0)
        assert ('execute', t2) == reporter.log.pop(0)
        assert ('success', t2) == reporter.log.pop(0)



@pytest.mark.skipif('not runner.MRunner.available()')
class TestMRunner_execute_task(object):
    def test_hold(self, reporter, dep_manager):
        run = runner.MRunner(dep_manager, reporter)
        task_q = Queue()
        task_q.put(runner.JobHold()) # to test
        task_q.put(None) # to terminate function
        result_q = Queue()
        run.execute_task_subprocess(task_q, result_q, reporter.__class__)
        run.finish()
        # nothing was done
        assert result_q.empty()

    def test_full_task(self, reporter, dep_manager):
        # test execute_task_subprocess can receive a full Task object
        run = runner.MRunner(dep_manager, reporter)
        t1 = Task('t1', [simple_result])
        task_q = Queue()
        task_q.put(runner.JobTask(t1)) # to test
        task_q.put(None) # to terminate function
        result_q = Queue()
        run.execute_task_subprocess(task_q, result_q, reporter.__class__)
        run.finish()
        # check result
        assert result_q.get() == {'name': 't1', 'reporter': 'execute_task'}
        res = result_q.get()
        assert res['task']['result'] == 'my-result'
        assert res['task']['executed']
        assert res['out'] == ['success output\n']
        assert res['err'] == ['success error\n']
        assert result_q.empty()


    def test_full_task_fail(self, reporter, dep_manager):
        # test execute_task_subprocess can receive a full Task object
        run = runner.MRunner(dep_manager, reporter)
        t1 = Task('t1', [simple_fail])
        task_q = Queue()
        task_q.put(runner.JobTask(t1)) # to test
        task_q.put(None) # to terminate function
        result_q = Queue()
        run.execute_task_subprocess(task_q, result_q, reporter.__class__)
        run.finish()
        # check result
        assert result_q.get() == {'name': 't1', 'reporter': 'execute_task'}
        res = result_q.get()
        assert res['name'] == 't1'
        assert isinstance(res['failure'], BaseFail)
        assert res['out'] == ['simple output\n']
        assert res['err'] == ['simple error\n']
        # assert result_q.get()['task']['result'] == 'my-result'
        assert result_q.empty()



def test_MThreadRunner_available():
    assert runner.MThreadRunner.available() == True