fune/taskcluster/taskgraph/test/test_taskgraph.py
Dustin J. Mitchell 91f4fe8c61 Bug 1398277: special-case retriggering of tasks not in the taskgraph; r=bstack
This will apply to cron tasks, action tasks, and decision tasks.  It is a
distinct retrigger implementation because (a) we do not want to follow
dependencies, and (b) it takes a lot of scopes to create a decision task, so we
need to limit access to this action.

MozReview-Commit-ID: 21DVSiagcrO

--HG--
extra : rebase_source : 6f027e349e245e4aa4dbed81145a0a5d75218cb1
extra : histedit_source : eff99aee5a0e7496b0734748b29739480eb0e3fb
2018-07-04 02:46:59 +00:00

100 lines
3.2 KiB
Python

# 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
from taskgraph.graph import Graph
from taskgraph.task import Task
from taskgraph.taskgraph import TaskGraph
from mozunit import main
class TestTaskGraph(unittest.TestCase):
maxDiff = None
def test_taskgraph_to_json(self):
tasks = {
'a': Task(kind='test', label='a',
attributes={'attr': 'a-task'},
task={'taskdef': True}),
'b': Task(kind='test', label='b',
attributes={},
task={'task': 'def'},
optimization={'seta': None},
# note that this dep is ignored, superseded by that
# from the taskgraph's edges
dependencies={'first': 'a'}),
}
graph = Graph(nodes=set('ab'), edges={('a', 'b', 'edgelabel')})
taskgraph = TaskGraph(tasks, graph)
res = taskgraph.to_json()
self.assertEqual(res, {
'a': {
'kind': 'test',
'label': 'a',
'attributes': {'attr': 'a-task', 'kind': 'test'},
'task': {'taskdef': True},
'dependencies': {'edgelabel': 'b'},
'optimization': None,
},
'b': {
'kind': 'test',
'label': 'b',
'attributes': {'kind': 'test'},
'task': {'task': 'def'},
'dependencies': {},
'optimization': {'seta': None},
}
})
def test_round_trip(self):
graph = TaskGraph(tasks={
'a': Task(
kind='fancy',
label='a',
attributes={},
dependencies={'prereq': 'b'}, # must match edges, below
optimization={'seta': None},
task={'task': 'def'}),
'b': Task(
kind='pre',
label='b',
attributes={},
dependencies={},
optimization={'seta': None},
task={'task': 'def2'}),
}, graph=Graph(nodes={'a', 'b'}, edges={('a', 'b', 'prereq')}))
tasks, new_graph = TaskGraph.from_json(graph.to_json())
self.assertEqual(graph, new_graph)
simple_graph = TaskGraph(tasks={
'a': Task(
kind='fancy',
label='a',
attributes={},
dependencies={'prereq': 'b'}, # must match edges, below
optimization={'seta': None},
task={'task': 'def'}),
'b': Task(
kind='pre',
label='b',
attributes={},
dependencies={},
optimization={'seta': None},
task={'task': 'def2'}),
}, graph=Graph(nodes={'a', 'b'}, edges={('a', 'b', 'prereq')}))
def test_contains(self):
assert 'a' in self.simple_graph
assert 'c' not in self.simple_graph
if __name__ == '__main__':
main()