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