test_result_util.py

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