| # 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. |
| http://src.chromium.org/viewvc/chrome/trunk/tools/build/scripts/master/ |
| log_parser/gtest_command.py?view=markup |
| """ |
| |
| 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('\[\s+RUN\s+\] ' + test_name_regexp) |
| self._test_ok = re.compile('\[\s+OK\s+\] ' + test_name_regexp) |
| self._test_fail = re.compile('\[\s+FAILED\s+\] ' + test_name_regexp) |
| self._test_timeout = re.compile( |
| 'Test timeout \([0-9]+ ms\) exceeded for ' + test_name_regexp) |
| self._disabled = re.compile(' YOU HAVE (\d+) DISABLED TEST') |
| self._flaky = re.compile(' YOU HAVE (\d+) FLAKY TEST') |
| |
| self._suppression_start = re.compile( |
| 'Suppression \(error hash=#([0-9A-F]+)#\):') |
| self._suppression_end = re.compile('^}\s*$') |
| |
| self._master_name_re = re.compile('\[Running for master: "([^"]*)"') |
| self.master_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 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.master_name: |
| results = self._master_name_re.search(line) |
| if results: |
| self.master_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) |