1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123
|
from io import StringIO
import pytest
from doit.cmd_resetdep import ResetDep
from doit.dependency import TimestampChecker, get_md5, get_file_md5
from doit.exceptions import InvalidCommand
from doit.task import Task
from tests.conftest import tasks_sample, CmdFactory, get_abspath
class TestCmdResetDep(object):
def test_execute(self, dep_manager, dependency1):
output = StringIO()
tasks = tasks_sample()
cmd_reset = CmdFactory(ResetDep, outstream=output, task_list=tasks,
dep_manager=dep_manager)
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)
def test_file_dep(self, dep_manager, dependency1):
my_task = Task("t2", [""], file_dep=['tests/data/dependency1'])
output = StringIO()
cmd_reset = CmdFactory(ResetDep, outstream=output, task_list=[my_task],
dep_manager=dep_manager)
cmd_reset._execute()
got = output.getvalue()
assert "processed t2\n" == got
dep = list(my_task.file_dep)[0]
timestamp, size, md5 = dep_manager._get(my_task.name, dep)
assert get_file_md5(get_abspath("data/dependency1")) == md5
def test_file_dep_up_to_date(self, dep_manager, dependency1):
my_task = Task("t2", [""], file_dep=['tests/data/dependency1'])
dep_manager.save_success(my_task)
output = StringIO()
cmd_reset = CmdFactory(ResetDep, outstream=output, task_list=[my_task],
dep_manager=dep_manager)
cmd_reset._execute()
got = output.getvalue()
assert "skip t2\n" == got
def test_file_dep_change_checker(self, dep_manager, dependency1):
my_task = Task("t2", [""], file_dep=['tests/data/dependency1'])
dep_manager.save_success(my_task)
dep_manager.checker = TimestampChecker()
output = StringIO()
cmd_reset = CmdFactory(ResetDep, outstream=output, task_list=[my_task],
dep_manager=dep_manager)
cmd_reset._execute()
got = output.getvalue()
assert "processed t2\n" == got
def test_filter(self, dep_manager, dependency1):
output = StringIO()
tasks = tasks_sample()
cmd_reset = CmdFactory(ResetDep, outstream=output, task_list=tasks,
dep_manager=dep_manager)
cmd_reset._execute(pos_args=['t2'])
got = output.getvalue()
assert "processed t2\n" == got
def test_invalid_task(self, dep_manager):
output = StringIO()
tasks = tasks_sample()
cmd_reset = CmdFactory(ResetDep, outstream=output, task_list=tasks,
dep_manager=dep_manager)
pytest.raises(InvalidCommand, cmd_reset._execute, pos_args=['xxx'])
def test_missing_file_dep(self, dep_manager):
my_task = Task("t2", [""], file_dep=['tests/data/missing'])
output = StringIO()
cmd_reset = CmdFactory(ResetDep, outstream=output, task_list=[my_task],
dep_manager=dep_manager)
cmd_reset._execute()
got = output.getvalue()
assert ("failed t2 (Dependent file 'tests/data/missing' does not "
"exist.)\n") == got
def test_missing_dep_and_target(self, dep_manager, dependency1, dependency2):
task_a = Task("task_a", [""],
file_dep=['tests/data/dependency1'],
targets=['tests/data/dependency2'])
task_b = Task("task_b", [""],
file_dep=['tests/data/dependency2'],
targets=['tests/data/dependency3'])
task_c = Task("task_c", [""],
file_dep=['tests/data/dependency3'],
targets=['tests/data/dependency4'])
output = StringIO()
tasks = [task_a, task_b, task_c]
cmd_reset = CmdFactory(ResetDep, outstream=output, task_list=tasks,
dep_manager=dep_manager)
cmd_reset._execute()
got = output.getvalue()
assert ("processed task_a\n"
"processed task_b\n"
"failed task_c (Dependent file 'tests/data/dependency3'"
" does not exist.)\n") == got
def test_values_and_results(self, dep_manager, dependency1):
my_task = Task("t2", [""], file_dep=['tests/data/dependency1'])
my_task.result = "result"
my_task.values = {'x': 5, 'y': 10}
dep_manager.save_success(my_task)
dep_manager.checker = TimestampChecker() # trigger task update
reseted = Task("t2", [""], file_dep=['tests/data/dependency1'])
output = StringIO()
cmd_reset = CmdFactory(ResetDep, outstream=output, task_list=[reseted],
dep_manager=dep_manager)
cmd_reset._execute()
got = output.getvalue()
assert "processed t2\n" == got
assert {'x': 5, 'y': 10} == dep_manager.get_values(reseted.name)
assert get_md5('result') == dep_manager.get_result(reseted.name)
|