# Copyright 2021 The Chromium Authors
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""Test result related classes."""
from collections import OrderedDict
import shard_util
import time
from result_sink_util import ResultSinkClient
_VALID_RESULT_COLLECTION_INIT_KWARGS = set(['test_results', 'crashed'])
_VALID_TEST_RESULT_INIT_KWARGS = set(
['attachments', 'duration', 'expected_status', 'test_log', 'test_loc'])
_VALID_TEST_STATUSES = set(['PASS', 'FAIL', 'CRASH', 'ABORT', 'SKIP'])
class TestStatus:
"""Enum storing possible test status(outcome).
Confirms to ResultDB TestStatus definitions:
https://source.chromium.org/chromium/infra/infra/+/main:go/src/go.chromium.org/luci/resultdb/proto/v1/test_result.proto
"""
PASS = 'PASS'
FAIL = 'FAIL'
CRASH = 'CRASH'
ABORT = 'ABORT'
SKIP = 'SKIP'
def _validate_kwargs(kwargs, valid_args_set):
"""Validates if keywords in kwargs are accepted."""
diff = set(kwargs.keys()) - valid_args_set
assert len(diff) == 0, 'Invalid keyword argument(s) in %s passed in!' % diff
def _validate_test_status(status):
"""Raises if input isn't valid."""
if not status in _VALID_TEST_STATUSES:
raise TypeError('Invalid test status: %s. Should be one of %s.' %
(status, _VALID_TEST_STATUSES))
def _to_standard_json_literal(status):
"""Converts TestStatus literal to standard JSON format requirement.
Standard JSON format defined at:
https://source.chromium.org/chromium/infra/infra/+/main:go/src/go.chromium.org/luci/resultdb/proto/v1/test_result.proto
ABORT is reported as "TIMEOUT" in standard JSON. The rest are the same.
"""
_validate_test_status(status)
return 'TIMEOUT' if status == TestStatus.ABORT else status
class TestResult(object):
"""Stores test outcome information of a single test run."""
def __init__(self, name, status, **kwargs):
"""Initializes an object.
Args:
name: (str) Name of a test. Typically includes
status: (str) Outcome of the test.
(Following are possible arguments in **kwargs):
attachments: (dict): Dict of unique attachment name to abs path mapping.
duration: (int) Test duration in milliseconds or None if unknown.
expected_status: (str) Expected test outcome for the run.
test_log: (str) Logs of the test.
test_loc: (dict): This is used to report test location info to resultSink.
data required in the dict can be found in
https://source.chromium.org/chromium/infra/infra/+/main:go/src/go.chromium.org/luci/resultdb/proto/v1/test_metadata.proto;l=32;drc=37488404d1c8aa8fccca8caae4809ece08828bae
"""
_validate_kwargs(kwargs, _VALID_TEST_RESULT_INIT_KWARGS)
assert isinstance(name, str), (
'Test name should be an instance of str. We got: %s') % type(name)
self.name = name
_validate_test_status(status)
self.status = status
self.attachments = kwargs.get('attachments', {})
self.duration = kwargs.get('duration')
self.expected_status = kwargs.get('expected_status', TestStatus.PASS)
self.test_log = kwargs.get('test_log', '')
self.test_loc = kwargs.get('test_loc', None)
# Use the var to avoid duplicate reporting.
self._reported_to_result_sink = False
def _compose_result_sink_tags(self):
"""Composes tags received by Result Sink from test result info."""
tags = [('test_name', self.name)]
# Only SKIP results have tags other than test name, to distinguish whether
# the SKIP is expected (disabled test) or not.
if self.status == TestStatus.SKIP:
if self.disabled():
tags.append(('disabled_test', 'true'))
else:
tags.append(('disabled_test', 'false'))
return tags
def disabled(self):
"""Returns whether the result represents a disabled test."""
return self.expected() and self.status == TestStatus.SKIP
def expected(self):
"""Returns whether the result is expected."""
return self.expected_status == self.status
def report_to_result_sink(self, result_sink_client):
"""Reports the single result to result sink if never reported.
Args:
result_sink_client: (result_sink_util.ResultSinkClient) Result sink client
to report test result.
"""
if not self._reported_to_result_sink:
result_sink_client.post(
self.name,
self.status,
self.expected(),
duration=self.duration,
test_log=self.test_log,
test_loc=self.test_loc,
tags=self._compose_result_sink_tags(),
file_artifacts=self.attachments)
self._reported_to_result_sink = True
class ResultCollection(object):
"""Stores a collection of TestResult for one or more test app launches."""
def __init__(self, **kwargs):
"""Initializes the object.
Args:
(Following are possible arguments in **kwargs):
crashed: (bool) Whether the ResultCollection is of a crashed test launch.
test_results: (list) A list of test_results to initialize the collection.
"""
_validate_kwargs(kwargs, _VALID_RESULT_COLLECTION_INIT_KWARGS)
self._test_results = []
self._crashed = kwargs.get('crashed', False)
self._crash_message = ''
self._spawning_test_launcher = False
self.add_results(kwargs.get('test_results', []))
@property
def crashed(self):
"""Whether the invocation(s) of the collection is regarded as crashed.
Crash indicates there might be tests unexpectedly not run that's not
included in |_test_results| in the collection.
"""
return self._crashed
@crashed.setter
def crashed(self, value):
"""Sets crash value."""
assert (type(value) == bool)
self._crashed = value
@property
def crash_message(self):
"""Logs from crashes in collection which are unrelated to single tests."""
return self._crash_message
@crash_message.setter
def crash_message(self, value):
"""Sets crash_message value."""
self._crash_message = value
@property
def test_results(self):
return self._test_results
@property
def spawning_test_launcher(self):
return self._spawning_test_launcher
@spawning_test_launcher.setter
def spawning_test_launcher(self, value):
"""Sets spawning_test_launcher value."""
assert (type(value) == bool)
self._spawning_test_launcher = value
def add_test_result(self, test_result):
"""Adds a single test result to collection.
Any new test addition should go through this method for all needed setups.
"""
self._test_results.append(test_result)
def add_result_collection(self,
another_collection,
ignore_crash=False,
overwrite_crash=False):
"""Adds results and status from another ResultCollection.
Args:
another_collection: (ResultCollection) The other collection to be added.
ignore_crash: (bool) Ignore any crashes from newly added collection.
overwrite_crash: (bool) Overwrite crash status of |self| and crash
message. Only applicable when ignore_crash=False.
"""
assert (not (ignore_crash and overwrite_crash))
if not ignore_crash:
if overwrite_crash:
self._crashed = False
self._crash_message = ''
self._crashed = self.crashed or another_collection.crashed
self.append_crash_message(another_collection.crash_message)
for test_result in another_collection.test_results:
self.add_test_result(test_result)
def add_results(self, test_results):
"""Adds a list of |TestResult|."""
for test_result in test_results:
self.add_test_result(test_result)
def add_name_prefix_to_tests(self, prefix):
"""Adds a prefix to all test names of results."""
for test_result in self._test_results:
test_result.name = '%s%s' % (prefix, test_result.name)
def add_test_names_status(self, test_names, test_status, **kwargs):
"""Adds a list of test names with given test status.
Args:
test_names: (list) A list of names of tests to add.
test_status: (str) The test outcome of the tests to add.
**kwargs: See possible **kwargs in TestResult.__init__ docstring.
"""
for test_name in test_names:
self.add_test_result(TestResult(test_name, test_status, **kwargs))
def add_and_report_test_names_status(self, test_names, test_status, **kwargs):
"""Adds a list of test names with status and report these to ResultSink.
Args:
test_names: (list) A list of names of tests to add.
test_status: (str) The test outcome of the tests to add.
**kwargs: See possible **kwargs in TestResult.__init__ docstring.
"""
another_collection = ResultCollection()
another_collection.add_test_names_status(test_names, test_status, **kwargs)
another_collection.report_to_result_sink()
self.add_result_collection(another_collection)
def append_crash_message(self, message):
"""Appends crash message str to current."""
if not message:
return
if self._crash_message:
self._crash_message += '\n'
self._crash_message += message
def all_test_names(self):
"""Returns a set of all test names in collection."""
return self.tests_by_expression(lambda result: True)
def tests_by_expression(self, expression):
"""A set of test names by filtering test results with given |expression|.
Args:
expression: (TestResult -> bool) A function or lambda expression which
accepts a TestResult object and returns bool.
"""
return set(
map(lambda result: result.name, filter(expression, self._test_results)))
def crashed_tests(self):
"""A set of test names with any crashed status in the collection."""
return self.tests_by_expression(lambda result: result.status == TestStatus.
CRASH)
def disabled_tests(self):
"""A set of disabled test names in the collection."""
return self.tests_by_expression(lambda result: result.disabled())
def expected_tests(self):
"""A set of test names with any expected status in the collection."""
return self.tests_by_expression(lambda result: result.expected())
def unexpected_tests(self):
"""A set of test names with any unexpected status in the collection."""
return self.tests_by_expression(lambda result: not result.expected())
def passed_tests(self):
"""A set of test names with any passed status in the collection."""
return self.tests_by_expression(lambda result: result.status == TestStatus.
PASS)
def failed_tests(self):
"""A set of test names with any failed status in the collection."""
return self.tests_by_expression(lambda result: result.status == TestStatus.
FAIL)
def flaky_tests(self):
"""A set of flaky test names in the collection."""
return self.expected_tests().intersection(self.unexpected_tests())
def never_expected_tests(self):
"""A set of test names with only unexpected status in the collection."""
return self.unexpected_tests().difference(self.expected_tests())
def pure_expected_tests(self):
"""A set of test names with only expected status in the collection."""
return self.expected_tests().difference(self.unexpected_tests())
def set_crashed_with_prefix(self, crash_message_prefix_line=''):
"""Updates collection with the crash status and add prefix to crash message.
Typically called at the end of runner run when runner reports failure due to
crash but there isn't unexpected tests. The crash status and crash message
will reflect in LUCI build page step log.
"""
self._crashed = True
if crash_message_prefix_line:
crash_message_prefix_line += '\n'
self._crash_message = crash_message_prefix_line + self.crash_message
def report_to_result_sink(self):
"""Reports current results to result sink once.
Note that each |TestResult| object stores whether it's been reported and
will only report itself once.
"""
result_sink_client = ResultSinkClient()
for test_result in self._test_results:
test_result.report_to_result_sink(result_sink_client)
result_sink_client.close()
def standard_json_output(self, path_delimiter='.'):
"""Returns a dict object confirming to Chromium standard format.
Format defined at:
https://chromium.googlesource.com/chromium/src/+/main/docs/testing/json_test_results_format.md
"""
num_failures_by_type = {}
tests = OrderedDict()
seen_names = set()
shard_index = shard_util.gtest_shard_index()
for test_result in self._test_results:
test_name = test_result.name
# For "num_failures_by_type" field. The field contains result count map of
# the first result of each test.
if test_name not in seen_names:
seen_names.add(test_name)
result_type = _to_standard_json_literal(test_result.status)
num_failures_by_type[result_type] = num_failures_by_type.get(
result_type, 0) + 1
# For "tests" field.
if test_name not in tests:
tests[test_name] = {
'expected': _to_standard_json_literal(test_result.expected_status),
'actual': _to_standard_json_literal(test_result.status),
'shard': shard_index,
'is_unexpected': not test_result.expected()
}
else:
tests[test_name]['actual'] += (
' ' + _to_standard_json_literal(test_result.status))
# This means there are both expected & unexpected results for the test.
# Thus, the overall status would be expected (is_unexpected = False)
# and the test is regarded flaky.
if tests[test_name]['is_unexpected'] != (not test_result.expected()):
tests[test_name]['is_unexpected'] = False
tests[test_name]['is_flaky'] = True
return {
'version': 3,
'path_delimiter': path_delimiter,
'seconds_since_epoch': int(time.time()),
'interrupted': self.crashed,
'num_failures_by_type': num_failures_by_type,
'tests': tests
}
def test_runner_logs(self):
"""Returns a dict object with test results as part of test runner logs."""
# Test name to merged test log in all unexpected results. Logs are
# only preserved for unexpected results.
unexpected_logs = {}
name_count = {}
for test_result in self._test_results:
if not test_result.expected():
test_name = test_result.name
name_count[test_name] = name_count.get(test_name, 0) + 1
logs = unexpected_logs.get(test_name, [])
logs.append('Failure log of attempt %d:' % name_count[test_name])
logs.extend(test_result.test_log.split('\n'))
unexpected_logs[test_name] = logs
passed = list(self.passed_tests() & self.pure_expected_tests())
disabled = list(self.disabled_tests())
flaked = {
test_name: unexpected_logs[test_name]
for test_name in self.flaky_tests()
}
# "failed" in test runner logs are all unexpected failures (including
# crash, etc).
failed = {
test_name: unexpected_logs[test_name]
for test_name in self.never_expected_tests()
}
logs = OrderedDict()
logs['passed tests'] = passed
if disabled:
logs['disabled tests'] = disabled
if flaked:
logs['flaked tests'] = flaked
if failed:
logs['failed tests'] = failed
for test, log_lines in failed.items():
logs[test] = log_lines
for test, log_lines in flaked.items():
logs[test] = log_lines
if self.crashed:
logs['test suite crash'] = self.crash_message.split('\n')
return logs
Codebase Browser
chromium