123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640 |
- # Copyright (c) 2016 The Chromium Authors. All rights reserved.
- # Use of this source code is governed by a BSD-style license that can be
- # found in the LICENSE file.
- import collections
- import copy
- import json
- import logging
- import re
- from test_result_util import ResultCollection, TestResult, TestStatus
- LOGGER = logging.getLogger(__name__)
- # These labels should match the ones output by gtest's JSON.
- TEST_UNKNOWN_LABEL = 'UNKNOWN'
- TEST_SUCCESS_LABEL = 'SUCCESS'
- TEST_FAILURE_LABEL = 'FAILURE'
- TEST_SKIPPED_LABEL = 'SKIPPED'
- TEST_TIMEOUT_LABEL = 'TIMEOUT'
- TEST_WARNING_LABEL = 'WARNING'
- DID_NOT_COMPLETE = 'Did not complete.'
- class GTestResult(object):
- """A result of gtest.
- The class will be depreacated soon. Please use
- |test_result_util.ResultCollection| instead. (crbug.com/1132476)
- Properties:
- command: The command argv.
- crashed: Whether or not the test crashed.
- crashed_test: The name of the test during which execution crashed, or
- None if a particular test didn't crash.
- failed_tests: A dict mapping the names of failed tests to a list of
- lines of output from those tests.
- flaked_tests: A dict mapping the names of failed flaky tests to a list
- of lines of output from those tests.
- passed_tests: A list of passed tests.
- perf_links: A dict mapping the names of perf data points collected
- to links to view those graphs.
- return_code: The return code of the command.
- success: Whether or not this run of the command was considered a
- successful GTest execution.
- """
- @property
- def crashed(self):
- return self._crashed
- @property
- def crashed_test(self):
- return self._crashed_test
- @property
- def command(self):
- return self._command
- @property
- def disabled_tests_from_compiled_tests_file(self):
- if self.__finalized:
- return copy.deepcopy(self._disabled_tests_from_compiled_tests_file)
- return self._disabled_tests_from_compiled_tests_file
- @property
- def failed_tests(self):
- if self.__finalized:
- return copy.deepcopy(self._failed_tests)
- return self._failed_tests
- @property
- def flaked_tests(self):
- if self.__finalized:
- return copy.deepcopy(self._flaked_tests)
- return self._flaked_tests
- @property
- def passed_tests(self):
- if self.__finalized:
- return copy.deepcopy(self._passed_tests)
- return self._passed_tests
- @property
- def perf_links(self):
- if self.__finalized:
- return copy.deepcopy(self._perf_links)
- return self._perf_links
- @property
- def return_code(self):
- return self._return_code
- @property
- def success(self):
- return self._success
- def __init__(self, command):
- if not isinstance(command, collections.Iterable):
- raise ValueError('Expected an iterable of command arguments.', command)
- if not command:
- raise ValueError('Expected a non-empty command.', command)
- self._command = tuple(command)
- self._crashed = False
- self._crashed_test = None
- self._disabled_tests_from_compiled_tests_file = []
- self._failed_tests = collections.OrderedDict()
- self._flaked_tests = collections.OrderedDict()
- self._passed_tests = []
- self._perf_links = collections.OrderedDict()
- self._return_code = None
- self._success = None
- self.__finalized = False
- def finalize(self, return_code, success):
- self._return_code = return_code
- self._success = success
- # If the test was not considered to be a GTest success, but had no
- # failing tests, conclude that it must have crashed.
- if not self._success and not self._failed_tests and not self._flaked_tests:
- self._crashed = True
- # At most one test can crash the entire app in a given parsing.
- for test, log_lines in self._failed_tests.items():
- # A test with no output would have crashed. No output is replaced
- # by the GTestLogParser by a sentence indicating non-completion.
- if DID_NOT_COMPLETE in log_lines:
- self._crashed = True
- self._crashed_test = test
- # A test marked as flaky may also have crashed the app.
- for test, log_lines in self._flaked_tests.items():
- if DID_NOT_COMPLETE in log_lines:
- self._crashed = True
- self._crashed_test = test
- self.__finalized = True
- class GTestLogParser(object):
- """This helper class process GTest test output."""
- def __init__(self):
- # Test results from the parser.
- self._result_collection = ResultCollection()
- # State tracking for log parsing
- self.completed = False
- self._current_test = ''
- self._failure_description = []
- self._parsing_failures = False
- # 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',
- # 'warning'). Warning indicates that a test did not pass when run in
- # parallel with other tests but passed when run alone. The description is
- # a list of lines detailing the test's error, as reported in the log.
- self._test_status = {}
- # 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
- # Disabled tests by parsing the compiled tests json file output from GTest.
- self._disabled_tests_from_compiled_tests_file = []
- self._flaky_tests = 0
- # Regular expressions for parsing GTest logs. Test names look like
- # "x.y", with 0 or more "w/" prefixes and 0 or more "/z" suffixes.
- # e.g.:
- # SomeName/SomeTestCase.SomeTest/1
- # SomeName/SomeTestCase/1.SomeTest
- # SomeName/SomeTestCase/1.SomeTest/SomeModifider
- test_name_regexp = r'((\w+/)*\w+\.\w+(/\w+)*)'
- self._master_name_re = re.compile(r'\[Running for master: "([^"]*)"')
- self.master_name = ''
- self._test_name = re.compile(test_name_regexp)
- 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_passed = re.compile(r'\[\s+PASSED\s+\] \d+ tests?.')
- self._test_skipped = re.compile(r'\[\s+SKIPPED\s+\] ' + test_name_regexp)
- self._run_test_cases_line = re.compile(
- r'\[\s*\d+\/\d+\]\s+[0-9\.]+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'\s*YOU HAVE (\d+) DISABLED TEST')
- self._flaky = re.compile(r'\s*YOU HAVE (\d+) FLAKY TEST')
- self._retry_message = re.compile('RETRYING FAILED TESTS:')
- self.retrying_failed = False
- self._compiled_tests_file_path = re.compile(
- '.*Wrote compiled tests to file: (\S+)')
- self.TEST_STATUS_MAP = {
- 'OK': TEST_SUCCESS_LABEL,
- 'failed': TEST_FAILURE_LABEL,
- 'skipped': TEST_SKIPPED_LABEL,
- 'timeout': TEST_TIMEOUT_LABEL,
- 'warning': TEST_WARNING_LABEL
- }
- def GetCurrentTest(self):
- return self._current_test
- def GetResultCollection(self):
- return self._result_collection
- 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 _TestsByStatus(self, status, include_fails, include_flaky):
- """Returns list of tests with the given status.
- Args:
- 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.
- """
- 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 _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 RunningTests(self):
- """Returns list of tests that appear to be currently running."""
- return self._TestsByStatus('started', True, True)
- def ParsingErrors(self):
- """Returns a list of lines that have caused parsing errors."""
- return self._internal_error_lines
- def ClearParsingErrors(self):
- """Clears the currently stored parsing errors."""
- self._internal_error_lines = ['Cleared.']
- def PassedTests(self, include_fails=False, include_flaky=False):
- """Returns list of tests that passed."""
- return self._TestsByStatus('OK', include_fails, include_flaky)
- def FailedTests(self, include_fails=False, include_flaky=False):
- """Returns list of tests that failed, timed out, or didn't finish
- (crashed).
- 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 or
- timed out.
- include_flaky: If true, all failing tests with FLAKY_ in their names will
- be included. Otherwise, they will only be included if they crashed or
- timed out.
- """
- return (self._TestsByStatus('failed', include_fails, include_flaky) +
- self._TestsByStatus('timeout', True, True) +
- self._TestsByStatus('warning', include_fails, include_flaky) +
- self.RunningTests())
- def SkippedTests(self, include_fails=False, include_flaky=False):
- """Returns list of tests that were skipped"""
- return self._TestsByStatus('skipped', include_fails, include_flaky)
- def TriesForTest(self, test):
- """Returns a list containing the state for all tries of the given test.
- This parser doesn't support retries so a single result is returned."""
- return [self.TEST_STATUS_MAP.get(self._StatusOfTest(test),
- TEST_UNKNOWN_LABEL)]
- def DisabledTests(self):
- """Returns the name of the disabled test (if there is only 1) or the number
- of disabled tests.
- """
- return self._disabled_tests
- def DisabledTestsFromCompiledTestsFile(self):
- """Returns the list of disabled tests in format '{TestCaseName}/{TestName}'.
- Find all test names starting with DISABLED_ from the compiled test json
- file if there is one. If there isn't or error in parsing, returns an
- empty list.
- """
- return self._disabled_tests_from_compiled_tests_file
- def FlakyTests(self):
- """Returns the name of the flaky test (if there is only 1) or the number
- of flaky tests.
- """
- return self._flaky_tests
- def FailureDescription(self, test):
- """Returns a list containing the failure description for the given test.
- If the test didn't fail or timeout, returns [].
- """
- test_status = self._test_status.get(test, ('', []))
- return ['%s: ' % test] + test_status[1]
- def CompletedWithoutFailure(self):
- """Returns True if all tests completed and no tests failed unexpectedly."""
- return self.completed and not self.FailedTests()
- def Finalize(self):
- """Finalize for |self._result_collection|.
- Called at the end to add unfinished tests and crash status for
- self._result_collection.
- """
- # Remaining logs after crash before exit.
- raw_remaining_logs = self._failure_description
- for test in self.RunningTests():
- self._test_status[test][1].extend(
- ['Potential test logs from crash until the end of test program:'])
- self._test_status[test][1].extend(raw_remaining_logs)
- self._result_collection.add_test_result(
- TestResult(
- test,
- TestStatus.CRASH,
- test_log='\n'.join(self._test_status[test][1])))
- self._result_collection.crashed = True
- if not self.completed:
- self._result_collection.crashed = True
- def _ParseDuration(self, line):
- """Returns test duration in milliseconds, None if not present."""
- # Test duration appears as suffix of status line like:
- # "[ OK ] SomeTest.SomeTestName (539 ms)"
- test_duration_regex = re.compile(r'\s+\(([0-9]+)\s+ms\)')
- results = test_duration_regex.search(line)
- if results:
- return int(results.group(1))
- return None
- def ProcessLine(self, line):
- """This is called once with each line of the test log."""
- # Track line number for error messages.
- self._line_number += 1
- # Some tests (net_unittests in particular) run subprocesses which can write
- # stuff to shared stdout buffer. Sometimes such output appears between new
- # line and gtest directives ('[ RUN ]', etc) which breaks the parser.
- # Code below tries to detect such cases and recognize a mixed line as two
- # separate lines.
- # List of regexps that parses expects to find at the start of a line but
- # which can be somewhere in the middle.
- gtest_regexps = [
- self._test_start,
- self._test_ok,
- self._test_fail,
- self._test_passed,
- self._test_skipped,
- ]
- for regexp in gtest_regexps:
- match = regexp.search(line)
- if match:
- break
- if not match or match.start() == 0:
- self._ProcessLine(line)
- else:
- self._ProcessLine(line[:match.start()])
- self._ProcessLine(line[match.start():])
- def _ProcessLine(self, line):
- """Parses the line and changes the state of parsed tests accordingly.
- Will recognize newly started tests, OK or FAILED statuses, timeouts, etc.
- """
- # Note: When sharding, the number of disabled and flaky tests will be read
- # multiple times, so this will only show the most recent values (but they
- # should all be the same anyway).
- # Is it a line listing the master name?
- if not self.master_name:
- results = self._master_name_re.match(line)
- if results:
- self.master_name = results.group(1)
- results = self._run_test_cases_line.match(line)
- if results:
- # A run_test_cases.py output.
- if self._current_test:
- if self._test_status[self._current_test][0] == 'started':
- self._test_status[self._current_test] = (
- 'timeout', self._failure_description)
- self._result_collection.add_test_result(
- TestResult(
- self._current_test,
- TestStatus.ABORT,
- test_log='\n'.join(self._failure_description)))
- self._current_test = ''
- self._failure_description = []
- return
- # Is it a line declaring all tests passed?
- results = self._test_passed.match(line)
- if results:
- self.completed = True
- self._current_test = ''
- return
- # Is it a line reporting disabled tests?
- results = self._disabled.match(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.match(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.match(line)
- if results:
- if self._current_test:
- if self._test_status[self._current_test][0] == 'started':
- self._test_status[self._current_test] = (
- 'timeout', self._failure_description)
- self._result_collection.add_test_result(
- TestResult(
- self._current_test,
- TestStatus.ABORT,
- test_log='\n'.join(self._failure_description)))
- test_name = results.group(1)
- self._test_status[test_name] = ('started', [DID_NOT_COMPLETE])
- self._current_test = test_name
- if self.retrying_failed:
- self._failure_description = self._test_status[test_name][1]
- self._failure_description.extend(['', 'RETRY OUTPUT:', ''])
- else:
- self._failure_description = []
- return
- # Is it a test success line?
- results = self._test_ok.match(line)
- if results:
- test_name = results.group(1)
- status = self._StatusOfTest(test_name)
- duration = self._ParseDuration(line)
- if status != 'started':
- self._RecordError(line, 'success while in status %s' % status)
- if self.retrying_failed:
- self._test_status[test_name] = ('warning', self._failure_description)
- # This is a passed result. Previous failures were reported in separate
- # TestResult objects.
- self._result_collection.add_test_result(
- TestResult(
- test_name,
- TestStatus.PASS,
- duration=duration,
- test_log='\n'.join(self._failure_description)))
- else:
- self._test_status[test_name] = ('OK', [])
- self._result_collection.add_test_result(
- TestResult(test_name, TestStatus.PASS, duration=duration))
- self._failure_description = []
- self._current_test = ''
- return
- # Is it a test skipped line?
- results = self._test_skipped.match(line)
- if results:
- test_name = results.group(1)
- status = self._StatusOfTest(test_name)
- # Skipped tests are listed again in the summary.
- if status not in ('started', 'skipped'):
- self._RecordError(line, 'skipped while in status %s' % status)
- self._test_status[test_name] = ('skipped', [])
- self._result_collection.add_test_result(
- TestResult(
- test_name,
- TestStatus.SKIP,
- expected_status=TestStatus.SKIP,
- test_log='Test skipped when running suite.'))
- self._failure_description = []
- self._current_test = ''
- return
- # Is it a test failure line?
- results = self._test_fail.match(line)
- if results:
- test_name = results.group(1)
- status = self._StatusOfTest(test_name)
- duration = self._ParseDuration(line)
- if status not in ('started', 'failed', 'timeout'):
- self._RecordError(line, 'failure while in status %s' % status)
- if self._current_test != test_name:
- if self._current_test:
- self._RecordError(
- line,
- '%s failure while in test %s' % (test_name, self._current_test))
- 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)
- # Add to |test_results| regardless whether the test ran before.
- self._result_collection.add_test_result(
- TestResult(
- test_name,
- TestStatus.FAIL,
- duration=duration,
- test_log='\n'.join(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)
- logs = self._failure_description + ['Killed (timed out).']
- self._test_status[test_name] = ('timeout', logs)
- self._result_collection.add_test_result(
- TestResult(test_name, TestStatus.ABORT, test_log='\n'.join(logs)))
- self._failure_description = []
- self._current_test = ''
- return
- # Is it the start of the retry tests?
- results = self._retry_message.match(line)
- if results:
- self.retrying_failed = True
- return
- # Is it the line containing path to the compiled tests json file?
- results = self._compiled_tests_file_path.match(line)
- if results:
- path = results.group(1)
- LOGGER.info('Compiled tests json file path: %s' % path)
- try:
- # TODO(crbug.com/1091345): Read the file when running on device.
- with open(path) as f:
- disabled_tests_from_json = []
- compiled_tests = json.load(f)
- for single_test in compiled_tests:
- test_case_name = single_test.get('test_case_name')
- test_name = single_test.get('test_name')
- if test_case_name and test_name and test_name.startswith(
- 'DISABLED_'):
- full_test_name = str('%s/%s' % (test_case_name, test_name))
- disabled_tests_from_json.append(full_test_name)
- self._result_collection.add_test_result(
- TestResult(
- test_name,
- TestStatus.SKIP,
- expected_status=TestStatus.SKIP,
- test_log='Test disabled.'))
- self._disabled_tests_from_compiled_tests_file = (
- disabled_tests_from_json)
- except Exception as e:
- LOGGER.warning(
- 'Error when finding disabled tests in compiled tests json file: %s'
- % e)
- 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.
- # This also won't work if a test times out before it begins running.
- if self._current_test:
- self._failure_description.append(line)
- # Parse the "Failing tests:" list at the end of the output, and add any
- # additional failed tests to the list. For example, this includes tests
- # that crash after the OK line.
- if self._parsing_failures:
- results = self._test_name.match(line)
- if results:
- test_name = results.group(1)
- status = self._StatusOfTest(test_name)
- if status in ('not known', 'OK'):
- unknown_error_log = 'Unknown error, see stdio log.'
- self._test_status[test_name] = ('failed', [unknown_error_log])
- self._result_collection.add_test_result(
- TestResult(
- test_name, TestStatus.FAIL, test_log=unknown_error_log))
- else:
- self._parsing_failures = False
- elif line.startswith('Failing tests:'):
- self._parsing_failures = True
|