# 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)