<reponame>zealoussnow/chromium<filename>tools/perf/core/results_processor/formatters/json3_output.py
# Copyright 2019 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.

"""Output formatter for JSON Test Results Format.

Format specification:
https://chromium.googlesource.com/chromium/src/+/main/docs/testing/json_test_results_format.md
"""

import collections
import datetime
import json
import os

from core.results_processor import util


OUTPUT_FILENAME = 'test-results.json'


def ProcessIntermediateResults(test_results, options):
  """Process intermediate results and write output in output_dir."""
  results = Convert(test_results, options.output_dir, options.test_path_format)
  output_file = os.path.join(options.output_dir, OUTPUT_FILENAME)
  with open(output_file, 'w') as f:
    json.dump(results, f, sort_keys=True, indent=4, separators=(',', ': '))
  return output_file


def Convert(test_results, base_dir, test_path_format):
  """Convert intermediate results to the JSON Test Results Format.

  Args:
    test_results: The parsed intermediate results.
    base_dir: A string with the path to a base directory; artifact file paths
      will be written relative to this.

  Returns:
    A JSON serializable dict with the converted results.
  """
  results = {'tests': {}}
  status_counter = collections.Counter()

  for result in test_results:
    benchmark_name, story_name = util.SplitTestPath(result, test_path_format)
    actual_status = result['status']
    expected_status = actual_status if result['expected'] else 'PASS'
    status_counter[actual_status] += 1
    artifacts = result.get('outputArtifacts', {})
    shard = _GetTagValue(result.get('tags', []), 'shard', as_type=int)
    _MergeDict(
        results['tests'],
        {
            benchmark_name: {
                story_name: {
                    'actual': actual_status,
                    'expected': expected_status,
                    'is_unexpected': not result['expected'],
                    'times': float(result['runDuration'].rstrip('s')),
                    'shard': shard,
                    'artifacts': {
                        name: _ArtifactPath(artifact, base_dir)
                        for name, artifact in artifacts.items()
                    }
                }
            }
        }
    )

  for stories in results['tests'].values():
    for test in stories.values():
      test['actual'] = _DedupedStatus(test['actual'])
      test['expected'] = ' '.join(sorted(set(test['expected'])))
      test['is_unexpected'] = any(test['is_unexpected'])
      test['time'] = test['times'][0]
      test['shard'] = test['shard'][0]  # All shard values should be the same.
      if test['shard'] is None:
        del test['shard']

  # Test results are written in order of execution, so the first test start
  # time is approximately the start time of the whole suite.
  test_suite_start_time = (test_results[0]['startTime'] if test_results
                           else datetime.datetime.utcnow().isoformat() + 'Z')
  # If Telemetry stops with a unhandleable error, then remaining stories
  # are marked as unexpectedly skipped.
  interrupted = any(t['status'] == 'SKIP' and not t['expected']
                    for t in test_results)
  results.update(
      seconds_since_epoch=util.IsoTimestampToEpoch(test_suite_start_time),
      interrupted=interrupted,
      num_failures_by_type=dict(status_counter),
      path_delimiter='/',
      version=3,
  )

  return results


def _DedupedStatus(values):
  # TODO(crbug.com/754825): The following logic is a workaround for how the
  # flakiness dashboard determines whether a test is flaky. As a test_case
  # (i.e. story) may be run multiple times, we squash as sequence of PASS
  # results to a single one. Note this does not affect the total number of
  # passes in num_failures_by_type.
  deduped = set(values)
  if deduped == {'PASS'}:
    return 'PASS'
  elif deduped == {'SKIP'}:
    return 'SKIP'
  else:
    return ' '.join(values)


def _GetTagValue(tags, key, default=None, as_type=None):
  """Get the value of the first occurrence of a tag with a given key."""
  if as_type is None:
    as_type = lambda x: x
  return next((as_type(t['value']) for t in tags if t['key'] == key), default)


def _ArtifactPath(artifact, base_dir):
  """Extract either remote or local path of an artifact."""
  if 'fetchUrl' in artifact:
    return artifact['fetchUrl']
  else:
    # The spec calls for paths to be relative to the output directory and
    # '/'-delimited on all platforms.
    path = os.path.relpath(artifact['filePath'], base_dir)
    return path.replace(os.sep, '/')


def _MergeDict(target, values):
  # Is used to merge multiple runs of a story into a single test result.
  for key, value in values.items():
    if isinstance(value, dict):
      _MergeDict(target.setdefault(key, {}), value)
    elif isinstance(value, list):
      raise TypeError('Value to merge should not contain lists')
    else:  # i.e. a scalar value.
      target.setdefault(key, []).append(value)

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

"""Output formatter for JSON Test Results Format.

Format specification:
https://chromium.googlesource.com/chromium/src/+/main/docs/testing/json_test_results_format.md
"""

import collections
import datetime
import json
import os

from core.results_processor import util


OUTPUT_FILENAME = 'test-results.json'


def ProcessIntermediateResults(test_results, options):
  """Process intermediate results and write output in output_dir."""
  results = Convert(test_results, options.output_dir, options.test_path_format)
  output_file = os.path.join(options.output_dir, OUTPUT_FILENAME)
  with open(output_file, 'w') as f:
    json.dump(results, f, sort_keys=True, indent=4, separators=(',', ': '))
  return output_file


