forked from pydoit/doit
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathtest_cmd_resetdep.py
98 lines (85 loc) · 4.03 KB
/
test_cmd_resetdep.py
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
from six 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, 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)
def test_file_dep(self, depfile, dependency1):
my_task = Task("t2", [""], file_dep=['tests/data/dependency1'])
output = StringIO()
cmd_reset = CmdFactory(ResetDep, outstream=output, task_list=[my_task],
dep_manager=depfile)
cmd_reset._execute()
got = output.getvalue()
assert "processed t2\n" == got
dep = list(my_task.file_dep)[0]
timestamp, size, md5 = depfile._get(my_task.name, dep)
assert get_file_md5(get_abspath("data/dependency1")) == md5
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_file_dep_change_checker(self, depfile, dependency1):
my_task = Task("t2", [""], file_dep=['tests/data/dependency1'])
depfile.save_success(my_task)
depfile.checker = TimestampChecker()
output = StringIO()
cmd_reset = CmdFactory(ResetDep, outstream=output, task_list=[my_task],
dep_manager=depfile)
cmd_reset._execute()
got = output.getvalue()
assert "processed t2\n" == 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 test_invalid_task(self, depfile):
output = StringIO()
tasks = tasks_sample()
cmd_reset = CmdFactory(ResetDep, outstream=output, task_list=tasks,
dep_manager=depfile)
pytest.raises(InvalidCommand, cmd_reset._execute, pos_args=['xxx'])
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_values_and_results(self, depfile, dependency1):
my_task = Task("t2", [""], file_dep=['tests/data/dependency1'])
my_task.result = "result"
my_task.values = {'x': 5, 'y': 10}
depfile.save_success(my_task)
depfile.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=depfile)
cmd_reset._execute()
got = output.getvalue()
assert "processed t2\n" == got
assert {'x': 5, 'y': 10} == depfile.get_values(reseted.name)
assert get_md5('result') == depfile.get_result(reseted.name)