mirror/trigger_unittest.py

   1 #!/usr/bin/python
   2 # Copyright 2009 Google Inc. Released under the GPL v2
   3
   4 import unittest
   5
   6 import common
   7 from autotest_lib.mirror import trigger
   8 from autotest_lib.client.common_lib.test_utils import mock
   9
  10
  11 class map_action_unittest(unittest.TestCase):
  12     def setUp(self):
  13         self.god = mock.mock_god()
  14
  15
  16     def tearDown(self):
  17         pass
  18
  19
  20     def test_machine_info_api(self):
  21         tests = object()
  22         configs = object()
  23
  24         info = trigger.map_action.machine_info(tests, configs)
  25         self.assertEquals(tests, info.tests)
  26         self.assertEquals(configs, info.kernel_configs)
  27
  28
  29     @staticmethod
  30     def _make_control_dict(contents, is_server=False, synch_count=1,
  31                            dependencies=()):
  32         class ControlFile(object):
  33             def __init__(self, contents, is_server, synch_count, dependencies):
  34                 self.control_file = contents
  35                 self.is_server = is_server
  36                 self.synch_count = synch_count
  37                 self.dependencies = dependencies
  38
  39         return ControlFile(contents, is_server, synch_count, dependencies)
  40
  41
  42     def test_job_grouping(self):
  43         tests_map = {
  44             'mach1': trigger.map_action.machine_info(
  45                     ('test1', 'test2'), {'2.6.20': 'config1'}),
  46             'mach2': trigger.map_action.machine_info(
  47                     ('test3',), {'2.6.10': 'config2', '2.6.20': 'config1'}),
  48             'mach3': trigger.map_action.machine_info(
  49                     ('test2', 'test3'), {'2.6.20': 'config1'}),
  50             }
  51         action = trigger.map_action(tests_map, 'jobname %s')
  52         self.assertTrue(isinstance(action._afe, trigger.frontend.AFE))
  53         action._afe = self.god.create_mock_class(trigger.frontend.AFE, 'AFE')
  54
  55         control2 = self._make_control_dict('control contents2')
  56         (action._afe.generate_control_file.expect_call(
  57                 tests=['test2'],
  58                 kernel=[dict(version='2.6.21', config_file='config1')],
  59                 upload_kernel_config=False)
  60                 .and_return(control2))
  61         action._afe.create_job.expect_call(
  62                 control2.control_file, 'jobname 2.6.21',
  63                 control_type='Client', hosts=['mach1', 'mach3'])
  64
  65         control3 = self._make_control_dict('control contents3', is_server=True)
  66         (action._afe.generate_control_file.expect_call(
  67                 tests=['test3'],
  68                 kernel=[dict(version='2.6.21', config_file='config1')],
  69                 upload_kernel_config=False)
  70                 .and_return(control3))
  71         action._afe.create_job.expect_call(
  72                 control3.control_file, 'jobname 2.6.21',
  73                 control_type='Server', hosts=['mach2', 'mach3'])
  74
  75         control1 = self._make_control_dict('control contents1')
  76         (action._afe.generate_control_file.expect_call(
  77                 tests=['test1'],
  78                 kernel=[dict(version='2.6.21', config_file='config1')],
  79                 upload_kernel_config=False)
  80                 .and_return(control1))
  81         action._afe.create_job.expect_call(
  82                 control1.control_file, 'jobname 2.6.21',
  83                 control_type='Client', hosts=['mach1'])
  84
  85         action(['2.6.21'])
  86         self.god.check_playback()
  87
  88
  89     def test_kver_cmp(self):
  90         def check_cmp(ver1, ver2):
  91             # function to make sure "cmp" invariants are followed
  92             cmp_func = trigger.map_action._kver_cmp
  93             if ver1 != ver2:
  94                 self.assertEquals(cmp_func(ver1, ver2), -1)
  95                 self.assertEquals(cmp_func(ver2, ver1), 1)
  96             else:
  97                 self.assertEquals(cmp_func(ver1, ver2), 0)
  98                 self.assertEquals(cmp_func(ver2, ver1), 0)
  99
 100         check_cmp('2.6.20', '2.6.20')
 101         check_cmp('2.6.20', '2.6.21')
 102         check_cmp('2.6.20', '2.6.21-rc2')
 103         check_cmp('2.6.20-rc2-git2', '2.6.20-rc2')
 104
 105
 106     def test_upload_kernel_config(self):
 107         tests_map = {
 108             'mach1': trigger.map_action.machine_info(
 109                     ('test1',), {'2.6.20': 'config1'}),
 110             'mach3': trigger.map_action.machine_info(
 111                     ('test1',), {'2.6.20': 'config1'})
 112             }
 113
 114         action = trigger.map_action(tests_map, 'jobname %s',
 115                                     upload_kernel_config=True)
 116         self.assertTrue(isinstance(action._afe, trigger.frontend.AFE))
 117         action._afe = self.god.create_mock_class(trigger.frontend.AFE, 'AFE')
 118
 119         control = self._make_control_dict('control contents', is_server=True)
 120         (action._afe.generate_control_file.expect_call(
 121                 tests=['test1'],
 122                 kernel=[dict(version='2.6.21', config_file='config1')],
 123                 upload_kernel_config=True)
 124                 .and_return(control))
 125         action._afe.create_job.expect_call(
 126                 control.control_file, 'jobname 2.6.21',
 127                 control_type='Server', hosts=['mach1', 'mach3'])
 128
 129         action(['2.6.21'])
 130         self.god.check_playback()
 131
 132
 133 if __name__ == "__main__":
 134     unittest.main()