1#!/usr/bin/python3 2 3from autotest_lib.tko import models 4import datetime, time, unittest, mock 5 6import common 7from autotest_lib.client.common_lib import utils 8from autotest_lib.tko.parsers import version_1 9 10 11class test_status_line(unittest.TestCase): 12 """Tests for status lines.""" 13 14 statuses = ['GOOD', 'WARN', 'FAIL', 'ABORT'] 15 16 17 def test_handles_start(self): 18 """Tests that START is handled properly.""" 19 line = version_1.status_line(0, 'START', '----', 'test', 20 '', {}) 21 self.assertEquals(line.type, 'START') 22 self.assertEquals(line.status, None) 23 24 25 def test_handles_info(self): 26 """Tests that INFO is handled properly.""" 27 line = version_1.status_line(0, 'INFO', '----', '----', 28 '', {}) 29 self.assertEquals(line.type, 'INFO') 30 self.assertEquals(line.status, None) 31 32 33 def test_handles_status(self): 34 """Tests that STATUS is handled properly.""" 35 for stat in self.statuses: 36 line = version_1.status_line(0, stat, '----', 'test', 37 '', {}) 38 self.assertEquals(line.type, 'STATUS') 39 self.assertEquals(line.status, stat) 40 41 42 def test_handles_endstatus(self): 43 """Tests that END is handled properly.""" 44 for stat in self.statuses: 45 line = version_1.status_line(0, 'END ' + stat, '----', 46 'test', '', {}) 47 self.assertEquals(line.type, 'END') 48 self.assertEquals(line.status, stat) 49 50 51 def test_fails_on_bad_status(self): 52 """Tests that an exception is raised on a bad status.""" 53 for stat in self.statuses: 54 self.assertRaises(AssertionError, 55 version_1.status_line, 0, 56 'BAD ' + stat, '----', 'test', 57 '', {}) 58 59 60 def test_saves_all_fields(self): 61 """Tests that all fields are saved.""" 62 line = version_1.status_line(5, 'GOOD', 'subdir_name', 63 'test_name', 'my reason here', 64 {'key1': 'value', 65 'key2': 'another value', 66 'key3': 'value3'}) 67 self.assertEquals(line.indent, 5) 68 self.assertEquals(line.status, 'GOOD') 69 self.assertEquals(line.subdir, 'subdir_name') 70 self.assertEquals(line.testname, 'test_name') 71 self.assertEquals(line.reason, 'my reason here') 72 self.assertEquals(line.optional_fields, 73 {'key1': 'value', 'key2': 'another value', 74 'key3': 'value3'}) 75 76 77 def test_parses_blank_subdir(self): 78 """Tests that a blank subdirectory is parsed properly.""" 79 line = version_1.status_line(0, 'GOOD', '----', 'test', 80 '', {}) 81 self.assertEquals(line.subdir, None) 82 83 84 def test_parses_blank_testname(self): 85 """Tests that a blank test name is parsed properly.""" 86 line = version_1.status_line(0, 'GOOD', 'subdir', '----', 87 '', {}) 88 self.assertEquals(line.testname, None) 89 90 91 def test_parse_line_smoketest(self): 92 """Runs a parse line smoke test.""" 93 input_data = ('\t\t\tGOOD\t----\t----\t' 94 'field1=val1\tfield2=val2\tTest Passed') 95 line = version_1.status_line.parse_line(input_data) 96 self.assertEquals(line.indent, 3) 97 self.assertEquals(line.type, 'STATUS') 98 self.assertEquals(line.status, 'GOOD') 99 self.assertEquals(line.subdir, None) 100 self.assertEquals(line.testname, None) 101 self.assertEquals(line.reason, 'Test Passed') 102 self.assertEquals(line.optional_fields, 103 {'field1': 'val1', 'field2': 'val2'}) 104 105 def test_parse_line_handles_newline(self): 106 """Tests that newlines are handled properly.""" 107 input_data = ('\t\tGOOD\t----\t----\t' 108 'field1=val1\tfield2=val2\tNo newline here!') 109 for suffix in ('', '\n'): 110 line = version_1.status_line.parse_line(input_data + 111 suffix) 112 self.assertEquals(line.indent, 2) 113 self.assertEquals(line.type, 'STATUS') 114 self.assertEquals(line.status, 'GOOD') 115 self.assertEquals(line.subdir, None) 116 self.assertEquals(line.testname, None) 117 self.assertEquals(line.reason, 'No newline here!') 118 self.assertEquals(line.optional_fields, 119 {'field1': 'val1', 120 'field2': 'val2'}) 121 122 123 def test_parse_line_fails_on_untabbed_lines(self): 124 """Tests that untabbed lines do not parse.""" 125 input_data = ' GOOD\trandom\tfields\tof text' 126 line = version_1.status_line.parse_line(input_data) 127 self.assertEquals(line, None) 128 line = version_1.status_line.parse_line(input_data.lstrip()) 129 self.assertEquals(line.indent, 0) 130 self.assertEquals(line.type, 'STATUS') 131 self.assertEquals(line.status, 'GOOD') 132 self.assertEquals(line.subdir, 'random') 133 self.assertEquals(line.testname, 'fields') 134 self.assertEquals(line.reason, 'of text') 135 self.assertEquals(line.optional_fields, {}) 136 137 138 def test_parse_line_fails_on_incomplete_lines(self): 139 """Tests that incomplete lines do not parse.""" 140 input_data = '\t\tGOOD\tfield\tsecond field' 141 complete_data = input_data + '\tneeded last field' 142 line = version_1.status_line.parse_line(input_data) 143 self.assertEquals(line, None) 144 line = version_1.status_line.parse_line(complete_data) 145 self.assertEquals(line.indent, 2) 146 self.assertEquals(line.type, 'STATUS') 147 self.assertEquals(line.status, 'GOOD') 148 self.assertEquals(line.subdir, 'field') 149 self.assertEquals(line.testname, 'second field') 150 self.assertEquals(line.reason, 'needed last field') 151 self.assertEquals(line.optional_fields, {}) 152 153 154 def test_good_reboot_passes_success_test(self): 155 """Tests good reboot statuses.""" 156 line = version_1.status_line(0, 'NOSTATUS', None, 'reboot', 157 'reboot success', {}) 158 self.assertEquals(line.is_successful_reboot('GOOD'), True) 159 self.assertEquals(line.is_successful_reboot('WARN'), True) 160 161 162 def test_bad_reboot_passes_success_test(self): 163 """Tests bad reboot statuses.""" 164 line = version_1.status_line(0, 'NOSTATUS', None, 'reboot', 165 'reboot success', {}) 166 self.assertEquals(line.is_successful_reboot('FAIL'), False) 167 self.assertEquals(line.is_successful_reboot('ABORT'), False) 168 169 170 def test_get_kernel_returns_kernel_plus_patches(self): 171 """Tests that get_kernel returns the appropriate info.""" 172 line = version_1.status_line(0, 'GOOD', 'subdir', 'testname', 173 'reason text', 174 {'kernel': '2.6.24-rc40', 175 'patch0': 'first_patch 0 0', 176 'patch1': 'another_patch 0 0'}) 177 kern = line.get_kernel() 178 kernel_hash = utils.hash('md5', '2.6.24-rc40,0,0').hexdigest() 179 self.assertEquals(kern.base, '2.6.24-rc40') 180 self.assertEquals(kern.patches[0].spec, 'first_patch') 181 self.assertEquals(kern.patches[1].spec, 'another_patch') 182 self.assertEquals(len(kern.patches), 2) 183 self.assertEquals(kern.kernel_hash, kernel_hash) 184 185 186 def test_get_kernel_ignores_out_of_sequence_patches(self): 187 """Tests that get_kernel ignores patches that are out of sequence.""" 188 line = version_1.status_line(0, 'GOOD', 'subdir', 'testname', 189 'reason text', 190 {'kernel': '2.6.24-rc40', 191 'patch0': 'first_patch 0 0', 192 'patch2': 'another_patch 0 0'}) 193 kern = line.get_kernel() 194 kernel_hash = utils.hash('md5', '2.6.24-rc40,0').hexdigest() 195 self.assertEquals(kern.base, '2.6.24-rc40') 196 self.assertEquals(kern.patches[0].spec, 'first_patch') 197 self.assertEquals(len(kern.patches), 1) 198 self.assertEquals(kern.kernel_hash, kernel_hash) 199 200 201 def test_get_kernel_returns_unknown_with_no_kernel(self): 202 """Tests that a missing kernel is handled properly.""" 203 line = version_1.status_line(0, 'GOOD', 'subdir', 'testname', 204 'reason text', 205 {'patch0': 'first_patch 0 0', 206 'patch2': 'another_patch 0 0'}) 207 kern = line.get_kernel() 208 self.assertEquals(kern.base, 'UNKNOWN') 209 self.assertEquals(kern.patches, []) 210 self.assertEquals(kern.kernel_hash, 'UNKNOWN') 211 212 213 def test_get_timestamp_returns_timestamp_field(self): 214 """Tests that get_timestamp returns the expected info.""" 215 timestamp = datetime.datetime(1970, 1, 1, 4, 30) 216 timestamp -= datetime.timedelta(seconds=time.timezone) 217 line = version_1.status_line(0, 'GOOD', 'subdir', 'testname', 218 'reason text', 219 {'timestamp': '16200'}) 220 self.assertEquals(timestamp, line.get_timestamp()) 221 222 223 def test_get_timestamp_returns_none_on_missing_field(self): 224 """Tests that get_timestamp returns None if no timestamp exists.""" 225 line = version_1.status_line(0, 'GOOD', 'subdir', 'testname', 226 'reason text', {}) 227 self.assertEquals(None, line.get_timestamp()) 228 229 230class iteration_parse_line_into_dicts(unittest.TestCase): 231 """Tests for parsing iteration keyvals into dictionaries.""" 232 233 def parse_line(self, line): 234 """ 235 Invokes parse_line_into_dicts with two empty dictionaries. 236 237 @param line: The line to parse. 238 239 @return A 2-tuple representing the filled-in attr and perf dictionaries, 240 respectively. 241 242 """ 243 attr, perf = {}, {} 244 version_1.iteration.parse_line_into_dicts(line, attr, perf) 245 return attr, perf 246 247 248 def test_perf_entry(self): 249 """Tests a basic perf keyval line.""" 250 result = self.parse_line('perf-val{perf}=-173') 251 self.assertEqual(({}, {'perf-val': -173}), result) 252 253 254 def test_attr_entry(self): 255 """Tests a basic attr keyval line.""" 256 result = self.parse_line('attr-val{attr}=173') 257 self.assertEqual(({'attr-val': '173'}, {}), result) 258 259 260 def test_untagged_is_perf(self): 261 """Tests that an untagged keyval is considered to be perf by default.""" 262 result = self.parse_line('untagged=-678.5e5') 263 self.assertEqual(({}, {'untagged': -678.5e5}), result) 264 265 266 def test_invalid_tag_ignored(self): 267 """Tests that invalid tags are ignored.""" 268 result = self.parse_line('bad-tag{invalid}=56') 269 self.assertEqual(({}, {}), result) 270 271 272 def test_non_numeric_perf_ignored(self): 273 """Tests that non-numeric perf values are ignored.""" 274 result = self.parse_line('perf-val{perf}=FooBar') 275 self.assertEqual(({}, {}), result) 276 277 278 def test_non_numeric_untagged_ignored(self): 279 """Tests that non-numeric untagged keyvals are ignored.""" 280 result = self.parse_line('untagged=FooBar') 281 self.assertEqual(({}, {}), result) 282 283 284class perf_value_iteration_parse_line_into_dict(unittest.TestCase): 285 """Tests for parsing perf value iterations into a dictionary.""" 286 287 def parse_line(self, line): 288 """ 289 Invokes parse_line_into_dict with a line to parse. 290 291 @param line: The string line to parse. 292 293 @return A dictionary containing the information parsed from the line. 294 295 """ 296 return version_1.perf_value_iteration.parse_line_into_dict(line) 297 298 def test_invalid_json(self): 299 """Tests that a non-JSON line is handled properly.""" 300 result = self.parse_line('{"invalid_json" "string"}') 301 self.assertEqual(result, {}) 302 303 def test_single_value_int(self): 304 """Tests that a single integer value is parsed properly.""" 305 result = self.parse_line('{"value": 7}') 306 self.assertEqual(result, {'value': 7, 'stddev': 0}) 307 308 def test_single_value_float(self): 309 """Tests that a single float value is parsed properly.""" 310 result = self.parse_line('{"value": 1.298}') 311 self.assertEqual(result, {'value': 1.298, 'stddev': 0}) 312 313 def test_value_list_int(self): 314 """Tests that an integer list is parsed properly.""" 315 result = self.parse_line('{"value": [10, 20, 30]}') 316 self.assertEqual(result, {'value': 20.0, 'stddev': 10.0}) 317 318 def test_value_list_float(self): 319 """Tests that a float list is parsed properly.""" 320 result = self.parse_line('{"value": [2.0, 3.0, 4.0]}') 321 self.assertEqual(result, {'value': 3.0, 'stddev': 1.0}) 322 323 324class DummyAbortTestCase(unittest.TestCase): 325 """Tests for the make_stub_abort function.""" 326 327 def setUp(self): 328 self.indent = 3 329 self.subdir = "subdir" 330 self.testname = 'testname' 331 self.timestamp = 1220565792 332 self.reason = 'Job aborted unexpectedly' 333 334 335 def test_make_stub_abort_with_timestamp(self): 336 """Tests make_stub_abort with a timestamp specified.""" 337 abort = version_1.parser.make_stub_abort( 338 self.indent, self.subdir, self.testname, self.timestamp, 339 self.reason) 340 self.assertEquals( 341 abort, '%sEND ABORT\t%s\t%s\ttimestamp=%d\t%s' % ( 342 '\t' * self.indent, self.subdir, self.testname, self.timestamp, 343 self.reason)) 344 345 def test_make_stub_abort_with_no_subdir(self): 346 """Tests make_stub_abort with no subdir specified.""" 347 abort= version_1.parser.make_stub_abort( 348 self.indent, None, self.testname, self.timestamp, self.reason) 349 self.assertEquals( 350 abort, '%sEND ABORT\t----\t%s\ttimestamp=%d\t%s' % ( 351 '\t' * self.indent, self.testname, self.timestamp, self.reason)) 352 353 def test_make_stub_abort_with_no_testname(self): 354 """Tests make_stub_abort with no testname specified.""" 355 abort= version_1.parser.make_stub_abort( 356 self.indent, self.subdir, None, self.timestamp, self.reason) 357 self.assertEquals( 358 abort, '%sEND ABORT\t%s\t----\ttimestamp=%d\t%s' % ( 359 '\t' * self.indent, self.subdir, self.timestamp, self.reason)) 360 361 def test_make_stub_abort_no_timestamp(self): 362 """Tests make_stub_abort with no timestamp specified.""" 363 abort = version_1.parser.make_stub_abort( 364 self.indent, self.subdir, self.testname, None, self.reason) 365 self.assertEquals( 366 abort, '%sEND ABORT\t%s\t%s\t%s' % ( 367 '\t' * self.indent, self.subdir, self.testname, self.reason)) 368 369 370class test_parse_file(unittest.TestCase): 371 """Tests for parsing a status.log file.""" 372 373 class fake_job(models.job): 374 """Fake job object.""" 375 376 def exit_status(self): 377 """Fake exit_status method.""" 378 return 'FAIL' 379 380 @staticmethod 381 def _parse_host_keyval(job_dir, hostname): 382 return {} 383 384 @mock.patch.object(models.test, 'parse_host_keyval', _parse_host_keyval) 385 def test_top_level_fail_with_reason(self): 386 """Tests that a status.log with a FAIL keeps the reason.""" 387 job = self.fake_job('dir', 'user', 'label', 'machine', None, None, 388 None, None, None, None, None, None) 389 parser = version_1.parser() 390 parser.start(job) 391 tests = parser.end([ 392 'FAIL\t----\t----\ttimestamp=1615249387\tlocaltime=Mar 09 00:23:07\tThis is the reason.' 393 ]) 394 self.assertEquals(tests[0].status, 'FAIL') 395 self.assertEquals(tests[0].reason, 'This is the reason.') 396 397 398if __name__ == '__main__': 399 unittest.main() 400