|
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
|
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at http://mozilla.org/MPL/2.0/.
from __future__ import absolute_import, print_function, unicode_literals
import unittest
import os
import mock
from taskgraph import create
from taskgraph.config import GraphConfig
from taskgraph.graph import Graph
from taskgraph.taskgraph import TaskGraph
from taskgraph.task import Task
from mozunit import main
GRAPH_CONFIG = GraphConfig({'trust-domain': 'domain'}, '/var/empty')
class TestCreate(unittest.TestCase):
def setUp(self):
self.old_task_id = os.environ.get('TASK_ID')
if 'TASK_ID' in os.environ:
del os.environ['TASK_ID']
self.created_tasks = {}
self.old_create_task = create.create_task
create.create_task = self.fake_create_task
def tearDown(self):
create.create_task = self.old_create_task
if self.old_task_id:
os.environ['TASK_ID'] = self.old_task_id
elif 'TASK_ID' in os.environ:
del os.environ['TASK_ID']
def fake_create_task(self, session, task_id, label, task_def):
self.created_tasks[task_id] = task_def
def test_create_tasks(self):
tasks = {
'tid-a': Task(kind='test', label='a', attributes={}, task={'payload': 'hello world'}),
'tid-b': Task(kind='test', label='b', attributes={}, task={'payload': 'hello world'}),
}
label_to_taskid = {'a': 'tid-a', 'b': 'tid-b'}
graph = Graph(nodes={'tid-a', 'tid-b'}, edges={('tid-a', 'tid-b', 'edge')})
taskgraph = TaskGraph(tasks, graph)
create.create_tasks(GRAPH_CONFIG, taskgraph, label_to_taskid, {'level': '4'})
for tid, task in self.created_tasks.iteritems():
self.assertEqual(task['payload'], 'hello world')
self.assertEqual(task['schedulerId'], 'domain-level-4')
# make sure the dependencies exist, at least
for depid in task.get('dependencies', []):
if depid == 'decisiontask':
# Don't look for decisiontask here
continue
self.assertIn(depid, self.created_tasks)
def test_create_task_without_dependencies(self):
"a task with no dependencies depends on the decision task"
os.environ['TASK_ID'] = 'decisiontask'
tasks = {
'tid-a': Task(kind='test', label='a', attributes={}, task={'payload': 'hello world'}),
}
label_to_taskid = {'a': 'tid-a'}
graph = Graph(nodes={'tid-a'}, edges=set())
taskgraph = TaskGraph(tasks, graph)
create.create_tasks(GRAPH_CONFIG, taskgraph, label_to_taskid, {'level': '4'})
for tid, task in self.created_tasks.iteritems():
self.assertEqual(task.get('dependencies'), [os.environ['TASK_ID']])
@mock.patch('taskgraph.create.create_task')
def test_create_tasks_fails_if_create_fails(self, create_task):
"creat_tasks fails if a single create_task call fails"
os.environ['TASK_ID'] = 'decisiontask'
tasks = {
'tid-a': Task(kind='test', label='a', attributes={}, task={'payload': 'hello world'}),
}
label_to_taskid = {'a': 'tid-a'}
graph = Graph(nodes={'tid-a'}, edges=set())
taskgraph = TaskGraph(tasks, graph)
def fail(*args):
print("UHOH")
raise RuntimeError('oh noes!')
create_task.side_effect = fail
with self.assertRaises(RuntimeError):
create.create_tasks(GRAPH_CONFIG, taskgraph, label_to_taskid, {'level': '4'})
if __name__ == '__main__':
main()
|