#!/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.subdir = "subdir"
|
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, self.subdir, self.testname, self.timestamp,
|
self.reason)
|
self.assertEquals(
|
abort, '%sEND ABORT\t%s\t%s\ttimestamp=%d\t%s' % (
|
'\t' * self.indent, self.subdir, self.testname, self.timestamp,
|
self.reason))
|
|
def test_make_dummy_abort_with_no_subdir(self):
|
"""Tests make_dummy_abort with no subdir 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_with_no_testname(self):
|
"""Tests make_dummy_abort with no testname specified."""
|
abort= version_1.parser.make_dummy_abort(
|
self.indent, self.subdir, None, self.timestamp, self.reason)
|
self.assertEquals(
|
abort, '%sEND ABORT\t%s\t----\ttimestamp=%d\t%s' % (
|
'\t' * self.indent, self.subdir, 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, self.subdir, self.testname, None, self.reason)
|
self.assertEquals(
|
abort, '%sEND ABORT\t%s\t%s\t%s' % (
|
'\t' * self.indent, self.subdir, self.testname, self.reason))
|
|
|
if __name__ == '__main__':
|
unittest.main()
|
