#!/usr/bin/python import datetime, time, unittest import common from autotest_lib.client.common_lib import utils from autotest_lib.tko.parsers import version_1 class test_status_line(unittest.TestCase): """Tests for status lines.""" statuses = ['GOOD', 'WARN', 'FAIL', 'ABORT'] def test_handles_start(self): """Tests that START is handled properly.""" line = version_1.status_line(0, 'START', '----', 'test', '', {}) self.assertEquals(line.type, 'START') self.assertEquals(line.status, None) def test_handles_info(self): """Tests that INFO is handled properly.""" line = version_1.status_line(0, 'INFO', '----', '----', '', {}) self.assertEquals(line.type, 'INFO') self.assertEquals(line.status, None) def test_handles_status(self): """Tests that STATUS is handled properly.""" for stat in self.statuses: line = version_1.status_line(0, stat, '----', 'test', '', {}) self.assertEquals(line.type, 'STATUS') self.assertEquals(line.status, stat) def test_handles_endstatus(self): """Tests that END is handled properly.""" for stat in self.statuses: line = version_1.status_line(0, 'END ' + stat, '----', 'test', '', {}) self.assertEquals(line.type, 'END') self.assertEquals(line.status, stat) def test_fails_on_bad_status(self): """Tests that an exception is raised on a bad status.""" for stat in self.statuses: self.assertRaises(AssertionError, version_1.status_line, 0, 'BAD ' + stat, '----', 'test', '', {}) def test_saves_all_fields(self): """Tests that all fields are saved.""" line = version_1.status_line(5, 'GOOD', 'subdir_name', 'test_name', 'my reason here', {'key1': 'value', 'key2': 'another value', 'key3': 'value3'}) self.assertEquals(line.indent, 5) self.assertEquals(line.status, 'GOOD') self.assertEquals(line.subdir, 'subdir_name') self.assertEquals(line.testname, 'test_name') self.assertEquals(line.reason, 'my reason here') self.assertEquals(line.optional_fields, {'key1': 'value', 'key2': 'another value', 'key3': 'value3'}) def test_parses_blank_subdir(self): """Tests that a blank subdirectory is parsed properly.""" line = version_1.status_line(0, 'GOOD', '----', 'test', '', {}) self.assertEquals(line.subdir, None) def test_parses_blank_testname(self): """Tests that a blank test name is parsed properly.""" line = version_1.status_line(0, 'GOOD', 'subdir', '----', '', {}) self.assertEquals(line.testname, None) def test_parse_line_smoketest(self): """Runs a parse line smoke test.""" input_data = ('\t\t\tGOOD\t----\t----\t' 'field1=val1\tfield2=val2\tTest Passed') line = version_1.status_line.parse_line(input_data) self.assertEquals(line.indent, 3) self.assertEquals(line.type, 'STATUS') self.assertEquals(line.status, 'GOOD') self.assertEquals(line.subdir, None) self.assertEquals(line.testname, None) self.assertEquals(line.reason, 'Test Passed') self.assertEquals(line.optional_fields, {'field1': 'val1', 'field2': 'val2'}) def test_parse_line_handles_newline(self): """Tests that newlines are handled properly.""" input_data = ('\t\tGOOD\t----\t----\t' 'field1=val1\tfield2=val2\tNo newline here!') for suffix in ('', '\n'): line = version_1.status_line.parse_line(input_data + suffix) self.assertEquals(line.indent, 2) self.assertEquals(line.type, 'STATUS') self.assertEquals(line.status, 'GOOD') self.assertEquals(line.subdir, None) self.assertEquals(line.testname, None) self.assertEquals(line.reason, 'No newline here!') self.assertEquals(line.optional_fields, {'field1': 'val1', 'field2': 'val2'}) def test_parse_line_fails_on_untabbed_lines(self): """Tests that untabbed lines do not parse.""" input_data = ' GOOD\trandom\tfields\tof text' line = version_1.status_line.parse_line(input_data) self.assertEquals(line, None) line = version_1.status_line.parse_line(input_data.lstrip()) self.assertEquals(line.indent, 0) self.assertEquals(line.type, 'STATUS') self.assertEquals(line.status, 'GOOD') self.assertEquals(line.subdir, 'random') self.assertEquals(line.testname, 'fields') self.assertEquals(line.reason, 'of text') self.assertEquals(line.optional_fields, {}) def test_parse_line_fails_on_incomplete_lines(self): """Tests that incomplete lines do not parse.""" input_data = '\t\tGOOD\tfield\tsecond field' complete_data = input_data + '\tneeded last field' line = version_1.status_line.parse_line(input_data) self.assertEquals(line, None) line = version_1.status_line.parse_line(complete_data) self.assertEquals(line.indent, 2) self.assertEquals(line.type, 'STATUS') self.assertEquals(line.status, 'GOOD') self.assertEquals(line.subdir, 'field') self.assertEquals(line.testname, 'second field') self.assertEquals(line.reason, 'needed last field') self.assertEquals(line.optional_fields, {}) def test_good_reboot_passes_success_test(self): """Tests good reboot statuses.""" line = version_1.status_line(0, 'NOSTATUS', None, 'reboot', 'reboot success', {}) self.assertEquals(line.is_successful_reboot('GOOD'), True) self.assertEquals(line.is_successful_reboot('WARN'), True) def test_bad_reboot_passes_success_test(self): """Tests bad reboot statuses.""" line = version_1.status_line(0, 'NOSTATUS', None, 'reboot', 'reboot success', {}) self.assertEquals(line.is_successful_reboot('FAIL'), False) self.assertEquals(line.is_successful_reboot('ABORT'), False) def test_get_kernel_returns_kernel_plus_patches(self): """Tests that get_kernel returns the appropriate info.""" line = version_1.status_line(0, 'GOOD', 'subdir', 'testname', 'reason text', {'kernel': '2.6.24-rc40', 'patch0': 'first_patch 0 0', 'patch1': 'another_patch 0 0'}) kern = line.get_kernel() kernel_hash = utils.hash('md5', '2.6.24-rc40,0,0').hexdigest() self.assertEquals(kern.base, '2.6.24-rc40') self.assertEquals(kern.patches[0].spec, 'first_patch') self.assertEquals(kern.patches[1].spec, 'another_patch') self.assertEquals(len(kern.patches), 2) self.assertEquals(kern.kernel_hash, kernel_hash) def test_get_kernel_ignores_out_of_sequence_patches(self): """Tests that get_kernel ignores patches that are out of sequence.""" line = version_1.status_line(0, 'GOOD', 'subdir', 'testname', 'reason text', {'kernel': '2.6.24-rc40', 'patch0': 'first_patch 0 0', 'patch2': 'another_patch 0 0'}) kern = line.get_kernel() kernel_hash = utils.hash('md5', '2.6.24-rc40,0').hexdigest() self.assertEquals(kern.base, '2.6.24-rc40') self.assertEquals(kern.patches[0].spec, 'first_patch') self.assertEquals(len(kern.patches), 1) self.assertEquals(kern.kernel_hash, kernel_hash) def test_get_kernel_returns_unknown_with_no_kernel(self): """Tests that a missing kernel is handled properly.""" line = version_1.status_line(0, 'GOOD', 'subdir', 'testname', 'reason text', {'patch0': 'first_patch 0 0', 'patch2': 'another_patch 0 0'}) kern = line.get_kernel() self.assertEquals(kern.base, 'UNKNOWN') self.assertEquals(kern.patches, []) self.assertEquals(kern.kernel_hash, 'UNKNOWN') def test_get_timestamp_returns_timestamp_field(self): """Tests that get_timestamp returns the expected info.""" timestamp = datetime.datetime(1970, 1, 1, 4, 30) timestamp -= datetime.timedelta(seconds=time.timezone) line = version_1.status_line(0, 'GOOD', 'subdir', 'testname', 'reason text', {'timestamp': '16200'}) self.assertEquals(timestamp, line.get_timestamp()) def test_get_timestamp_returns_none_on_missing_field(self): """Tests that get_timestamp returns None if no timestamp exists.""" line = version_1.status_line(0, 'GOOD', 'subdir', 'testname', 'reason text', {}) self.assertEquals(None, line.get_timestamp()) class iteration_parse_line_into_dicts(unittest.TestCase): """Tests for parsing iteration keyvals into dictionaries.""" def parse_line(self, line): """ Invokes parse_line_into_dicts with two empty dictionaries. @param line: The line to parse. @return A 2-tuple representing the filled-in attr and perf dictionaries, respectively. """ attr, perf = {}, {} version_1.iteration.parse_line_into_dicts(line, attr, perf) return attr, perf def test_perf_entry(self): """Tests a basic perf keyval line.""" result = self.parse_line('perf-val{perf}=-173') self.assertEqual(({}, {'perf-val': -173}), result) def test_attr_entry(self): """Tests a basic attr keyval line.""" result = self.parse_line('attr-val{attr}=173') self.assertEqual(({'attr-val': '173'}, {}), result) def test_untagged_is_perf(self): """Tests that an untagged keyval is considered to be perf by default.""" result = self.parse_line('untagged=-678.5e5') self.assertEqual(({}, {'untagged': -678.5e5}), result) def test_invalid_tag_ignored(self): """Tests that invalid tags are ignored.""" result = self.parse_line('bad-tag{invalid}=56') self.assertEqual(({}, {}), result) def test_non_numeric_perf_ignored(self): """Tests that non-numeric perf values are ignored.""" result = self.parse_line('perf-val{perf}=FooBar') self.assertEqual(({}, {}), result) def test_non_numeric_untagged_ignored(self): """Tests that non-numeric untagged keyvals are ignored.""" result = self.parse_line('untagged=FooBar') self.assertEqual(({}, {}), result) class perf_value_iteration_parse_line_into_dict(unittest.TestCase): """Tests for parsing perf value iterations into a dictionary.""" def parse_line(self, line): """ Invokes parse_line_into_dict with a line to parse. @param line: The string line to parse. @return A dictionary containing the information parsed from the line. """ return version_1.perf_value_iteration.parse_line_into_dict(line) def test_invalid_json(self): """Tests that a non-JSON line is handled properly.""" result = self.parse_line('{"invalid_json" "string"}') self.assertEqual(result, {}) def test_single_value_int(self): """Tests that a single integer value is parsed properly.""" result = self.parse_line('{"value": 7}') self.assertEqual(result, {'value': 7, 'stddev': 0}) def test_single_value_float(self): """Tests that a single float value is parsed properly.""" result = self.parse_line('{"value": 1.298}') self.assertEqual(result, {'value': 1.298, 'stddev': 0}) def test_value_list_int(self): """Tests that an integer list is parsed properly.""" result = self.parse_line('{"value": [10, 20, 30]}') self.assertEqual(result, {'value': 20.0, 'stddev': 10.0}) def test_value_list_float(self): """Tests that a float list is parsed properly.""" result = self.parse_line('{"value": [2.0, 3.0, 4.0]}') self.assertEqual(result, {'value': 3.0, 'stddev': 1.0}) class DummyAbortTestCase(unittest.TestCase): """Tests for the make_dummy_abort function.""" def setUp(self): self.indent = 3 self.testname = 'testname' self.timestamp = 1220565792 self.reason = 'Job aborted unexpectedly' def test_make_dummy_abort_with_timestamp(self): """Tests make_dummy_abort with a timestamp specified.""" abort = version_1.parser.make_dummy_abort( self.indent, None, self.testname, self.timestamp, self.reason) self.assertEquals( abort, '%sEND ABORT\t----\t%s\ttimestamp=%d\t%s' % ( '\t' * self.indent, self.testname, self.timestamp, self.reason)) def test_make_dummy_abort_no_timestamp(self): """Tests make_dummy_abort with no timestamp specified.""" abort = version_1.parser.make_dummy_abort( self.indent, None, self.testname, None, self.reason) self.assertEquals( abort, '%sEND ABORT\t----\t%s\t%s' % ( '\t' * self.indent, self.testname, self.reason)) if __name__ == '__main__': unittest.main()