• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
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 testCreateFromInvalidCrOSSuiteConfig(self):
194        """Ensure testbed_dut_count specified in boards is only applicable for
195        testing Launch Control builds."""
196        self.config.set(self._TASK_NAME, 'boards', 'shamu-2')
197        self.assertRaises(task.MalformedConfigEntry,
198                          task.Task.CreateFromConfigSection,
199                          self.config,
200                          self._TASK_NAME)
201
202
203    def testFileBugsNoConfigValue(self):
204        """Ensure not setting file bugs in a config leads to file_bugs=False."""
205        keyword, new_task = task.Task.CreateFromConfigSection(self.config,
206                                                              self._TASK_NAME)
207        self.assertFalse(new_task._file_bugs)
208
209
210class TaskTest(TaskTestBase):
211    """Unit tests for Task."""
212
213
214    def setUp(self):
215        super(TaskTest, self).setUp()
216        self.task = task.Task(self._TASK_NAME, self._SUITE, [self._BRANCH_SPEC],
217                              None, None, self._BOARD, self._PRIORITY,
218                              self._TIMEOUT)
219
220
221    def testRun(self):
222        """Test running a recurring task."""
223        self.sched.ScheduleSuite(self._SUITE, self._BOARD, self._BUILD,
224                                 None, None, self._PRIORITY, self._TIMEOUT,
225                                 False, file_bugs=self._FILE_BUGS,
226                                 firmware_rw_build=None,
227                                 firmware_ro_build=None,
228                                 test_source_build=None,
229                                 job_retry=False,
230                                 launch_control_build=None,
231                                 run_prod_code=False,
232                                 testbed_dut_count=None,
233                                 no_delay=False).AndReturn(True)
234        self.mox.ReplayAll()
235        self.assertTrue(self.task.Run(self.sched, self._MAP, self._BOARD))
236
237
238    def testRunCustomSharding(self):
239        """Test running a recurring task with non-default sharding."""
240        expected_sharding = 2
241        mytask = task.Task(self._TASK_NAME, self._SUITE, [self._BRANCH_SPEC],
242                           num=expected_sharding)
243        self.sched.ScheduleSuite(self._SUITE, self._BOARD, self._BUILD,
244                                 None, expected_sharding, None, None,
245                                 False, file_bugs=self._FILE_BUGS,
246                                 firmware_rw_build=None,
247                                 firmware_ro_build=None,
248                                 test_source_build=None,
249                                 job_retry=False,
250                                 launch_control_build=None,
251                                 run_prod_code=False,
252                                 testbed_dut_count=None,
253                                 no_delay=False).AndReturn(True)
254        self.mox.ReplayAll()
255        self.assertTrue(mytask.Run(self.sched, self._MAP, self._BOARD))
256
257
258    def testRunDuplicate(self):
259        """Test running a task that schedules a duplicate suite task."""
260        self.sched.ScheduleSuite(self._SUITE, self._BOARD, self._BUILD,
261                                 None, None, self._PRIORITY, self._TIMEOUT,
262                                 False, file_bugs=self._FILE_BUGS,
263                                 firmware_rw_build=None,
264                                 firmware_ro_build=None,
265                                 test_source_build=None,
266                                 job_retry=False,
267                                 launch_control_build=None,
268                                 run_prod_code=False,
269                                 testbed_dut_count=None,
270                                 no_delay=False).AndReturn(True)
271        self.mox.ReplayAll()
272        self.assertTrue(self.task.Run(self.sched, self._MAP, self._BOARD))
273
274
275    def testRunUnrunnablePool(self):
276        """Test running a task that cannot run on this pool."""
277        self.sched.CheckHostsExist(
278                multiple_labels=mox.IgnoreArg()).AndReturn(None)
279        self.mox.ReplayAll()
280        t = task.Task(self._TASK_NAME, self._SUITE,
281                      [self._BRANCH_SPEC], "BadPool")
282        self.assertTrue(not t.AvailableHosts(self.sched, self._BOARD))
283
284
285    def testRunUnrunnableBoard(self):
286        """Test running a task that cannot run on this board."""
287        self.mox.ReplayAll()
288        t = task.Task(self._TASK_NAME, self._SUITE,
289                      [self._BRANCH_SPEC], self._POOL, boards="BadBoard")
290        self.assertTrue(not t.AvailableHosts(self.sched, self._BOARD))
291
292
293    def testNoRunBranchMismatch(self):
294        """Test running a recurring task with no matching builds."""
295        t = task.Task(self._TASK_NAME, self._SUITE, task.BARE_BRANCHES)
296        self.mox.ReplayAll()
297        self.assertTrue(t.Run(self.sched, self._MAP, self._BOARD))
298
299
300    def testNoRunBareBranchMismatch(self):
301        """Test running a recurring task with no matching builds (factory)."""
302        self.mox.ReplayAll()
303        self.assertTrue(
304            self.task.Run(self.sched, {'factory': 'build2'}, self._BOARD))
305
306
307    def testRunNoSpec(self):
308        """Test running a recurring task with default branch specs."""
309        t = task.Task(self._TASK_NAME, self._SUITE, [])
310        self.sched.ScheduleSuite(self._SUITE, self._BOARD, self._BUILD,
311                                 None, None, None, None,
312                                 False, file_bugs=self._FILE_BUGS,
313                                 firmware_rw_build=None,
314                                 firmware_ro_build=None,
315                                 test_source_build=None,
316                                 job_retry=False,
317                                 launch_control_build=None,
318                                 run_prod_code=False,
319                                 testbed_dut_count=None,
320                                 no_delay=False).AndReturn(True)
321        self.mox.ReplayAll()
322        self.assertTrue(t.Run(self.sched, self._MAP, self._BOARD))
323
324
325    def testRunExplodes(self):
326        """Test a failure to schedule while running task."""
327        # Barf while scheduling.
328        self.sched.ScheduleSuite(
329                self._SUITE, self._BOARD, self._BUILD, None, None,
330                self._PRIORITY, self._TIMEOUT, False, file_bugs=self._FILE_BUGS,
331                firmware_rw_build=None, firmware_ro_build=None,
332                test_source_build=None, job_retry=False,
333                launch_control_build=None, run_prod_code=False,
334                testbed_dut_count=None, no_delay=False).AndRaise(
335                        deduping_scheduler.ScheduleException(
336                                'Simulated Failure'))
337        self.mox.ReplayAll()
338        self.assertTrue(self.task.Run(self.sched, self._MAP, self._BOARD))
339
340
341    def testForceRun(self):
342        """Test force running a recurring task."""
343        self.sched.ScheduleSuite(self._SUITE, self._BOARD, self._BUILD,
344                                 None, None, self._PRIORITY, self._TIMEOUT,
345                                 True, file_bugs=self._FILE_BUGS,
346                                 firmware_rw_build=None,
347                                 firmware_ro_build=None,
348                                 test_source_build=None,
349                                 job_retry=False,
350                                 launch_control_build=None,
351                                 run_prod_code=False,
352                                 testbed_dut_count=None,
353                                 no_delay=False).AndReturn(True)
354        self.mox.ReplayAll()
355        self.assertTrue(self.task.Run(self.sched, self._MAP, self._BOARD, True))
356
357
358    def testHash(self):
359        """Test hash function for Task classes."""
360        same_task = task.Task(self._TASK_NAME, self._SUITE, [self._BRANCH_SPEC],
361                              boards=self._BOARD)
362        other_task = task.Task(self._TASK_NAME, self._SUITE,
363                               [self._BRANCH_SPEC, '>=RX1'], 'pool')
364        self.assertEquals(hash(self.task), hash(same_task))
365        self.assertNotEquals(hash(self.task), hash(other_task))
366
367
368class OneShotTaskTest(TaskTestBase):
369    """Unit tests for OneShotTask."""
370
371
372    def setUp(self):
373        super(OneShotTaskTest, self).setUp()
374        self.task = task.OneShotTask(self._TASK_NAME, self._SUITE,
375                                     [self._BRANCH_SPEC])
376
377
378    def testRun(self):
379        """Test running a one-shot task."""
380        self.sched.ScheduleSuite(self._SUITE, self._BOARD, self._BUILD,
381                                 None, None, None, None, False,
382                                 file_bugs=self._FILE_BUGS,
383                                 firmware_rw_build=None,
384                                 firmware_ro_build=None,
385                                 test_source_build=None,
386                                 job_retry=False,
387                                 launch_control_build=None,
388                                 run_prod_code=False,
389                                 testbed_dut_count=None,
390                                 no_delay=False).AndReturn(True)
391        self.mox.ReplayAll()
392        self.assertFalse(self.task.Run(self.sched, self._MAP, self._BOARD))
393
394
395    def testRunDuplicate(self):
396        """Test running a one-shot task that schedules a dup suite task."""
397        self.sched.ScheduleSuite(self._SUITE, self._BOARD, self._BUILD,
398                                 None, None, None, None, False,
399                                 file_bugs=self._FILE_BUGS,
400                                 firmware_rw_build=None,
401                                 firmware_ro_build=None,
402                                 test_source_build=None,
403                                 job_retry=False,
404                                 launch_control_build=None,
405                                 run_prod_code=False,
406                                 testbed_dut_count=None,
407                                 no_delay=False).AndReturn(False)
408        self.mox.ReplayAll()
409        self.assertFalse(self.task.Run(self.sched, self._MAP, self._BOARD))
410
411
412    def testRunExplodes(self):
413        """Test a failure to schedule while running one-shot task."""
414        # Barf while scheduling.
415        self.sched.ScheduleSuite(
416                self._SUITE, self._BOARD, self._BUILD, None, None,
417                None, None, False, file_bugs=self._FILE_BUGS,
418                firmware_rw_build=None, firmware_ro_build=None,
419                test_source_build=None, job_retry=False,
420                launch_control_build=None, run_prod_code=False,
421                testbed_dut_count=None, no_delay=False).AndRaise(
422                        deduping_scheduler.ScheduleException(
423                                'Simulated Failure'))
424        self.mox.ReplayAll()
425        self.assertFalse(self.task.Run(self.sched, self._MAP, self._BOARD))
426
427
428    def testForceRun(self):
429        """Test force running a one-shot task."""
430        self.sched.ScheduleSuite(self._SUITE, self._BOARD, self._BUILD,
431                                 None, None, None, None, True,
432                                 file_bugs=self._FILE_BUGS,
433                                 firmware_rw_build=None,
434                                 firmware_ro_build=None,
435                                 test_source_build=None,
436                                 job_retry=False,
437                                 launch_control_build=None,
438                                 run_prod_code=False,
439                                 testbed_dut_count=None,
440                                 no_delay=False).AndReturn(True)
441        self.mox.ReplayAll()
442        self.assertFalse(self.task.Run(self.sched, self._MAP, self._BOARD,
443                                       force=True))
444
445
446    def testFileBugs(self):
447        """Test that file_bugs is passed from the task to ScheduleSuite."""
448        self.sched.ScheduleSuite(self._SUITE, self._BOARD, self._BUILD,
449                                 None, None, None, None, True,
450                                 file_bugs=True, firmware_rw_build=None,
451                                 firmware_ro_build=None,
452                                 test_source_build=None,
453                                 job_retry=False,
454                                 launch_control_build=None,
455                                 run_prod_code=False,
456                                 testbed_dut_count=None,
457                                 no_delay=False).AndReturn(True)
458        self.mox.ReplayAll()
459        self.task._file_bugs = True
460        self.assertFalse(self.task.Run(self.sched, self._MAP, self._BOARD,
461                                       force=True))
462
463
464if __name__ == '__main__':
465    unittest.main()
466