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
5from __future__ import absolute_import, print_function, unicode_literals
6
7import contextlib
8import unittest
9
10from taskgraph import target_tasks
11from taskgraph import try_option_syntax
12from taskgraph.graph import Graph
13from taskgraph.taskgraph import TaskGraph
14from taskgraph.task import Task
15from mozunit import main
16
17
18class FakeTryOptionSyntax(object):
19
20    def __init__(self, message, task_graph, graph_config):
21        self.trigger_tests = 0
22        self.talos_trigger_tests = 0
23        self.raptor_trigger_tests = 0
24        self.notifications = None
25        self.env = []
26        self.profile = False
27        self.tag = None
28        self.no_retry = False
29
30    def task_matches(self, task):
31        return 'at-at' in task.attributes
32
33
34class TestTargetTasks(unittest.TestCase):
35
36    def default_matches_project(self, run_on_projects, project):
37        return self.default_matches(
38            attributes={
39                'run_on_projects': run_on_projects,
40            },
41            parameters={
42                'project': project,
43                'hg_branch': 'default',
44            },
45        )
46
47    def default_matches_hg_branch(self, run_on_hg_branches, hg_branch):
48        attributes = {'run_on_projects': ['all']}
49        if run_on_hg_branches is not None:
50            attributes['run_on_hg_branches'] = run_on_hg_branches
51
52        return self.default_matches(
53            attributes=attributes,
54            parameters={
55                'project': 'mozilla-central',
56                'hg_branch': hg_branch,
57            },
58        )
59
60    def default_matches(self, attributes, parameters):
61        method = target_tasks.get_method('default')
62        graph = TaskGraph(tasks={
63            'a': Task(kind='build', label='a',
64                      attributes=attributes,
65                      task={}),
66        }, graph=Graph(nodes={'a'}, edges=set()))
67        return 'a' in method(graph, parameters, {})
68
69    def test_default_all(self):
70        """run_on_projects=[all] includes release, integration, and other projects"""
71        self.assertTrue(self.default_matches_project(['all'], 'mozilla-central'))
72        self.assertTrue(self.default_matches_project(['all'], 'baobab'))
73
74    def test_default_integration(self):
75        """run_on_projects=[integration] includes integration projects"""
76        self.assertFalse(self.default_matches_project(['integration'], 'mozilla-central'))
77        self.assertFalse(self.default_matches_project(['integration'], 'baobab'))
78
79    def test_default_release(self):
80        """run_on_projects=[release] includes release projects"""
81        self.assertTrue(self.default_matches_project(['release'], 'mozilla-central'))
82        self.assertFalse(self.default_matches_project(['release'], 'baobab'))
83
84    def test_default_nothing(self):
85        """run_on_projects=[] includes nothing"""
86        self.assertFalse(self.default_matches_project([], 'mozilla-central'))
87        self.assertFalse(self.default_matches_project([], 'baobab'))
88
89    def test_default_hg_branch(self):
90        self.assertTrue(self.default_matches_hg_branch(None, 'default'))
91        self.assertTrue(self.default_matches_hg_branch(None, 'GECKOVIEW_62_RELBRANCH'))
92
93        self.assertFalse(self.default_matches_hg_branch([], 'default'))
94        self.assertFalse(self.default_matches_hg_branch([], 'GECKOVIEW_62_RELBRANCH'))
95
96        self.assertTrue(self.default_matches_hg_branch(['all'], 'default'))
97        self.assertTrue(self.default_matches_hg_branch(['all'], 'GECKOVIEW_62_RELBRANCH'))
98
99        self.assertTrue(self.default_matches_hg_branch(['default'], 'default'))
100        self.assertTrue(self.default_matches_hg_branch([r'default'], 'default'))
101        self.assertFalse(self.default_matches_hg_branch([r'default'], 'GECKOVIEW_62_RELBRANCH'))
102
103        self.assertTrue(
104            self.default_matches_hg_branch(['GECKOVIEW_62_RELBRANCH'], 'GECKOVIEW_62_RELBRANCH')
105        )
106        self.assertTrue(
107            self.default_matches_hg_branch(['GECKOVIEW_\d+_RELBRANCH'], 'GECKOVIEW_62_RELBRANCH')
108        )
109        self.assertTrue(
110            self.default_matches_hg_branch([r'GECKOVIEW_\d+_RELBRANCH'], 'GECKOVIEW_62_RELBRANCH')
111        )
112        self.assertFalse(self.default_matches_hg_branch([r'GECKOVIEW_\d+_RELBRANCH'], 'default'))
113
114    def make_task_graph(self):
115        tasks = {
116            'a': Task(kind=None, label='a', attributes={}, task={}),
117            'b': Task(kind=None, label='b', attributes={'at-at': 'yep'}, task={}),
118            'c': Task(kind=None, label='c', attributes={'run_on_projects': ['try']}, task={}),
119        }
120        graph = Graph(nodes=set('abc'), edges=set())
121        return TaskGraph(tasks, graph)
122
123    @contextlib.contextmanager
124    def fake_TryOptionSyntax(self):
125        orig_TryOptionSyntax = try_option_syntax.TryOptionSyntax
126        try:
127            try_option_syntax.TryOptionSyntax = FakeTryOptionSyntax
128            yield
129        finally:
130            try_option_syntax.TryOptionSyntax = orig_TryOptionSyntax
131
132    def test_empty_try(self):
133        "try_mode = None runs nothing"
134        tg = self.make_task_graph()
135        method = target_tasks.get_method('try_tasks')
136        params = {
137            'try_mode': None,
138            'project': 'try',
139            'message': '',
140        }
141        # only runs the task with run_on_projects: try
142        self.assertEqual(method(tg, params, {}), [])
143
144    def test_try_option_syntax(self):
145        "try_mode = try_option_syntax uses TryOptionSyntax"
146        tg = self.make_task_graph()
147        method = target_tasks.get_method('try_tasks')
148        with self.fake_TryOptionSyntax():
149            params = {
150                'try_mode': 'try_option_syntax',
151                'message': 'try: -p all',
152            }
153            self.assertEqual(method(tg, params, {}), ['b'])
154
155    def test_try_task_config(self):
156        "try_mode = try_task_config uses the try config"
157        tg = self.make_task_graph()
158        method = target_tasks.get_method('try_tasks')
159        params = {
160            'try_mode': 'try_task_config',
161            'try_task_config': {'tasks': ['a']},
162        }
163        self.assertEqual(method(tg, params, {}), ['a'])
164
165
166if __name__ == '__main__':
167    main()
168