tko/parsers/version_0_unittest.py

   1 #!/usr/bin/python
   2
   3 import unittest
   4
   5 import common
   6 from autotest_lib.client.common_lib.test_utils import mock
   7 from autotest_lib.tko import models
   8 from autotest_lib.tko.parsers import version_0
   9
  10
  11 class test_job_load_from_dir(unittest.TestCase):
  12     keyval_return = {'job_queued': 1234567890,
  13                      'job_started': 1234567891,
  14                      'job_finished': 1234567892,
  15                      'user': 'janet',
  16                      'label': 'steeltown',
  17                      'hostname': 'abc123'}
  18
  19
  20     def setUp(self):
  21         self.god = mock.mock_god()
  22         self.god.stub_function(models.job, 'read_keyval')
  23         self.god.stub_function(version_0.job, 'find_hostname')
  24         self.god.stub_function(models.test, 'parse_host_keyval')
  25
  26
  27     def tearDown(self):
  28         self.god.unstub_all()
  29
  30
  31     def _expect_host_keyval(self, hostname, platform=None):
  32         return_dict = {}
  33         if platform:
  34             return_dict['platform'] = platform
  35             return_dict['labels'] = platform + ',other_label'
  36         (models.test.parse_host_keyval.expect_call('.', hostname)
  37                 .and_return(return_dict))
  38
  39
  40     def test_load_from_dir_simple(self):
  41         models.job.read_keyval.expect_call('.').and_return(
  42                 dict(self.keyval_return))
  43         self._expect_host_keyval('abc123', 'my_platform')
  44         job = version_0.job.load_from_dir('.')
  45         self.assertEqual('janet', job['user'])
  46         self.assertEqual('steeltown', job['label'])
  47         self.assertEqual('abc123', job['machine'])
  48         self.assertEqual('my_platform', job['machine_group'])
  49         self.god.check_playback()
  50
  51
  52     def _setup_two_machines(self):
  53         raw_keyval = dict(self.keyval_return)
  54         raw_keyval['hostname'] = 'easyas,abc123'
  55         models.job.read_keyval.expect_call('.').and_return(raw_keyval)
  56
  57
  58     def test_load_from_dir_two_machines(self):
  59         self._setup_two_machines()
  60         version_0.job.find_hostname.expect_call('.').and_raises(
  61                     version_0.NoHostnameError('find_hostname stubbed out'))
  62         self._expect_host_keyval('easyas', 'platform')
  63         self._expect_host_keyval('abc123', 'platform')
  64
  65         job = version_0.job.load_from_dir('.')
  66         self.assertEqual('easyas,abc123', job['machine'])
  67         self.assertEqual('platform', job['machine_group'])
  68
  69         self.god.check_playback()
  70
  71
  72     def test_load_from_dir_two_machines_with_find_hostname(self):
  73         self._setup_two_machines()
  74         version_0.job.find_hostname.expect_call('.').and_return('foo')
  75         self._expect_host_keyval('foo')
  76
  77         job = version_0.job.load_from_dir('.')
  78         self.assertEqual('foo', job['machine'])
  79
  80         self.god.check_playback()
  81
  82
  83     def test_load_from_dir_two_machines_different_platforms(self):
  84         self._setup_two_machines()
  85         version_0.job.find_hostname.expect_call('.').and_raises(
  86                     version_0.NoHostnameError('find_hostname stubbed out'))
  87         self._expect_host_keyval('easyas', 'platformZ')
  88         self._expect_host_keyval('abc123', 'platformA')
  89
  90         job = version_0.job.load_from_dir('.')
  91         self.assertEqual('easyas,abc123', job['machine'])
  92         self.assertEqual('platformA,platformZ', job['machine_group'])
  93
  94         self.god.check_playback()
  95
  96     def test_load_from_dir_one_machine_group_name(self):
  97         raw_keyval = dict(self.keyval_return)
  98         raw_keyval['host_group_name'] = 'jackson five'
  99         models.job.read_keyval.expect_call('.').and_return(raw_keyval)
 100         self._expect_host_keyval('abc123')
 101         job = version_0.job.load_from_dir('.')
 102         self.assertEqual('janet', job['user'])
 103         self.assertEqual('abc123', job['machine'])
 104         self.god.check_playback()
 105
 106
 107     def test_load_from_dir_multi_machine_group_name(self):
 108         raw_keyval = dict(self.keyval_return)
 109         raw_keyval['user'] = 'michael'
 110         raw_keyval['hostname'] = 'abc123,dancingmachine'
 111         raw_keyval['host_group_name'] = 'jackson five'
 112         models.job.read_keyval.expect_call('.').and_return(raw_keyval)
 113         self._expect_host_keyval('jackson five')
 114         job = version_0.job.load_from_dir('.')
 115         self.assertEqual('michael', job['user'])
 116         # The host_group_name is used instead because machine appeared to be
 117         # a comma separated list.
 118         self.assertEqual('jackson five', job['machine'])
 119         self.god.check_playback()
 120
 121
 122     def test_load_from_dir_no_machine_group_name(self):
 123         raw_keyval = dict(self.keyval_return)
 124         del raw_keyval['hostname']
 125         raw_keyval['host_group_name'] = 'jackson five'
 126         models.job.read_keyval.expect_call('.').and_return(raw_keyval)
 127         self._expect_host_keyval('jackson five')
 128         job = version_0.job.load_from_dir('.')
 129         # The host_group_name is used because there is no machine.
 130         self.assertEqual('jackson five', job['machine'])
 131         self.god.check_playback()
 132
 133
 134 class test_status_line(unittest.TestCase):
 135     statuses = ["GOOD", "WARN", "FAIL", "ABORT"]
 136
 137
 138     def test_handles_start(self):
 139         line = version_0.status_line(0, "START", "----", "test",
 140                                      "", {})
 141         self.assertEquals(line.type, "START")
 142         self.assertEquals(line.status, None)
 143
 144
 145     def test_fails_info(self):
 146         self.assertRaises(AssertionError,
 147                           version_0.status_line, 0, "INFO", "----", "----",
 148                           "", {})
 149
 150
 151     def test_handles_status(self):
 152         for stat in self.statuses:
 153             line = version_0.status_line(0, stat, "----", "test",
 154                                          "", {})
 155             self.assertEquals(line.type, "STATUS")
 156             self.assertEquals(line.status, stat)
 157
 158
 159     def test_handles_endstatus(self):
 160         for stat in self.statuses:
 161             line = version_0.status_line(0, "END " + stat, "----",
 162                                          "test", "", {})
 163             self.assertEquals(line.type, "END")
 164             self.assertEquals(line.status, stat)
 165
 166
 167     def test_fails_on_bad_status(self):
 168         for stat in self.statuses:
 169             self.assertRaises(AssertionError,
 170                               version_0.status_line, 0,
 171                               "BAD " + stat, "----", "test",
 172                               "", {})
 173
 174
 175     def test_saves_all_fields(self):
 176         line = version_0.status_line(5, "GOOD", "subdir_name",
 177                                      "test_name", "my reason here",
 178                                      {"key1": "value",
 179                                       "key2": "another value",
 180                                       "key3": "value3"})
 181         self.assertEquals(line.indent, 5)
 182         self.assertEquals(line.status, "GOOD")
 183         self.assertEquals(line.subdir, "subdir_name")
 184         self.assertEquals(line.testname, "test_name")
 185         self.assertEquals(line.reason, "my reason here")
 186         self.assertEquals(line.optional_fields,
 187                           {"key1": "value", "key2": "another value",
 188                            "key3": "value3"})
 189
 190
 191     def test_parses_blank_subdir(self):
 192         line = version_0.status_line(0, "GOOD", "----", "test",
 193                                      "", {})
 194         self.assertEquals(line.subdir, None)
 195
 196
 197     def test_parses_blank_testname(self):
 198         line = version_0.status_line(0, "GOOD", "subdir", "----",
 199                                      "", {})
 200         self.assertEquals(line.testname, None)
 201
 202
 203     def test_parse_line_smoketest(self):
 204         input_data = ("\t\t\tGOOD\t----\t----\t"
 205                       "field1=val1\tfield2=val2\tTest Passed")
 206         line = version_0.status_line.parse_line(input_data)
 207         self.assertEquals(line.indent, 3)
 208         self.assertEquals(line.type, "STATUS")
 209         self.assertEquals(line.status, "GOOD")
 210         self.assertEquals(line.subdir, None)
 211         self.assertEquals(line.testname, None)
 212         self.assertEquals(line.reason, "Test Passed")
 213         self.assertEquals(line.optional_fields,
 214                           {"field1": "val1", "field2": "val2"})
 215
 216     def test_parse_line_handles_newline(self):
 217         input_data = ("\t\tGOOD\t----\t----\t"
 218                       "field1=val1\tfield2=val2\tNo newline here!")
 219         for suffix in ("", "\n"):
 220             line = version_0.status_line.parse_line(input_data +
 221                                                     suffix)
 222             self.assertEquals(line.indent, 2)
 223             self.assertEquals(line.type, "STATUS")
 224             self.assertEquals(line.status, "GOOD")
 225             self.assertEquals(line.subdir, None)
 226             self.assertEquals(line.testname, None)
 227             self.assertEquals(line.reason, "No newline here!")
 228             self.assertEquals(line.optional_fields,
 229                               {"field1": "val1",
 230                                "field2": "val2"})
 231
 232
 233     def test_parse_line_handles_embedded_new_lines(self):
 234         input_data = ("\tEND FAIL\t----\ttest\tfield1=val1\tStatus\nwith\n"
 235                       "embedded\nnew lines\n")
 236
 237         line = version_0.status_line.parse_line(input_data)
 238         self.assertEquals(line.indent, 1)
 239         self.assertEquals(line.type, "END")
 240         self.assertEquals(line.status, "FAIL")
 241         self.assertEquals(line.subdir, None)
 242         self.assertEquals(line.testname, "test")
 243         self.assertEquals(line.reason, "Status\nwith\nembedded\nnew lines")
 244         self.assertEquals(line.optional_fields, {"field1": "val1"})
 245
 246
 247     def test_parse_line_fails_on_untabbed_lines(self):
 248         input_data = "   GOOD\trandom\tfields\tof text"
 249         line = version_0.status_line.parse_line(input_data)
 250         self.assertEquals(line, None)
 251         line = version_0.status_line.parse_line(input_data.lstrip())
 252         self.assertEquals(line.indent, 0)
 253         self.assertEquals(line.type, "STATUS")
 254         self.assertEquals(line.status, "GOOD")
 255         self.assertEquals(line.subdir, "random")
 256         self.assertEquals(line.testname, "fields")
 257         self.assertEquals(line.reason, "of text")
 258         self.assertEquals(line.optional_fields, {})
 259
 260
 261     def test_parse_line_fails_on_incomplete_lines(self):
 262         input_data = "\t\tGOOD\tfield\tsecond field"
 263         complete_data = input_data + "\tneeded last field"
 264         line = version_0.status_line.parse_line(input_data)
 265         self.assertEquals(line, None)
 266         line = version_0.status_line.parse_line(complete_data)
 267         self.assertEquals(line.indent, 2)
 268         self.assertEquals(line.type, "STATUS")
 269         self.assertEquals(line.status, "GOOD")
 270         self.assertEquals(line.subdir, "field")
 271         self.assertEquals(line.testname, "second field")
 272         self.assertEquals(line.reason, "needed last field")
 273         self.assertEquals(line.optional_fields, {})
 274
 275
 276     def test_parse_line_handles_tabs_in_reason(self):
 277         input_data = ("\tEND FAIL\t----\ttest\tfield1=val1\tfield2=val2\tReason"
 278                       " with\ta\tcouple\ttabs")
 279
 280         line = version_0.status_line.parse_line(input_data)
 281         self.assertEquals(line.indent, 1)
 282         self.assertEquals(line.type, "END")
 283         self.assertEquals(line.status, "FAIL")
 284         self.assertEquals(line.subdir, None)
 285         self.assertEquals(line.testname, "test")
 286         self.assertEquals(line.reason, "Reason with\ta\tcouple\ttabs")
 287         self.assertEquals(line.optional_fields, {"field1": "val1",
 288                                                  "field2": "val2"})
 289
 290
 291 if __name__ == "__main__":
 292     unittest.main()