taskcluster/taskgraph/test/test_target_tasks.py
author Andrew Halberstadt <ahalberstadt@mozilla.com>
Tue, 27 Jun 2017 13:33:20 -0700
changeset 422889 7af941648ca6696d49cbeb5f37b7c81b9292e824
parent 419715 77c1ead3ba581546677b5dcc8a62ce5f5096bda6
child 427442 a43cb4709418ca9b44a9328d79fa635fb6418c4b
permissions -rw-r--r--
Bug 1380306 - Create a new 'try_task_config' method for scheduling tasks on try, r=dustin This introduces a 'try_task_config' method of scheduling. En lieu of (or in addition to) try syntax, you can now check in a file called 'try_task_config.json' to the root of the source tree. The format is either a list of task labels, or dict where task labels are the keys. Taskcluster will simply schedule any tasks that are listed there. This file is primarily meant to be generated by tools (which don't exist yet), as the json format is much easier for tools to generate or consume. These tools should use an in-memory commit to add the file so it is automatically removed again after the push. A server-side hook will be added in bug 1380357 to prevent this file from accidentally landing on non-try trees. MozReview-Commit-ID: 2zKfZXuuDhH

# 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 os
import unittest

from taskgraph import target_tasks
from taskgraph import try_option_syntax
from taskgraph.graph import Graph
from taskgraph.taskgraph import TaskGraph
from taskgraph.task import Task
from mozunit import main


class FakeTryOptionSyntax(object):

    def __init__(self, message, task_graph):
        self.trigger_tests = 0
        self.talos_trigger_tests = 0
        self.notifications = None
        self.env = []
        self.profile = False
        self.tag = None
        self.no_retry = False

    def task_matches(self, task):
        return 'at-at' in task.attributes


class TestTargetTasks(unittest.TestCase):

    def default_matches(self, run_on_projects, project):
        method = target_tasks.get_method('default')
        graph = TaskGraph(tasks={
            'a': Task(kind='build', label='a',
                      attributes={'run_on_projects': run_on_projects},
                      task={}),
        }, graph=Graph(nodes={'a'}, edges=set()))
        parameters = {'project': project}
        return 'a' in method(graph, parameters)

    def test_default_all(self):
        """run_on_projects=[all] includes release, integration, and other projects"""
        self.assertTrue(self.default_matches(['all'], 'mozilla-central'))
        self.assertTrue(self.default_matches(['all'], 'mozilla-inbound'))
        self.assertTrue(self.default_matches(['all'], 'baobab'))

    def test_default_integration(self):
        """run_on_projects=[integration] includes integration projects"""
        self.assertFalse(self.default_matches(['integration'], 'mozilla-central'))
        self.assertTrue(self.default_matches(['integration'], 'mozilla-inbound'))
        self.assertFalse(self.default_matches(['integration'], 'baobab'))

    def test_default_release(self):
        """run_on_projects=[release] includes release projects"""
        self.assertTrue(self.default_matches(['release'], 'mozilla-central'))
        self.assertFalse(self.default_matches(['release'], 'mozilla-inbound'))
        self.assertFalse(self.default_matches(['release'], 'baobab'))

    def test_default_nothing(self):
        """run_on_projects=[] includes nothing"""
        self.assertFalse(self.default_matches([], 'mozilla-central'))
        self.assertFalse(self.default_matches([], 'mozilla-inbound'))
        self.assertFalse(self.default_matches([], 'baobab'))

    def test_try_tasks(self):
        tasks = {
            'a': Task(kind=None, label='a', attributes={}, task={}),
            'b': Task(kind=None, label='b', attributes={'at-at': 'yep'}, task={}),
            'c': Task(kind=None, label='c', attributes={}, task={}),
        }
        graph = Graph(nodes=set('abc'), edges=set())
        tg = TaskGraph(tasks, graph)

        method = target_tasks.get_method('try_tasks')
        config = os.path.join(os.getcwd(), 'try_task_config.json')

        orig_TryOptionSyntax = try_option_syntax.TryOptionSyntax
        try:
            try_option_syntax.TryOptionSyntax = FakeTryOptionSyntax

            # no try specifier
            self.assertEqual(method(tg, {'message': ''}), ['b'])

            # try syntax only
            self.assertEqual(method(tg, {'message': 'try: me'}), ['b'])

            # try task config only
            with open(config, 'w') as fh:
                fh.write('["c"]')
            self.assertEqual(method(tg, {'message': ''}), ['c'])

            with open(config, 'w') as fh:
                fh.write('{"c": {}}')
            self.assertEqual(method(tg, {'message': ''}), ['c'])

            # both syntax and config
            self.assertEqual(set(method(tg, {'message': 'try: me'})), set(['b', 'c']))
        finally:
            try_option_syntax.TryOptionSyntax = orig_TryOptionSyntax
            if os.path.isfile(config):
                os.remove(config)


if __name__ == '__main__':
    main()