| # Copyright 2018 The Chromium OS Authors. All rights reserved. |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| """Module for task config reader unittests.""" |
| # pylint: disable=g-missing-super-call |
| |
| import unittest |
| |
| import config_reader |
| import constants |
| import mock |
| import task_config_reader |
| import tot_manager |
| |
| |
| class FakeTotMilestoneManager(tot_manager.TotMilestoneManager): |
| |
| def __init__(self, is_sanity): |
| self.is_sanity = is_sanity |
| self.storage_client = None |
| self.tot = self._tot_milestone() |
| |
| |
| class BaseTaskConfigReaderTestCase(unittest.TestCase): |
| |
| _EVENT_TYPE = 'nightly' |
| _PRIORITY = task_config_reader.EVENT_CLASSES[_EVENT_TYPE].PRIORITY |
| _TIMEOUT = task_config_reader.EVENT_CLASSES[_EVENT_TYPE].TIMEOUT |
| _TASK_NAME = 'fake_task' |
| _SUITE = 'fake_suite' |
| _BRANCH_SPEC = 'tot-1' |
| _POOL = 'bvt' |
| _NUM = 2 |
| _BOARD = 'link' |
| |
| def setUp(self): |
| self.config = config_reader.ConfigReader(None) |
| self.config.add_section(self._TASK_NAME) |
| self.config.set(self._TASK_NAME, 'suite', self._SUITE) |
| self.config.set(self._TASK_NAME, 'branch_specs', self._BRANCH_SPEC) |
| self.config.set(self._TASK_NAME, 'run_on', self._EVENT_TYPE) |
| self.config.set(self._TASK_NAME, 'pool', self._POOL) |
| self.config.set(self._TASK_NAME, 'num', '%d' % self._NUM) |
| self.config.set(self._TASK_NAME, 'boards', self._BOARD) |
| |
| tot_patcher = mock.patch('tot_manager.TotMilestoneManager') |
| self._tot = tot_patcher.start() |
| self._tot.return_value = FakeTotMilestoneManager(True) |
| |
| board_family_patcher = mock.patch( |
| 'build_lib.get_board_family_mapping_from_gs') |
| board_family_getter = board_family_patcher.start() |
| board_family_getter.return_value = { |
| 'nyan': ['nyan', 'nyan_blaze', 'nyan_big'], |
| 'ivybridge': ['link', 'link_freon']} |
| |
| self.addCleanup(tot_patcher.stop) |
| self.addCleanup(board_family_patcher.stop) |
| |
| |
| class TaskCreationTestCase(BaseTaskConfigReaderTestCase): |
| |
| def testCreateTaskFromConfigNonexistentSection(self): |
| """Ensure a Task CANNOT be built without suite, and raise error.""" |
| task_config = task_config_reader.TaskConfig(self.config) |
| self.assertRaises(task_config_reader.MalformedConfigEntryError, |
| task_config._create_task_from_config_section, |
| 'non-existent section') |
| |
| def testCreateTaskFromConfigNotAllowedHeader(self): |
| """Ensure a Task CANNOT be built without suite, and raise error.""" |
| self.config.set(self._TASK_NAME, 'non-valid-header', 'test') |
| task_config = task_config_reader.TaskConfig(self.config) |
| self.assertRaises(task_config_reader.MalformedConfigEntryError, |
| task_config._create_task_from_config_section, |
| self._TASK_NAME) |
| |
| def testGetTaskByKeyword(self): |
| """Ensure a Task can be built from a correct config.""" |
| task_config = task_config_reader.TaskConfig(self.config) |
| tasks = task_config.get_tasks_by_keyword(self._EVENT_TYPE)['tasks'] |
| self.assertEqual(len(tasks), 1) |
| new_task = tasks[0] |
| self.assertEqual(new_task.name, self._TASK_NAME) |
| self.assertEqual(new_task.suite, self._SUITE) |
| self.assertEqual(new_task.branch_specs, [self._BRANCH_SPEC]) |
| self.assertEqual(new_task.pool, self._POOL) |
| self.assertEqual(new_task.num, self._NUM) |
| self.assertEqual(new_task.boards, |
| [t.lstrip() for t in self._BOARD.split(',')]) |
| self.assertEqual(new_task.priority, self._PRIORITY) |
| self.assertEqual(new_task.timeout, self._TIMEOUT) |
| |
| def testGetTaskByKeywordNoRunonNoRaise(self): |
| """Ensure a Task CANNOT be built without run_on, but not raise error.""" |
| self.config.remove_option(self._TASK_NAME, 'run_on') |
| task_config = task_config_reader.TaskConfig(self.config) |
| try: |
| response = task_config.get_tasks_by_keyword(self._EVENT_TYPE) |
| self.assertEqual(len(response['tasks']), 0) |
| self.assertEqual(len(response['exceptions']), 1) |
| except task_config_reader.MalformedConfigEntryError: |
| self.fail('task_config should not raise error without run_on!') |
| |
| def testGetTaskByKeywordNoSuiteNoRaise(self): |
| """Ensure a Task CANNOT be built without suite, but not raise error. |
| |
| This is to ensure a malformed task won't affect other tasks' reading. |
| """ |
| self.config.remove_option(self._TASK_NAME, 'suite') |
| task_config = task_config_reader.TaskConfig(self.config) |
| try: |
| response = task_config.get_tasks_by_keyword(self._EVENT_TYPE) |
| self.assertEqual(len(response['tasks']), 0) |
| self.assertEqual(len(response['exceptions']), 1) |
| except config_reader.MalformedConfigEntry: |
| self.fail('task_config should not raise error without suite!') |
| |
| def testGetTaskByKeywordNoBranch(self): |
| """Ensure a Task can be built without branch_specs.""" |
| self.config.remove_option(self._TASK_NAME, 'branch_specs') |
| task_config = task_config_reader.TaskConfig(self.config) |
| tasks = task_config.get_tasks_by_keyword(self._EVENT_TYPE)['tasks'] |
| self.assertEqual(tasks[0].branch_specs, []) |
| |
| def testGetTaskByKeywordNoNum(self): |
| """Ensure a Task can be built without num.""" |
| self.config.remove_option(self._TASK_NAME, 'num') |
| task_config = task_config_reader.TaskConfig(self.config) |
| tasks = task_config.get_tasks_by_keyword(self._EVENT_TYPE)['tasks'] |
| self.assertIsNone(tasks[0].num) |
| |
| def testSetPriority(self): |
| """Ensure priority can be set by user.""" |
| priority = constants.Priorities.DEFAULT |
| self.config.set(self._TASK_NAME, 'priority', str(priority)) |
| task_config = task_config_reader.TaskConfig(self.config) |
| tasks = task_config.get_tasks_by_keyword(self._EVENT_TYPE)['tasks'] |
| self.assertEqual(tasks[0].priority, priority) |
| |
| def testSetTimeout(self): |
| """Ensure timeout can be set by user.""" |
| timeout = 1000 |
| self.config.set(self._TASK_NAME, 'timeout', str(timeout)) |
| task_config = task_config_reader.TaskConfig(self.config) |
| tasks = task_config.get_tasks_by_keyword(self._EVENT_TYPE)['tasks'] |
| self.assertEqual(tasks[0].timeout, timeout) |
| |
| |
| class TaskInfoCheckTestCase(BaseTaskConfigReaderTestCase): |
| |
| def testNonIntNum(self): |
| """Ensure a Task's num is a Integer.""" |
| self.config.set(self._TASK_NAME, 'num', 'non_int') |
| task_config = task_config_reader.TaskConfig(self.config) |
| self.assertRaises(task_config_reader.MalformedConfigEntryError, |
| task_config._read_task_section, |
| self._TASK_NAME) |
| |
| def testNonIntHour(self): |
| """Ensure a Task's hour is a Integer.""" |
| self.config.set(self._TASK_NAME, 'hour', 'non_int') |
| task_config = task_config_reader.TaskConfig(self.config) |
| self.assertRaises(task_config_reader.MalformedConfigEntryError, |
| task_config._read_task_section, |
| self._TASK_NAME) |
| |
| def testNonIntDay(self): |
| """Ensure a Task's hour is a Integer.""" |
| self.config.set(self._TASK_NAME, 'day', 'non_int') |
| task_config = task_config_reader.TaskConfig(self.config) |
| self.assertRaises(task_config_reader.MalformedConfigEntryError, |
| task_config._read_task_section, |
| self._TASK_NAME) |
| |
| def testNoSuite(self): |
| """Ensure a Task's suite exists.""" |
| self.config.remove_option(self._TASK_NAME, 'suite') |
| task_config = task_config_reader.TaskConfig(self.config) |
| task_info = task_config._read_task_section(self._TASK_NAME) |
| self.assertRaises(task_config_reader.MalformedConfigEntryError, |
| task_config._validate_task_section, |
| task_info) |
| |
| def testInvalidIntHour(self): |
| """Ensure a Task's hour is in 0~23.""" |
| self.config.set(self._TASK_NAME, 'hour', '24') |
| task_config = task_config_reader.TaskConfig(self.config) |
| task_info = task_config._read_task_section(self._TASK_NAME) |
| self.assertRaises(task_config_reader.MalformedConfigEntryError, |
| task_config._validate_task_section, |
| task_info) |
| |
| def testNoHourForNonNightlyEvent(self): |
| """Ensure a non-nightly Task cannot specify hour.""" |
| self.config.set(self._TASK_NAME, 'hour', '10') |
| self.config.set(self._TASK_NAME, 'run_on', 'new_build') |
| task_config = task_config_reader.TaskConfig(self.config) |
| task_info = task_config._read_task_section(self._TASK_NAME) |
| self.assertRaises(task_config_reader.MalformedConfigEntryError, |
| task_config._validate_task_section, |
| task_info) |
| |
| def testInvalidIntDay(self): |
| """Ensure a Task's day is in 0~6.""" |
| self.config.set(self._TASK_NAME, 'day', '10') |
| self.config.set(self._TASK_NAME, 'run_on', 'weekly') |
| task_config = task_config_reader.TaskConfig(self.config) |
| task_info = task_config._read_task_section(self._TASK_NAME) |
| self.assertRaises(task_config_reader.MalformedConfigEntryError, |
| task_config._validate_task_section, |
| task_info) |
| |
| def testNoDayForNonWeeklyEvent(self): |
| """Ensure a non-weekly Task cannot specify day.""" |
| self.config.set(self._TASK_NAME, 'day', '1') |
| self.config.set(self._TASK_NAME, 'run_on', 'nightly') |
| task_config = task_config_reader.TaskConfig(self.config) |
| task_info = task_config._read_task_section(self._TASK_NAME) |
| self.assertRaises(task_config_reader.MalformedConfigEntryError, |
| task_config._validate_task_section, |
| task_info) |
| |
| def testInValidOSType(self): |
| """Ensure a Task's os_type is valid.""" |
| self.config.set(self._TASK_NAME, 'os_type', 'invalid') |
| task_config = task_config_reader.TaskConfig(self.config) |
| task_info = task_config._read_task_section(self._TASK_NAME) |
| self.assertRaises(task_config_reader.MalformedConfigEntryError, |
| task_config._check_os_type, |
| task_info) |
| |
| def testNoLaunchControlForCrOS(self): |
| """Ensure a CrOS Task doesn't have launch control settings.""" |
| self.config.set(self._TASK_NAME, 'os_type', 'CrOS') |
| self.config.set(self._TASK_NAME, 'branches', 'shamu-userdebug') |
| task_config = task_config_reader.TaskConfig(self.config) |
| task_info = task_config._read_task_section(self._TASK_NAME) |
| self.assertRaises(task_config_reader.MalformedConfigEntryError, |
| task_config._check_os_type, |
| task_info) |
| |
| def testLaunchControlForNonCrOS(self): |
| """Ensure an android Task has launch control settings.""" |
| self.config.set(self._TASK_NAME, 'os_type', 'android') |
| task_config = task_config_reader.TaskConfig(self.config) |
| task_info = task_config._read_task_section(self._TASK_NAME) |
| self.assertRaises(task_config_reader.MalformedConfigEntryError, |
| task_config._check_os_type, |
| task_info) |
| |
| def testGetTestbedCount(self): |
| """Ensure testbed count can be successfully parsed.""" |
| task_config = task_config_reader.TaskConfig(self.config) |
| self.assertEqual(task_config._get_testbed_dut_count('sharu-2'), 2) |
| |
| def testNoTestbedForCrOS(self): |
| """Ensure CrOS task won't specify testbed parameters.""" |
| self.config.set(self._TASK_NAME, 'boards', 'sharu-2') |
| task_config = task_config_reader.TaskConfig(self.config) |
| task_info = task_config._read_task_section(self._TASK_NAME) |
| self.assertRaises(task_config_reader.MalformedConfigEntryError, |
| task_config._check_testbed_parameter, |
| task_info) |
| |
| def testInvalidBoardFamilies(self): |
| """Ensure to raise exception when a board family is invalid.""" |
| self.config.set(self._TASK_NAME, 'board_families', 'fake_one') |
| task_config = task_config_reader.TaskConfig(self.config) |
| self.assertRaises(task_config_reader.MalformedConfigEntryError, |
| task_config._read_task_section, |
| self._TASK_NAME) |
| |
| def testInvalidExcludeBoardFamilies(self): |
| """Ensure to raise exception when an exclude board family is invalid.""" |
| self.config.set(self._TASK_NAME, 'exclude_board_families', 'fake_one') |
| task_config = task_config_reader.TaskConfig(self.config) |
| self.assertRaises(task_config_reader.MalformedConfigEntryError, |
| task_config._read_task_section, |
| self._TASK_NAME) |
