DXR is a code search and navigation tool aimed at making sense of large projects. It supports full-text and regex searches as well as structural queries.

Mercurial (1aeaa33a64f9)

VCS Links

Line Code
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
# 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

import unittest

from taskgraph.decision import full_task_graph_to_runnable_jobs
from taskgraph.graph import Graph
from taskgraph.taskgraph import TaskGraph
from taskgraph.task import Task
from mozunit import main


class TestRunnableJobs(unittest.TestCase):

    tasks = [
        {
            'kind': 'build',
            'label': 'a',
            'attributes': {},
            'task': {
                'extra': {
                    'treeherder': {
                        'symbol': 'B'
                    }
                },
            }
        },
        {
            'kind': 'test',
            'label': 'b',
            'attributes': {},
            'task': {
                'extra': {
                    'treeherder': {
                        'collection': {
                            'opt': True
                        },
                        'groupName': 'Some group',
                        'groupSymbol': 'GS',
                        'machine': {
                            'platform': 'linux64'
                        },
                        'symbol': 't'
                    }
                },
            }
        },
    ]

    def make_taskgraph(self, tasks):
        label_to_taskid = {k: k + '-tid' for k in tasks}
        for label, task_id in label_to_taskid.iteritems():
            tasks[label].task_id = task_id
        graph = Graph(nodes=set(tasks), edges=set())
        taskgraph = TaskGraph(tasks, graph)
        return taskgraph, label_to_taskid

    def test_taskgraph_to_runnable_jobs(self):
        tg, label_to_taskid = self.make_taskgraph({
            t['label']: Task(**t) for t in self.tasks[:]
        })

        res = full_task_graph_to_runnable_jobs(tg.to_json())

        self.assertEqual(res, {
            'a': {
                'symbol': 'B'
            },
            'b': {
                'collection': {'opt': True},
                'groupName': 'Some group',
                'groupSymbol': 'GS',
                'symbol': 't',
                'platform': 'linux64'
            }
        })


if __name__ == '__main__':
    main()