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