taskcluster/taskgraph/test/test_util_runnable_jobs.py

   1 # This Source Code Form is subject to the terms of the Mozilla Public
   2 # License, v. 2.0. If a copy of the MPL was not distributed with this
   3 # file, You can obtain one at http://mozilla.org/MPL/2.0/.
   4
   5 from __future__ import absolute_import
   6
   7 import unittest
   8
   9 from taskgraph.decision import full_task_graph_to_runnable_jobs
  10 from taskgraph.graph import Graph
  11 from taskgraph.taskgraph import TaskGraph
  12 from taskgraph.task import Task
  13 from mozunit import main
  14
  15
  16 class TestRunnableJobs(unittest.TestCase):
  17
  18     tasks = [
  19         {
  20             'kind': 'build',
  21             'label': 'a',
  22             'attributes': {},
  23             'task': {
  24                 'extra': {
  25                     'treeherder': {
  26                         'symbol': 'B'
  27                     }
  28                 },
  29             }
  30         },
  31         {
  32             'kind': 'test',
  33             'label': 'b',
  34             'attributes': {},
  35             'task': {
  36                 'extra': {
  37                     'treeherder': {
  38                         'collection': {
  39                             'opt': True
  40                         },
  41                         'groupName': 'Some group',
  42                         'groupSymbol': 'GS',
  43                         'machine': {
  44                             'platform': 'linux64'
  45                         },
  46                         'symbol': 't'
  47                     }
  48                 },
  49             }
  50         },
  51     ]
  52
  53     def make_taskgraph(self, tasks):
  54         label_to_taskid = {k: k + '-tid' for k in tasks}
  55         for label, task_id in label_to_taskid.iteritems():
  56             tasks[label].task_id = task_id
  57         graph = Graph(nodes=set(tasks), edges=set())
  58         taskgraph = TaskGraph(tasks, graph)
  59         return taskgraph, label_to_taskid
  60
  61     def test_taskgraph_to_runnable_jobs(self):
  62         tg, label_to_taskid = self.make_taskgraph({
  63             t['label']: Task(**t) for t in self.tasks[:]
  64         })
  65
  66         res = full_task_graph_to_runnable_jobs(tg.to_json())
  67
  68         self.assertEqual(res, {
  69             'a': {
  70                 'symbol': 'B'
  71             },
  72             'b': {
  73                 'collection': {'opt': True},
  74                 'groupName': 'Some group',
  75                 'groupSymbol': 'GS',
  76                 'symbol': 't',
  77                 'platform': 'linux64'
  78             }
  79         })
  80
  81
  82 if __name__ == '__main__':
  83     main()