def Convert(test_results, base_dir, test_path_format):
  """Convert intermediate results to the JSON Test Results Format.

  Args:
    test_results: The parsed intermediate results.
    base_dir: A string with the path to a base directory; artifact file paths
      will be written relative to this.

  Returns:
    A JSON serializable dict with the converted results.
  """
  results = {'tests': {}}
  status_counter = collections.Counter()

  for result in test_results:
    benchmark_name, story_name = util.SplitTestPath(result, test_path_format)
    actual_status = result['status']
    expected_status = actual_status if result['expected'] else 'PASS'
    status_counter[actual_status] += 1
    artifacts = result.get('outputArtifacts', {})
    shard = _GetTagValue(result.get('tags', []), 'shard', as_type=int)
    _MergeDict(
        results['tests'],
        {
            benchmark_name: {
                story_name: {
                    'actual': actual_status,
                    'expected': expected_status,
                    'is_unexpected': not result['expected'],
                    'times': float(result['runDuration'].rstrip('s')),
                    'shard': shard,
                    'artifacts': {
                        name: _ArtifactPath(artifact, base_dir)
                        for name, artifact in artifacts.items()
                    }
                }
            }
        }
    )

  for stories in results['tests'].values():
    for test in stories.values():
      test['actual'] = _DedupedStatus(test['actual'])
      test['expected'] = ' '.join(sorted(set(test['expected'])))
      test['is_unexpected'] = any(test['is_unexpected'])
      test['time'] = test['times'][0]
      test['shard'] = test['shard'][0]  # All shard values should be the same.
      if test['shard'] is None:
        del test['shard']

  # Test results are written in order of execution, so the first test start
  # time is approximately the start time of the whole suite.
  test_suite_start_time = (test_results[0]['startTime'] if test_results
                           else datetime.datetime.utcnow().isoformat() + 'Z')
  # If Telemetry stops with a unhandleable error, then remaining stories
  # are marked as unexpectedly skipped.
  interrupted = any(t['status'] == 'SKIP' and not t['expected']
                    for t in test_results)
  results.update(
      seconds_since_epoch=util.IsoTimestampToEpoch(test_suite_start_time),
      interrupted=interrupted,
      num_failures_by_type=dict(status_counter),
      path_delimiter='/',
      version=3,
  )

  return results


def _DedupedStatus(values):
  # TODO(crbug.com/754825): The following logic is a workaround for how the
  # flakiness dashboard determines whether a test is flaky. As a test_case
  # (i.e. story) may be run multiple times, we squash as sequence of PASS
  # results to a single one. Note this does not affect the total number of
  # passes in num_failures_by_type.
  deduped = set(values)
  if deduped == {'PASS'}:
    return 'PASS'
  elif deduped == {'SKIP'}:
    return 'SKIP'
  else:
    return ' '.join(values)


def _GetTagValue(tags, key, default=None, as_type=None):
  """Get the value of the first occurrence of a tag with a given key."""
  if as_type is None:
    as_type = lambda x: x
  return next((as_type(t['value']) for t in tags if t['key'] == key), default)


def _ArtifactPath(artifact, base_dir):
  """Extract either remote or local path of an artifact."""
  if 'fetchUrl' in artifact:
    return artifact['fetchUrl']
  else:
    # The spec calls for paths to be relative to the output directory and
    # '/'-delimited on all platforms.
    path = os.path.relpath(artifact['filePath'], base_dir)
    return path.replace(os.sep, '/')


def _MergeDict(target, values):
  # Is used to merge multiple runs of a story into a single test result.
  for key, value in values.items():
    if isinstance(value, dict):
      _MergeDict(target.setdefault(key, {}), value)
    elif isinstance(value, list):
      raise TypeError('Value to merge should not contain lists')
    else:  # i.e. a scalar value.
      target.setdefault(key, []).append(value)
# Copyright 2019 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. Output formatter for JSON Test Results Format.  Format specification: https://chromium.googlesource.com/chromium/src/+/main/docs/testing/json_test_results_format.md Process intermediate results and write output in output_dir. Convert intermediate results to the JSON Test Results Format.    Args:     test_results: The parsed intermediate results.     base_dir: A string with the path to a base directory; artifact file paths       will be written relative to this.    Returns:     A JSON serializable dict with the converted results. # All shard values should be the same. # Test results are written in order of execution, so the first test start # time is approximately the start time of the whole suite. # If Telemetry stops with a unhandleable error, then remaining stories # are marked as unexpectedly skipped. # TODO(crbug.com/754825): The following logic is a workaround for how the # flakiness dashboard determines whether a test is flaky. As a test_case # (i.e. story) may be run multiple times, we squash as sequence of PASS # results to a single one. Note this does not affect the total number of # passes in num_failures_by_type. Get the value of the first occurrence of a tag with a given key. Extract either remote or local path of an artifact. # The spec calls for paths to be relative to the output directory and # '/'-delimited on all platforms. # Is used to merge multiple runs of a story into a single test result. # i.e. a scalar value.