taskcluster/taskgraph/test/test_taskgraph.py
author Dustin J. Mitchell <dustin@mozilla.com>
Wed, 04 Jul 2018 02:46:59 +0000
changeset 488165 ab58645e9230620ca45de3ec03ee9e61eb4a7cbf
parent 432957 e4f1ed133c6826aabc73672cd379037d5c05952b
child 520117 5caf48a420eb337a296c8b337332f2933884f467
permissions -rw-r--r--
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

# 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()