1#!/usr/bin/python
2#
3# Copyright (c) 2012 The Chromium OS Authors. All rights reserved.
4# Use of this source code is governed by a BSD-style license that can be
5# found in the LICENSE file.
6
7"""Unit tests for site_utils/task.py."""
8
9import mox, unittest
10
11# driver must be imported first due to circular imports in base_event and task
12import driver  # pylint: disable-msg=W0611
13import deduping_scheduler, forgiving_config_parser, task, build_event
14
15
16class TaskTestBase(mox.MoxTestBase):
17    """Common code for Task test classes
18
19    @var _BUILD: fake build.
20    @var _BOARD: fake board to reimage.
21    @var _BRANCH: fake branch to run tests on.
22    @var _BRANCH_SPEC: fake branch specification for Tasks.
23    @var _MAP: fake branch:build map.
24    @var _POOL: fake pool of machines to test on.
25    @var _SUITE: fake suite name.
26    @var _TASK_NAME: fake name for tasks in config.
27    """
28
29    _BUILD = 'build'
30    _BOARD = 'board1'
31    _BRANCH = '20'
32    _BRANCH_SPEC = '>=R' + _BRANCH
33    _BRANCH_SPEC_EQUAL = '==R' + _BRANCH
34    _BRANCH_SPEC_LTE = '<=R' + _BRANCH
35    _MAP = {_BRANCH: [_BUILD]}
36    _NUM = 2
37    _POOL = 'fake_pool'
38    _SUITE = 'suite'
39    _TASK_NAME = 'fake_task_name'
40    _PRIORITY = build_event.BuildEvent.PRIORITY
41    _TIMEOUT = build_event.BuildEvent.TIMEOUT
42    _FILE_BUGS=False
43
44
45    def setUp(self):
46        super(TaskTestBase, self).setUp()
47        self.sched = self.mox.CreateMock(deduping_scheduler.DedupingScheduler)
48
49
50class TaskCreateTest(TaskTestBase):
51    """Unit tests for Task.CreateFromConfigSection().
52
53    @var _EVENT_KEY: fake event-to-run-on keyword for tasks in config.
54    """
55
56    _EVENT_KEY = 'new_build'
57
58
59    def setUp(self):
60        super(TaskCreateTest, self).setUp()
61        self.config = forgiving_config_parser.ForgivingConfigParser()
62        self.config.add_section(self._TASK_NAME)
63        self.config.set(self._TASK_NAME, 'suite', self._SUITE)
64        self.config.set(self._TASK_NAME, 'branch_specs', self._BRANCH_SPEC)
65        self.config.set(self._TASK_NAME, 'run_on', self._EVENT_KEY)
66        self.config.set(self._TASK_NAME, 'pool', self._POOL)
67        self.config.set(self._TASK_NAME, 'num', '%d' % self._NUM)
68        self.config.set(self._TASK_NAME, 'boards', self._BOARD)
69
70
71    def testCreateFromConfig(self):
72        """Ensure a Task can be built from a correct config."""
73        keyword, new_task = task.Task.CreateFromConfigSection(self.config,
74                                                              self._TASK_NAME)
75        self.assertEquals(keyword, self._EVENT_KEY)
76        self.assertEquals(new_task, task.Task(self._TASK_NAME, self._SUITE,
77                                              [self._BRANCH_SPEC], self._POOL,
78                                              self._NUM, self._BOARD,
79                                              self._PRIORITY, self._TIMEOUT))
80        self.assertTrue(new_task._FitsSpec(self._BRANCH))
81        self.assertFalse(new_task._FitsSpec('12'))
82
83
84    def testCreateFromConfigEqualBranch(self):
85        """Ensure a Task can be built from a correct config with support of
86        branch_specs: ==RXX."""
87        # Modify the branch_specs setting in self.config.
88        self.config.set(self._TASK_NAME, 'branch_specs',
89                        self._BRANCH_SPEC_EQUAL)
90        keyword, new_task = task.Task.CreateFromConfigSection(self.config,
91                                                              self._TASK_NAME)
92        self.assertEquals(keyword, self._EVENT_KEY)
93        self.assertEquals(new_task, task.Task(self._TASK_NAME, self._SUITE,
94                                              [self._BRANCH_SPEC_EQUAL],
95                                              self._POOL, self._NUM,
96                                              self._BOARD, self._PRIORITY,
97                                              self._TIMEOUT))
98        self.assertTrue(new_task._FitsSpec(self._BRANCH))
99        self.assertFalse(new_task._FitsSpec('12'))
100        self.assertFalse(new_task._FitsSpec('21'))
101        # Reset the branch_specs setting in self.config to >=R.
102        self.config.set(self._TASK_NAME, 'branch_specs', self._BRANCH_SPEC)
103
104
105    def testCreateFromConfigLessThanOrEqualBranch(self):
106        """Ensure a Task can be built from a correct config with support of
107        branch_specs: <=RXX."""
108        # Modify the branch_specs setting in self.config.
109        self.config.set(self._TASK_NAME, 'branch_specs',
110                        self._BRANCH_SPEC_LTE)
111        keyword, new_task = task.Task.CreateFromConfigSection(self.config,
112                                                              self._TASK_NAME)
113        self.assertEquals(keyword, self._EVENT_KEY)
114        self.assertEquals(new_task, task.Task(self._TASK_NAME, self._SUITE,
115                                              [self._BRANCH_SPEC_LTE],
116                                              self._POOL, self._NUM,
117                                              self._BOARD, self._PRIORITY,
118                                              self._TIMEOUT))
119        self.assertTrue(new_task._FitsSpec(self._BRANCH))
120        self.assertTrue(new_task._FitsSpec('12'))
121        self.assertFalse(new_task._FitsSpec('21'))
122        # Reset the branch_specs setting in self.config to >=R.
123        self.config.set(self._TASK_NAME, 'branch_specs', self._BRANCH_SPEC)
124
125
126    def testCreateFromConfigNoBranch(self):
127        """Ensure a Task can be built from a correct config with no branch."""
128        self.config.remove_option(self._TASK_NAME, 'branch_specs')
129        keyword, new_task = task.Task.CreateFromConfigSection(self.config,
130                                                              self._TASK_NAME)
131        self.assertEquals(keyword, self._EVENT_KEY)
132        self.assertEquals(new_task, task.Task(self._TASK_NAME, self._SUITE,
133                                              [], self._POOL, self._NUM,
134                                              self._BOARD, self._PRIORITY,
135                                              self._TIMEOUT))
136        self.assertTrue(new_task._FitsSpec(self._BRANCH))
137
138
139    def testCreateFromConfigMultibranch(self):
140        """Ensure a Task can be built from a correct config with >1 branches."""
141        specs = ['factory', self._BRANCH_SPEC]
142        self.config.set(self._TASK_NAME, 'branch_specs', ','.join(specs))
143        keyword, new_task = task.Task.CreateFromConfigSection(self.config,
144                                                              self._TASK_NAME)
145        self.assertEquals(keyword, self._EVENT_KEY)
146        self.assertEquals(new_task, task.Task(self._TASK_NAME, self._SUITE,
147                                              specs, self._POOL, self._NUM,
148                                              self._BOARD, self._PRIORITY,
149                                              self._TIMEOUT))
150        for spec in [specs[0], self._BRANCH]:
151            self.assertTrue(new_task._FitsSpec(spec))
152
153
154    def testCreateFromConfigNoNum(self):
155        """Ensure a Task can be built from a correct config with no num."""
156        self.config.remove_option(self._TASK_NAME, 'num')
157        keyword, new_task = task.Task.CreateFromConfigSection(self.config,
158                                                              self._TASK_NAME)
159        self.assertEquals(keyword, self._EVENT_KEY)
160        self.assertEquals(new_task, task.Task(self._TASK_NAME, self._SUITE,
161                                              [self._BRANCH_SPEC], self._POOL,
162                                              boards=self._BOARD))
163        self.assertTrue(new_task._FitsSpec(self._BRANCH))
164        self.assertFalse(new_task._FitsSpec('12'))
165
166
167    def testCreateFromNoSuiteConfig(self):
168        """Ensure we require a suite in Task config."""
169        self.config.remove_option(self._TASK_NAME, 'suite')
170        self.assertRaises(task.MalformedConfigEntry,
171                          task.Task.CreateFromConfigSection,
172                          self.config,
173                          self._TASK_NAME)
174
175
176    def testCreateFromNoKeywordConfig(self):
177        """Ensure we require a run_on event in Task config."""
178        self.config.remove_option(self._TASK_NAME, 'run_on')
179        self.assertRaises(task.MalformedConfigEntry,
180                          task.Task.CreateFromConfigSection,
181                          self.config,
182                          self._TASK_NAME)
183
184
185    def testCreateFromNonexistentConfig(self):
186        """Ensure we fail gracefully if we pass in a bad section name."""
187        self.assertRaises(task.MalformedConfigEntry,
188                          task.Task.CreateFromConfigSection,
189                          self.config,
190                          'not_a_thing')
191
192
193    def testFileBugsNoConfigValue(self):
194        """Ensure not setting file bugs in a config leads to file_bugs=False."""
195        keyword, new_task = task.Task.CreateFromConfigSection(self.config,
196                                                              self._TASK_NAME)
197        self.assertFalse(new_task._file_bugs)
198
199
200class TaskTest(TaskTestBase):
201    """Unit tests for Task."""
202
203
204    def setUp(self):
205        super(TaskTest, self).setUp()
206        self.task = task.Task(self._TASK_NAME, self._SUITE, [self._BRANCH_SPEC],
207                              None, None, self._BOARD, self._PRIORITY,
208                              self._TIMEOUT)
209
210
211    def testRun(self):
212        """Test running a recurring task."""
213        self.sched.ScheduleSuite(self._SUITE, self._BOARD, self._BUILD,
214                                 None, None, self._PRIORITY, self._TIMEOUT,
215                                 False, file_bugs=self._FILE_BUGS,
216                                 firmware_rw_build=None,
217                                 test_source_build=None,
218                                 job_retry=False).AndReturn(True)
219        self.mox.ReplayAll()
220        self.assertTrue(self.task.Run(self.sched, self._MAP, self._BOARD))
221
222
223    def testRunCustomSharding(self):
224        """Test running a recurring task with non-default sharding."""
225        expected_sharding = 2
226        mytask = task.Task(self._TASK_NAME, self._SUITE, [self._BRANCH_SPEC],
227                           num=expected_sharding)
228        self.sched.ScheduleSuite(self._SUITE, self._BOARD, self._BUILD,
229                                 None, expected_sharding, None, None,
230                                 False, file_bugs=self._FILE_BUGS,
231                                 firmware_rw_build=None,
232                                 test_source_build=None,
233                                 job_retry=False).AndReturn(True)
234        self.mox.ReplayAll()
235        self.assertTrue(mytask.Run(self.sched, self._MAP, self._BOARD))
236
237
238    def testRunDuplicate(self):
239        """Test running a task that schedules a duplicate suite task."""
240        self.sched.ScheduleSuite(self._SUITE, self._BOARD, self._BUILD,
241                                 None, None, self._PRIORITY, self._TIMEOUT,
242                                 False, file_bugs=self._FILE_BUGS,
243                                 firmware_rw_build=None,
244                                 test_source_build=None,
245                                 job_retry=False).AndReturn(True)
246        self.mox.ReplayAll()
247        self.assertTrue(self.task.Run(self.sched, self._MAP, self._BOARD))
248
249
250    def testRunUnrunnablePool(self):
251        """Test running a task that cannot run on this pool."""
252        self.sched.CheckHostsExist(
253                multiple_labels=mox.IgnoreArg()).AndReturn(None)
254        self.mox.ReplayAll()
255        t = task.Task(self._TASK_NAME, self._SUITE,
256                      [self._BRANCH_SPEC], "BadPool")
257        self.assertTrue(not t.AvailableHosts(self.sched, self._BOARD))
258
259
260    def testRunUnrunnableBoard(self):
261        """Test running a task that cannot run on this board."""
262        self.mox.ReplayAll()
263        t = task.Task(self._TASK_NAME, self._SUITE,
264                      [self._BRANCH_SPEC], self._POOL, boards="BadBoard")
265        self.assertTrue(not t.AvailableHosts(self.sched, self._BOARD))
266
267
268    def testNoRunBranchMismatch(self):
269        """Test running a recurring task with no matching builds."""
270        t = task.Task(self._TASK_NAME, self._SUITE, task.BARE_BRANCHES)
271        self.mox.ReplayAll()
272        self.assertTrue(t.Run(self.sched, self._MAP, self._BOARD))
273
274
275    def testNoRunBareBranchMismatch(self):
276        """Test running a recurring task with no matching builds (factory)."""
277        self.mox.ReplayAll()
278        self.assertTrue(
279            self.task.Run(self.sched, {'factory': 'build2'}, self._BOARD))
280
281
282    def testRunNoSpec(self):
283        """Test running a recurring task with default branch specs."""
284        t = task.Task(self._TASK_NAME, self._SUITE, [])
285        self.sched.ScheduleSuite(self._SUITE, self._BOARD, self._BUILD,
286                                 None, None, None, None,
287                                 False, file_bugs=self._FILE_BUGS,
288                                 firmware_rw_build=None,
289                                 test_source_build=None,
290                                 job_retry=False).AndReturn(True)
291        self.mox.ReplayAll()
292        self.assertTrue(t.Run(self.sched, self._MAP, self._BOARD))
293
294
295    def testRunExplodes(self):
296        """Test a failure to schedule while running task."""
297        # Barf while scheduling.
298        self.sched.ScheduleSuite(
299            self._SUITE, self._BOARD, self._BUILD, None, None, self._PRIORITY,
300            self._TIMEOUT, False, file_bugs=self._FILE_BUGS,
301            firmware_rw_build=None, test_source_build=None,
302            job_retry=False).AndRaise(
303                    deduping_scheduler.ScheduleException('Simulated Failure'))
304        self.mox.ReplayAll()
305        self.assertTrue(self.task.Run(self.sched, self._MAP, self._BOARD))
306
307
308    def testForceRun(self):
309        """Test force running a recurring task."""
310        self.sched.ScheduleSuite(self._SUITE, self._BOARD, self._BUILD,
311                                 None, None, self._PRIORITY, self._TIMEOUT,
312                                 True, file_bugs=self._FILE_BUGS,
313                                 firmware_rw_build=None,
314                                 test_source_build=None,
315                                 job_retry=False).AndReturn(True)
316        self.mox.ReplayAll()
317        self.assertTrue(self.task.Run(self.sched, self._MAP, self._BOARD, True))
318
319
320    def testHash(self):
321        """Test hash function for Task classes."""
322        same_task = task.Task(self._TASK_NAME, self._SUITE, [self._BRANCH_SPEC],
323                              boards=self._BOARD)
324        other_task = task.Task(self._TASK_NAME, self._SUITE,
325                               [self._BRANCH_SPEC, '>=RX1'], 'pool')
326        self.assertEquals(hash(self.task), hash(same_task))
327        self.assertNotEquals(hash(self.task), hash(other_task))
328
329
330class OneShotTaskTest(TaskTestBase):
331    """Unit tests for OneShotTask."""
332
333
334    def setUp(self):
335        super(OneShotTaskTest, self).setUp()
336        self.task = task.OneShotTask(self._TASK_NAME, self._SUITE,
337                                     [self._BRANCH_SPEC])
338
339
340    def testRun(self):
341        """Test running a one-shot task."""
342        self.sched.ScheduleSuite(self._SUITE, self._BOARD, self._BUILD,
343                                 None, None, None, None, False,
344                                 file_bugs=self._FILE_BUGS,
345                                 firmware_rw_build=None,
346                                 test_source_build=None,
347                                 job_retry=False).AndReturn(True)
348        self.mox.ReplayAll()
349        self.assertFalse(self.task.Run(self.sched, self._MAP, self._BOARD))
350
351
352    def testRunDuplicate(self):
353        """Test running a one-shot task that schedules a dup suite task."""
354        self.sched.ScheduleSuite(self._SUITE, self._BOARD, self._BUILD,
355                                 None, None, None, None, False,
356                                 file_bugs=self._FILE_BUGS,
357                                 firmware_rw_build=None,
358                                 test_source_build=None,
359                                 job_retry=False).AndReturn(False)
360        self.mox.ReplayAll()
361        self.assertFalse(self.task.Run(self.sched, self._MAP, self._BOARD))
362
363
364    def testRunExplodes(self):
365        """Test a failure to schedule while running one-shot task."""
366        # Barf while scheduling.
367        self.sched.ScheduleSuite(
368            self._SUITE, self._BOARD, self._BUILD, None, None,
369            None, None, False, file_bugs=self._FILE_BUGS,
370            firmware_rw_build=None, test_source_build=None,
371            job_retry=False).AndRaise(
372                deduping_scheduler.ScheduleException('Simulated Failure'))
373        self.mox.ReplayAll()
374        self.assertFalse(self.task.Run(self.sched, self._MAP, self._BOARD))
375
376
377    def testForceRun(self):
378        """Test force running a one-shot task."""
379        self.sched.ScheduleSuite(self._SUITE, self._BOARD, self._BUILD,
380                                 None, None, None, None, True,
381                                 file_bugs=self._FILE_BUGS,
382                                 firmware_rw_build=None,
383                                 test_source_build=None,
384                                 job_retry=False).AndReturn(True)
385        self.mox.ReplayAll()
386        self.assertFalse(self.task.Run(self.sched, self._MAP, self._BOARD,
387                                       force=True))
388
389
390    def testFileBugs(self):
391        """Test that file_bugs is passed from the task to ScheduleSuite."""
392        self.sched.ScheduleSuite(self._SUITE, self._BOARD, self._BUILD,
393                                 None, None, None, None, True,
394                                 file_bugs=True, firmware_rw_build=None,
395                                 test_source_build=None,
396                                 job_retry=False).AndReturn(True)
397        self.mox.ReplayAll()
398        self.task._file_bugs = True
399        self.assertFalse(self.task.Run(self.sched, self._MAP, self._BOARD,
400                                       force=True))
401
402
403if __name__ == '__main__':
404    unittest.main()
405