xref: /aosp_15_r20/external/autotest/tko/parsers/version_1_unittest.py (revision 9c5db1993ded3edbeafc8092d69fe5de2ee02df7)

#!/usr/bin/python3

from autotest_lib.tko import models
import datetime, time, unittest, mock

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_stub_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_stub_abort_with_timestamp(self):
        """Tests make_stub_abort with a timestamp specified."""
        abort = version_1.parser.make_stub_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_stub_abort_with_no_subdir(self):
        """Tests make_stub_abort with no subdir specified."""
        abort= version_1.parser.make_stub_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_stub_abort_with_no_testname(self):
        """Tests make_stub_abort with no testname specified."""
        abort= version_1.parser.make_stub_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_stub_abort_no_timestamp(self):
        """Tests make_stub_abort with no timestamp specified."""
        abort = version_1.parser.make_stub_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))


class test_parse_file(unittest.TestCase):
    """Tests for parsing a status.log file."""

    class fake_job(models.job):
        """Fake job object."""

        def exit_status(self):
            """Fake exit_status method."""
            return 'FAIL'

    @staticmethod
    def _parse_host_keyval(job_dir, hostname):
        return {}

    @mock.patch.object(models.test, 'parse_host_keyval', _parse_host_keyval)
    def test_top_level_fail_with_reason(self):
        """Tests that a status.log with a FAIL keeps the reason."""
        job = self.fake_job('dir', 'user', 'label', 'machine', None, None,
                            None, None, None, None, None, None)
        parser = version_1.parser()
        parser.start(job)
        tests = parser.end([
                'FAIL\t----\t----\ttimestamp=1615249387\tlocaltime=Mar 09 00:23:07\tThis is the reason.'
        ])
        self.assertEquals(tests[0].status, 'FAIL')
        self.assertEquals(tests[0].reason, 'This is the reason.')


if __name__ == '__main__':
    unittest.main()