xref: /aosp_15_r20/external/autotest/client/common_lib/gtest_parser.py (revision 9c5db1993ded3edbeafc8092d69fe5de2ee02df7)

# Lint as: python2, python3
# Copyright (c) 2017 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

"""Utility class to parse the output of a gtest suite run."""

import re


class gtest_parser(object):
    """This class knows how to understand GTest test output.

    The code was borrowed with minor changes from chrome utility gtest_command.
    """

    def __init__(self):
        # State tracking for log parsing
        self._current_test = ''
        self._failure_description = []
        self._current_suppression_hash = ''
        self._current_suppression = []

        # Line number currently being processed.
        self._line_number = 0

        # List of parsing errors, as human-readable strings.
        self.internal_error_lines = []

        # Tests are stored here as 'test.name': (status, [description]).
        # The status should be one of ('started', 'OK', 'failed', 'timeout').
        # The description is a list of lines detailing the test's error, as
        # reported in the log.
        self._test_status = {}

        # Suppressions are stored here as 'hash': [suppression].
        self._suppressions = {}

        # This may be either text or a number. It will be used in the phrase
        # '%s disabled' or '%s flaky' on the waterfall display.
        self.disabled_tests = 0
        self.flaky_tests = 0

        # Regular expressions for parsing GTest logs. Test names look like
        #   SomeTestCase.SomeTest
        #   SomeName/SomeTestCase.SomeTest/1
        # This regexp also matches SomeName.SomeTest/1, which should be
        # harmless.
        test_name_regexp = r'((\w+/)?\w+\.\w+(\.\w+)?(/\d+)?)'
        self._test_start = re.compile(r'\[\s+RUN\s+\] ' + test_name_regexp)
        self._test_ok = re.compile(r'\[\s+OK\s+\] ' + test_name_regexp)
        self._test_fail = re.compile(r'\[\s+FAILED\s+\] ' + test_name_regexp)
        self._test_timeout = re.compile(
                r'Test timeout \([0-9]+ ms\) exceeded for ' + test_name_regexp)
        self._disabled = re.compile(r'  YOU HAVE (\d+) DISABLED TEST')
        self._flaky = re.compile(r'  YOU HAVE (\d+) FLAKY TEST')

        self._suppression_start = re.compile(
                r'Suppression \(error hash=#([0-9A-F]+)#\):')
        self._suppression_end = re.compile(r'^}\s*$')

        # TODO b:169251326 terms below are set outside of this codebase
        # and should be updated when possible. ("master" -> "main")
        self._main_name_re = re.compile(
                r'\[Running for master: "([^"]*)"')  # nocheck
        self.main_name = ''

        self._error_logging_start_re = re.compile('=' * 70)
        self._error_logging_test_name_re = re.compile(
            '[FAIL|ERROR]: ' + test_name_regexp)
        self._error_logging_end_re = re.compile('-' * 70)
        self._error_logging_first_dash_found = False

    def _TestsByStatus(self, status, include_fails, include_flaky):
        """Returns list of tests with the given status.

        Args:
            status: test results status to search for.
            include_fails: If False, tests containing 'FAILS_' anywhere in
                their names will be excluded from the list.
            include_flaky: If False, tests containing 'FLAKY_' anywhere in
                their names will be excluded from the list.
        Returns:
            List of tests with the status.
        """
        test_list = [x[0] for x in self._test_status.items()
                     if self._StatusOfTest(x[0]) == status]

        if not include_fails:
            test_list = [x for x in test_list if x.find('FAILS_') == -1]
        if not include_flaky:
            test_list = [x for x in test_list if x.find('FLAKY_') == -1]

        return test_list

    def _StatusOfTest(self, test):
        """Returns the status code for the given test, or 'not known'."""
        test_status = self._test_status.get(test, ('not known', []))
        return test_status[0]

    def _RecordError(self, line, reason):
        """Record a log line that produced a parsing error.

        Args:
            line: text of the line at which the error occurred.
            reason: a string describing the error.
        """
        self.internal_error_lines.append("%s: %s [%s]" % (self._line_number,
                                                          line.strip(),
                                                          reason))

    def TotalTests(self):
        """Returns the number of parsed tests."""
        return len(self._test_status)

    def PassedTests(self):
        """Returns list of tests that passed."""
        return self._TestsByStatus('OK', False, False)

    def FailedTests(self, include_fails=False, include_flaky=False):
        """Returns list of tests that failed, timed out, or didn't finish.

        This list will be incorrect until the complete log has been processed,
        because it will show currently running tests as having failed.

        Args:
            include_fails: If true, all failing tests with FAILS_ in their
                names will be included. Otherwise, they will only be included
                if they crashed.
            include_flaky: If true, all failing tests with FLAKY_ in their
                names will be included. Otherwise, they will only be included
                if they crashed.
        Returns:
            List of failed tests.
        """
        return (self._TestsByStatus('failed', include_fails, include_flaky) +
                self._TestsByStatus('timeout', include_fails, include_flaky) +
                self._TestsByStatus('started', include_fails, include_flaky))

    def FailureDescription(self, test):
        """Returns a list containing the failure description for the given test.

        If the test didn't fail or timeout, returns [].
        Args:
            test: Name to test to find failure reason.
        Returns:
            List of test name, and failure string.
        """
        test_status = self._test_status.get(test, ('', []))
        return test_status[1]

    def SuppressionHashes(self):
        """Returns list of suppression hashes found in the log."""
        return list(self._suppressions.keys())

    def Suppression(self, suppression_hash):
        """Returns a list containing the suppression for a given hash.

        If the suppression hash doesn't exist, returns [].

        Args:
            suppression_hash: name of hash.
        Returns:
            List of suppression for the hash.
        """
        return self._suppressions.get(suppression_hash, [])

    def ProcessLogLine(self, line):
        """This is called once with each line of the test log."""

        # Track line number for error messages.
        self._line_number += 1

        if not self.main_name:
            results = self._main_name_re.search(line)
            if results:
                self.main_name = results.group(1)

        # Is it a line reporting disabled tests?
        results = self._disabled.search(line)
        if results:
            try:
                disabled = int(results.group(1))
            except ValueError:
                disabled = 0
            if disabled > 0 and isinstance(self.disabled_tests, int):
                self.disabled_tests += disabled
            else:
                # If we can't parse the line, at least give a heads-up. This is
                # a safety net for a case that shouldn't happen but isn't a
                # fatal error.
                self.disabled_tests = 'some'
            return

        # Is it a line reporting flaky tests?
        results = self._flaky.search(line)
        if results:
            try:
                flaky = int(results.group(1))
            except ValueError:
                flaky = 0
            if flaky > 0 and isinstance(self.flaky_tests, int):
                self.flaky_tests = flaky
            else:
                # If we can't parse the line, at least give a heads-up. This is
                # a safety net for a case that shouldn't happen but isn't a
                # fatal error.
                self.flaky_tests = 'some'
            return

        # Is it the start of a test?
        results = self._test_start.search(line)
        if results:
            test_name = results.group(1)
            if test_name in self._test_status:
                self._RecordError(line, 'test started more than once')
                return
            if self._current_test:
                status = self._StatusOfTest(self._current_test)
                if status in ('OK', 'failed', 'timeout'):
                    self._RecordError(line, 'start while in status %s' % status)
                    return
                if status not in ('failed', 'timeout'):
                    self._test_status[self._current_test] = (
                        'failed', self._failure_description)
            self._test_status[test_name] = ('started', ['Did not complete.'])
            self._current_test = test_name
            self._failure_description = []
            return

        # Is it a test success line?
        results = self._test_ok.search(line)
        if results:
            test_name = results.group(1)
            status = self._StatusOfTest(test_name)
            if status != 'started':
                self._RecordError(line, 'success while in status %s' % status)
                return
            self._test_status[test_name] = ('OK', [])
            self._failure_description = []
            self._current_test = ''
            return

        # Is it a test failure line?
        results = self._test_fail.search(line)
        if results:
            test_name = results.group(1)
            status = self._StatusOfTest(test_name)
            if status not in ('started', 'failed', 'timeout'):
                self._RecordError(line, 'failure while in status %s' % status)
                return
            # Don't overwrite the failure description when a failing test is
            # listed a second time in the summary, or if it was already
            # recorded as timing out.
            if status not in ('failed', 'timeout'):
                self._test_status[test_name] = ('failed',
                                                self._failure_description)
            self._failure_description = []
            self._current_test = ''
            return

        # Is it a test timeout line?
        results = self._test_timeout.search(line)
        if results:
            test_name = results.group(1)
            status = self._StatusOfTest(test_name)
            if status not in ('started', 'failed'):
                self._RecordError(line, 'timeout while in status %s' % status)
                return
            self._test_status[test_name] = (
                'timeout', self._failure_description + ['Killed (timed out).'])
            self._failure_description = []
            self._current_test = ''
            return

        # Is it the start of a new valgrind suppression?
        results = self._suppression_start.search(line)
        if results:
            suppression_hash = results.group(1)
            if suppression_hash in self._suppressions:
                self._RecordError(line, 'suppression reported more than once')
                return
            self._suppressions[suppression_hash] = []
            self._current_suppression_hash = suppression_hash
            self._current_suppression = [line]
            return

        # Is it the end of a valgrind suppression?
        results = self._suppression_end.search(line)
        if results and self._current_suppression_hash:
            self._current_suppression.append(line)
            self._suppressions[self._current_suppression_hash] = (
                self._current_suppression)
            self._current_suppression_hash = ''
            self._current_suppression = []
            return

        # Is it the start of a test summary error message?
        results = self._error_logging_test_name_re.search(line)
        if results:
            test_name = results.group(1)
            self._test_status[test_name] = ('failed', ['Output not found.'])
            self._current_test = test_name
            self._failure_description = []
            self._error_logging_first_dash_found = False
            return

        # Is it the start of the next test summary signaling the end
        # of the previous message?
        results = self._error_logging_start_re.search(line)
        if results and self._current_test:
            self._test_status[self._current_test] = ('failed',
                                                     self._failure_description)
            self._failure_description = []
            self._current_test = ''
            return

        # Is it the end of the extra test failure summaries?
        results = self._error_logging_end_re.search(line)
        if results and self._current_test:
            if self._error_logging_first_dash_found:
                self._test_status[self._current_test] = (
                    'failed', self._failure_description)
                self._failure_description = []
                self._current_test = ''
            self._error_logging_first_dash_found = True
            return

        # Random line: if we're in a suppression, collect it. Suppressions are
        # generated after all tests are finished, so this should always belong
        # to the current suppression hash.
        if self._current_suppression_hash:
            self._current_suppression.append(line)
            return

        # Random line: if we're in a test, collect it for the failure
        # description. Tests may run simultaneously, so this might be off, but
        # it's worth a try.
        if self._current_test:
            self._failure_description.append(line)