# Copyright 2019 The Fuchsia 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 operator

import attr

from recipe_engine import recipe_api
from recipe_engine.config_types import Path

from RECIPE_MODULES.fuchsia.testsharder import api as testsharder_api

# List of available targets.
TARGETS = ['x64', 'arm64']

# Name of BigQuery project and table for uploading artifacts.
BIGQUERY_PROJECT = 'fuchsia-infra'
BIGQUERY_ARTIFACTS_DATASET = 'artifacts'
TEST_SUMMARY_JSON = 'summary.json'
KERNEL_LOG = 'kernel_log.txt'
COVARGS_LOG_LEVEL = 'debug'
COVARGS_OUTPUT_JSON = 'covargs-output.json'


@attr.s
class FuchsiaTestResults(object):
  """Represents the result of testing of a Fuchsia build.

  Attributes:
    from_fuchsia (bool): Whether the tests ran on Fuchsia.
    results_dir (Path): The directory that the test results archive has
      been unpacked into.
    output_dir (Path): A directory containing the outputs of the swarming
      task that ran these tests. Anything that's in this directory will be
      uploaded to GCS when upload_results() is called.
    swarming_task_id (str): The ID of the task that ran these tests.
    symbolizer_json_output (Path or None): The path to the json trigger
      information produced by the symbolizer.
    env_name (str): The name of the task that ran these tests.
    tests (seq(testsharder.Test)): The tests that this task was instructed
      to run (as opposed to the results of the tests that this task *did*
      run, which are enumerated in `summary`).
    legacy_qemu (bool): Whether these tests were run using QEMU with
      runtests (no ssh).
    api (RecipeApi): The api to use for accessing recipe modules from this
      object.
    symbolizer_output (Path or None): The path to the symbolized log file
      produced by running these tests.
    overwrite_summary (bool): Whether to set the "name" and "gn_label" fields
      in the summary.json produced by these tests using the corresponding
      values from the input tests.json. Only affects legacy QEMU tests.
      (Solely for backwards compatibility with fuchsia_perf.)
  """

  from_fuchsia = attr.ib(type=bool)
  results_dir = attr.ib(type=Path)
  output_dir = attr.ib(type=Path)
  swarming_task_id = attr.ib(type=str)
  symbolizer_json_output = attr.ib(type=Path)
  _env_name = attr.ib(type=str)
  _tests = attr.ib(type=testsharder_api.Test)
  _legacy_qemu = attr.ib(type=bool)
  _api = attr.ib(type=recipe_api.RecipeApi)
  _symbolizer_output = attr.ib(None, type=Path)
  # TODO(fxb/10410): Get rid of overwrite_summary after fuchsia_perf is dead.
  _overwrite_summary = attr.ib(True, type=bool)

  # A mapping of relative paths to files in the results_dir containing
  # stdout+stderr data to strings containing those contents.
  _outputs = attr.ib(factory=dict, init=False)

  # Set lazily by the `summary` property, not a parameter to __init__.
  _summary = attr.ib(None, init=False)

  # Constants representing the result of running a test. These enumerate the
  # values of the 'results' field of the entries in the summary.json file
  # obtained from the target device.
  _TEST_RESULT_PASS = 'PASS'
  _TEST_RESULT_FAIL = 'FAIL'

  @property
  def summary(self):
    """The parsed summary file as a Dict or {} if missing."""
    if self._summary is None:
      self._summary = self._parse_summary()
    return self._summary

  @property
  def summary_lines(self):
    """Returns a list of the lines of the summary.json file."""
    return self._api.json.dumps(self.summary, indent=2).splitlines()

  @property
  def passed(self):
    """Whether all the tests passed."""
    tests = self.summary.get('tests', [])
    return all(test['result'] == self._TEST_RESULT_PASS for test in tests)

  @property
  def passed_test_outputs(self):
    """All entries in |self._outputs| for tests that passed."""
    return self._filter_outputs_by_test_result(self._TEST_RESULT_PASS)

  @property
  def failed_test_outputs(self):
    """All entries in |self._outputs| for tests that failed."""
    return self._filter_outputs_by_test_result(self._TEST_RESULT_FAIL)

  def _filter_outputs_by_test_result(self, result):
    """Returns all entries in |self._outputs| whose result is |result|.

    Args:
      result (String): one of the _TEST_RESULT_* constants from this class.

    Returns:
      A dict whose keys are the test names and whose values are the paths to
      the files containing each test's stderr+stdout data.
    """
    matches = collections.OrderedDict()
    for test in self.summary.get('tests', ()):
      if test['result'] == result:
        # The 'output_file' field is a path to the file containing the
        # stderr+stdout data for the test.
        matches[test['name']] = test['output_file']

    return matches

  def _parse_summary(self):
    summary_path = self.results_dir.join(TEST_SUMMARY_JSON)
    if not self._api.path.exists(summary_path):
      return {}

    try:
      summary = self._api.file.read_json('read %s' % TEST_SUMMARY_JSON,
                                         summary_path)
    except ValueError as e:  # pragma: no cover
      # TODO(olivernewman): JSONDecodeError in python >=3.5
      raise self._api.step.StepFailure('Invalid %s: %s' %
                                       (TEST_SUMMARY_JSON, e.args[0]))

    if not self._overwrite_summary or not self._legacy_qemu:
      return summary
    # We want all Fuchsia tests to have the package URL in the name field. But
    # legacy QEMU tests set "name" to be the test install path (since the test
    # list sent to QEMU is a list of paths). So overwrite the "name" field to
    # be the package URL instead.
    # Also set "gn_label", which doesn't automatically get passed through from
    # tests.json.
    # TODO(garymm,joshuaseaton): Once deprecated testing code path is removed,
    # self._legacy_qemu will always be False and we can remove this logic.
    tests_by_path = {test.path: test for test in self._tests}
    for summary_test in summary['tests']:
      path = summary_test['name']
      # Some zircon tests get run even though they don't show up in tests.json.
      # TODO(olivernewman): After build unification is complete we can assume
      # that every test in summary.json will have a corresponding entry in
      # tests.json, so get rid of this check and update every summary test.
      if path in tests_by_path:
        test = tests_by_path[path]
        assert test.package_url
        summary_test.update(
            name=test.package_url,
            gn_label=test.label,
        )
    return summary

  def present_tests(self, show_failures_in_red, show_passed):
    for test, stdio_file in self.failed_test_outputs.iteritems():
      name = 'failed: %s' % test
      if stdio_file not in self._outputs:
        self._outputs[stdio_file] = self._api.file.read_text(
            name, self.results_dir.join(stdio_file), test_data='output')
        step = self._api.step.active_result
        # file.read_text creates a step that displays a log of the file contents
        # with the name as the basename of the file read. We want to remove this
        # log and rename it to `stdio`.
        step.presentation.logs.pop(self._api.path.basename(stdio_file))
      else:
        step = self._api.step(name, None)
      step.presentation.logs['stdio'] = self._outputs[stdio_file].splitlines()
      if show_failures_in_red:
        step.presentation.status = self._api.step.FAILURE

    # There's recipe overhead that makes step creation slow, which we mitigate
    # by cramming all the passed tests into a single step. We also skip
    # presenting stdio since it's generally only useful if the test failed.
    with self._api.step.nest('all passed tests') as passed_tests_step:
      passed_tests = self.passed_test_outputs
      passed_tests_step.presentation.step_summary_text = ('%d passed tests' %
                                                          len(passed_tests))
      if show_passed:
        # Start with a newline to prevent the first test from showing up on
        # the same line as the step name.
        passed_tests_step.presentation.step_text = ''.join(
            '\n' + test_name for test_name in passed_tests)

  def get_output(self, output_path):
    """Returns the contents of the file at output_path.

    The output_path should be a relative path to the results_dir.
    """
    if output_path not in self._outputs:
      self._outputs[output_path] = self._api.file.read_text(
          'read %s' % output_path,
          self.results_dir.join(output_path),
          test_data='output')
    return self._outputs[output_path]

  def upload_results(self, gcs_bucket, upload_to_catapult):
    """Upload select test results (e.g., coverage data) to a given GCS bucket."""
    assert gcs_bucket
    with self._api.step.nest('upload %s test results' % self._env_name):
      if self.summary:
        # Save the summary JSON to the test shard output dir so it gets
        # uploaded to GCS for easy access by e.g. Dachsiaboard.
        summary_path = self.output_dir.join(TEST_SUMMARY_JSON)
        assert not self._api.path.exists(summary_path), (
            'test output files should not be named %s' % TEST_SUMMARY_JSON)
        self._api.file.write_json('write %s' % TEST_SUMMARY_JSON, summary_path,
                                  self.summary)

      self._upload_outputs(gcs_bucket)

      if upload_to_catapult:
        self._api.upload.test_outputs_to_catapult(self.output_dir)

  def _upload_outputs(self, gcs_bucket):
    self._api.upload.directory_to_gcs(
        source=self.output_dir,
        bucket=gcs_bucket,
        # Namespace to avoid collision across shards and attempts.
        subpath='%s/%s' % (self._env_name, self.swarming_task_id),
    )

  def raise_failures(self):
    """Raises a step failure if there were test failures."""
    if not self.summary:
      # Halt with step failure if summary file is missing.
      raise self._api.step.StepFailure(
          'Test summary JSON not found, see symbolized log for details')

    failed_tests = self.failed_test_outputs.keys()
    if failed_tests:
      # Halt with a step failure.
      raise self._api.step.StepFailure('Test failure(s): ' +
                                       ', '.join(failed_tests))

    # Check serial log for failure messages
    # TODO(9936): Replace with running binary tool once created.
    fail_strings = ['DEVICE SUSPEND TIMED OUT', 'ASSERT FAILED']
    log_path = self.output_dir.join(self._api.testing_requests.SERIAL_LOG_NAME)
    self._api.path.mock_add_paths(log_path)
    if self._api.path.exists(log_path):
      self._check_log_for_failures(log_path, fail_strings)

  def _check_log_for_failures(self, log_path, fail_strings):
    """Checks for fail strings in log and fails accordingly."""
    log_name = self._api.path.basename(log_path)
    with self._api.step.nest('check log %s:%s' %
                             (self._env_name, log_name)) as check_log_step:
      contents = self._api.file.read_text(
          'read %s' % log_name,
          log_path,
          test_data='extra log contents',
      ).get_result()
      for fail_str in fail_strings:
        if fail_str in contents:
          check_log_step.presentation.logs[log_name] = contents.splitlines()
          raise self._api.step.StepFailure(
              'Found failure string in log %s: %s' % (log_name, fail_str))


def create_task(api, *args, **kwargs):
  """Create a Task object.

  The base class of the class is inside the api object, so it can't be
  top-level or otherwise defined at module load time. Defining it in this
  function as an alternative.

  For full args list see Task.__init__ a few lines down.
  """

  class Task(api.swarming_retry.TriggeredTask):

    def __init__(self, api, name, request, uses_legacy_qemu, targets_fuchsia,
                 symbolize_tool, llvm_symbolizer, tests,
                 debug_symbol_gcs_bucket, *args, **kwargs):
      super(Task, self).__init__(
          api=api, name=name, request=request, *args, **kwargs)
      self._uses_legacy_qemu = uses_legacy_qemu
      self._targets_fuchsia = targets_fuchsia
      self._symbolize_tool = symbolize_tool
      self._llvm_symbolizer = llvm_symbolizer
      self._tests = tests
      self._debug_symbol_gcs_bucket = debug_symbol_gcs_bucket

      # Test shards with the 'multiplied:' prefix come from
      # tools/integration/testsharder/shard.go in fuchsia.git. They were
      # specifically created to run a test or set of tests many times to look
      # for flakes. It doesn't make sense to retry these when they fail--the
      # goal is to see if they fail not to get them to pass.
      if name.startswith('multiplied:'):
        self.max_attempts = 1

    def process_result(self):
      """Unpacks the results archive produced by a test shard."""

      attempt = self.attempts[-1]
      assert attempt.result
      result = attempt.result

      if result.isolated_outputs:
        attempt.task_outputs_link = result.isolated_outputs.url

      if result.state == self._api.swarming.TaskState.TIMED_OUT:
        attempt.failure_reason = 'timed out'

      attempt.test_results = None

      with self._api.step.nest(result.name):
        attempt.symbolizer_output = result.output_dir.join(
            self._api.symbolize.LOG)
        attempt.symbolizer_json_output = self._api.path['cleanup'].join(
            '%s-%d-%s' %
            (self.name, attempt.index, self._api.symbolize.OUTPUT_JSON))
        # Figure out what happened to the swarming task.
        if result.output:
          # Fuchsia crashes have to be symbolized on the host.
          if self._targets_fuchsia:
            attempt.logs['symbolized log'] = self._api.symbolize(
                symbolize_tool=self._symbolize_tool,
                debug_symbol_gcs_bucket=self._debug_symbol_gcs_bucket,
                llvm_symbolizer=self._llvm_symbolizer,
                data=result.output,
                symbolizer_output=attempt.symbolizer_output,
                json_output=attempt.symbolizer_json_output,
            )
          # Non-Fuchsia should already be symbolized, and attempting to use
          # the symbolizer may fail, if e.g. it was built on Mac and this is
          # running on Linux.
          else:
            attempt.logs['symbolized log'] = result.output

        if 'KERNEL PANIC' in result.output:
          attempt.failure_reason = 'KERNEL PANIC'  # pragma: no cover

        self._check_logs_for_failures(attempt)

        if result.success:
          self._process_outputs(attempt)

    def _process_outputs(self, attempt):
      """Reads the test results and output files of a swarming TaskResult.

      Sets attempt.test_results if successful.

      Args:
        attempt (swarming_retry.Attempt): the attempt to process
      """

      assert attempt.result
      result = attempt.result

      # Extract results if the task was not subject to an infra failure;
      # otherwise, a step failure will be raised on exiting the
      # defer_results() scope.
      attempt.test_results_archive = None
      for relative_path, absolute_path in sorted(result.outputs.iteritems()):
        if relative_path in [
            self._api.testing_requests.TEST_RESULTS_ARCHIVE_NAME,
            self._api.testing_requests.TEST_RESULTS_MINFS_NAME
        ]:
          attempt.test_results_archive = absolute_path

      assert attempt.test_results_archive, (
          'test archive not found amongst outputs of task %s' % result.name)
      self._parse_test_results(attempt)

      attempt.logs[TEST_SUMMARY_JSON] = attempt.test_results.summary_lines

      # Delete the archive so it doesn't get uploaded with the other files in
      # the swarming task's output directory.
      self._api.file.remove(
          'remove %s' % self._api.path.basename(attempt.test_results_archive),
          attempt.test_results_archive)

    def _parse_test_results(self, attempt):
      """Parse test results from attempt into a FuchsiaTestResults object.

      Args:
        attempt (swarming_retry.Attempt): the attempt to parse
      """
      assert attempt.result
      result = attempt.result

      results_dir = self._api.testing.results_dir_on_host.join(result.id)
      # pylint: disable=protected-access
      self._api.testing._extract_test_results_archive(
          step_name='extract',
          archive_path=attempt.test_results_archive,
          directory=results_dir,
          is_minfs=self._uses_legacy_qemu,
      )
      # pylint: enable=protected-access

      attempt.test_results = FuchsiaTestResults(
          from_fuchsia=self._targets_fuchsia,
          results_dir=results_dir,
          swarming_task_id=attempt.task_id,
          symbolizer_json_output=attempt.symbolizer_json_output,
          env_name=result.name,
          tests=self._tests,
          legacy_qemu=self._uses_legacy_qemu,
          api=api,
          symbolizer_output=attempt.symbolizer_output,
          output_dir=result.output_dir,
      )

      failed_tests = attempt.test_results.failed_test_outputs
      if failed_tests:
        attempt.failure_reason = '%d test(s) failed' % len(failed_tests)

    def _check_logs_for_failures(self, attempt):
      """Check for failure strings in logs.

      Args:
        attempt (swarming_retry.Attempt): the attempt to check for logs in
      """
      # Check serial log for failure messages
      # TODO(9936): Replace with running binary tool once created.
      fail_strings = ['DEVICE SUSPEND TIMED OUT', 'ASSERT FAILED']
      log_path = attempt.result.output_dir.join(
          self._api.testing_requests.SERIAL_LOG_NAME)
      self._api.path.mock_add_paths(log_path)
      if self._api.path.exists(log_path):
        log_name = self._api.path.basename(log_path)
        with self._api.step.nest('check log %s' % log_name) as presentation:
          contents = self._api.file.read_text('read', log_path)
          for fail_str in fail_strings:
            if fail_str in contents:
              presentation.logs[log_name] = contents.splitlines()
              presentation.status = self._api.step.FAILURE
              presentation.step_summary_text = 'found "%s"' % fail_str
              attempt.failure_reason = ('found "%s" in %s' %
                                        (fail_str, log_name))

    def present_status(self, parent_step, attempt, **kwargs):
      """Present an Attempt while showing progress in launch/collect step.

      Args:
        parent_step (Step): will always be 'passed tasks' or 'failed tasks'
        attempt (Attempt): the Attempt to present
      """
      del kwargs, parent_step  # Unused.
      with api.step.nest('%s (%s)' % (self.name, attempt.name)) as presentation:
        self._present(
            presentation,
            attempt,
            show_failures_in_red=False,
            show_passed=False)

    def present_attempt(self, _, attempt, category=None):
      """Present an Attempt when summarizing results at the end of the run.

      Args:
        attempt (Attempt): the Attempt to present
        category (str): the group of tasks ('passes', 'failures', or
          'flakes') that this attempt should be presented under
      """
      show_failures_in_red = True
      # The 'passes' category includes all attempts of all tasks that
      # eventually passed, so it includes some failures. Show those in
      # green so people don't get confused and think the overall task
      # failed.
      # TODO(fxb/36647) after this bug is fixed show these steps in
      # red, but show parent steps of those in green.
      if category == 'passes':
        show_failures_in_red = False

      name = '%s (%s)' % (attempt.name, 'pass' if attempt.success else 'fail')
      with api.step.nest(name) as presentation:
        if show_failures_in_red and not attempt.success:
          presentation.status = self._api.step.FAILURE
        self._present(
            presentation,
            attempt,
            show_failures_in_red=show_failures_in_red,
            show_passed=True,
        )

    def _present(self, presentation, attempt, show_failures_in_red,
                 show_passed):
      """Present an Attempt.

      Choosing to do largely the same thing for both kinds of presentations.

      Args:
        presentation (StepPresentation): where to present the attempt info
        attempt (api.swarming_retry.Attempt): object to present
        show_failures_in_red (bool): show failures in red (for final
            'flakes' and 'failures' steps) or not (for 'launch/collect'
            progress and 'passes' steps)
        show_passed (bool): show the names of passed tests (only done for
            the end)

      Note: the 'passes' step can have failures underneath it because the
      first attempt can fail but the retry passed.
      """
      if attempt.result.duration_secs:
        presentation.step_text = nice_time(attempt.result.duration_secs)

      presentation.presentation.links['swarming task'] = attempt.task_ui_link
      if attempt.task_outputs_link:
        presentation.links['task outputs'] = attempt.task_outputs_link

      if attempt.failure_reason:
        presentation.step_summary_text = attempt.failure_reason

      for log, data in attempt.logs.iteritems():
        presentation.logs[log] = data

      if attempt.test_results:
        test_results = attempt.test_results

        # Log the contents of each output file mentioned in the summary.
        # Note this assumes the outputs are all valid UTF-8 (See fxb/9500).
        for name, path in test_results.summary.get('outputs', {}).iteritems():
          presentation.logs[name] = test_results.get_output(path).split('\n')

        test_results.present_tests(
            show_failures_in_red=show_failures_in_red, show_passed=show_passed)

      for log_name in [
          self._api.testing_requests.SYSLOG_NAME,
          self._api.testing_requests.SERIAL_LOG_NAME
      ]:
        if log_name in attempt.result.outputs:
          self._present_output_file(
              name=log_name,
              path=attempt.result.outputs[log_name],
              step=presentation)

    def _present_output_file(self, name, path, step):
      """Records file contents to the test results step's presentation."""
      contents = self._api.file.read_text(
          'read %s' % name,
          path,
          test_data='extra log contents',
      )
      step.presentation.logs[name] = contents.splitlines()

  return Task(*args, api=api, **kwargs)


class _ShardedTestRunner(object):
  """Handles running and analyzing tests that have been split into shards."""

  def __init__(self, api, collect_timeout, debug_symbol_gcs_bucket,
               llvm_symbolizer, max_attempts, rerun_budget_secs,
               swarming_output_dir, symbolize_tool, shard_requests):
    self._api = api
    self._collect_timeout = collect_timeout
    self._max_attempts = max_attempts
    launch_deadline_time = None
    self._present = True
    if rerun_budget_secs:
      assert max_attempts == 1, (
          'If rerun_budget_secs is set, max_attempts should be set to 1')
      # Presenting results is currently slow due to inefficiencies in
      # the recipe engine. For builds that have rerun_budget_secs enabled,
      # we only care about consuming the data after it's been uploaded, not
      # from the Milo build page, so skip presentation.
      self._present = False
      launch_deadline_time = self._api.time.time() + rerun_budget_secs
    self._swarming_output_dir = swarming_output_dir

    self.tasks = []
    for shard_request in shard_requests:
      uses_legacy_qemu = any(tag.lower() == 'uses_legacy_qemu:true'
                             for tag in shard_request.task_request.tags)
      targets_fuchsia = shard_request.task_request[0].dimensions.get(
          'os', '').lower() not in ('linux', 'mac')
      self.tasks.append(
          create_task(
              api=self._api,
              name=shard_request.task_request.name,
              request=shard_request.task_request,
              symbolize_tool=symbolize_tool,
              llvm_symbolizer=llvm_symbolizer,
              tests=shard_request.shard.tests,
              debug_symbol_gcs_bucket=debug_symbol_gcs_bucket,
              uses_legacy_qemu=uses_legacy_qemu,
              targets_fuchsia=targets_fuchsia,
              launch_deadline_time=launch_deadline_time,
          ))

  def run_tests(self):
    """Runs all test shards and returns a Task object for each."""
    # TODO(fxb/35021) use context manager.
    self._api.swarming_retry.run_tasks(
        tasks=self.tasks,
        collect_output_dir=self._swarming_output_dir,
        max_attempts=self._max_attempts,
        collect_timeout=self._collect_timeout,
    )

    if self._present:
      self._api.swarming_retry.present_tasks(tasks=self.tasks)

    return self.tasks

  def raise_failures(self):
    self._api.swarming_retry.raise_failures(self.tasks)


class FuchsiaTestApi(recipe_api.RecipeApi):
  """API for running tests and processing test results."""

  FuchsiaTestResults = FuchsiaTestResults

  def __init__(self, *args, **kwargs):
    super(FuchsiaTestApi, self).__init__(*args, **kwargs)
    self._test_runner = None

  def _analyze_test_results(self, test_results, presentation=None):
    """Analyzes test results represented by FuchsiaTestResults objects

    Logs individual test results in separate steps.

    Args:
      test_results (FuchsiaTestResults): Fuchsia test result object
      presentation (dict or None): A particular step's presentation on which to log
        test result outputs; if not provided, that of the active result will be
        used.
    """
    presentation = presentation or self.m.step.active_result.presentation

    if not test_results.summary:
      return

    # Log the summary file's contents.
    presentation.logs[TEST_SUMMARY_JSON] = test_results.summary_lines

    # Log the contents of each output file mentioned in the summary.
    # Note this assumes the outputs are all valid UTF-8 (See fxb/9500).
    for output_name, output_path in test_results.summary.get('outputs',
                                                             {}).iteritems():
      output_str = test_results.get_output(output_path)
      presentation.logs[output_name] = output_str.split('\n')

    test_results.present_tests(show_failures_in_red=True, show_passed=True)

  def process_coverage(self, covargs_path, test_results, ids_txt, llvm_profdata,
                       llvm_cov, gcs_bucket):
    output_dir = self.m.path['cleanup'].join('coverage')

    cmd = [
        covargs_path,
        '-level',
        COVARGS_LOG_LEVEL,
        '-json-output',
        self.m.json.output(name=COVARGS_OUTPUT_JSON),
        '-output-dir',
        output_dir,
        '-llvm-profdata',
        llvm_profdata,
        '-llvm-cov',
        llvm_cov,
        '-ids',
        ids_txt,
    ]

    for result in test_results:
      cmd.extend([
          '-summary', result.results_dir.join(TEST_SUMMARY_JSON),
          '-symbolize-dump', result.symbolizer_json_output,
      ])  # yapf: disable

    self.m.step('covargs', cmd)

    # TODO: move this into gsutil module/deduplicate this with other GCS logic
    dst = 'builds/%s/coverage' % self.m.buildbucket_util.id
    step_result = self.m.gsutil.rsync(
        name='upload coverage',
        src=output_dir,
        bucket=gcs_bucket,
        dst=dst,
        recursive=True,
        gzip_exts=['html'],
        options={
            'parallel_process_count': self.m.platform.cpu_count,
            'parallel_thread_count': 1,
        },
        multithreaded=True)
    step_result.presentation.links['index.html'] = self.m.gsutil._http_url(
        gcs_bucket, self.m.gsutil.join(dst, 'index.html'), True)

  @property
  def results_dir_on_host(self):
    """The directory on host to which host and target test results will be written.

    Target test results will be copied over to this location and host test
    results will be written here. Host and target tests on should write to
    separate subdirectories so as not to collide.
    """
    return self.m.path['cleanup'].join('test_results')

  def _extract_test_results_archive(self,
                                    step_name,
                                    archive_path,
                                    is_minfs=False,
                                    directory=None):
    """Extracts test results from an archive.

    Args:
      step_name (str): The name of the step.
      archive_path (Path): The path to the archive which contains test results.
      is_minfs (bool): Whether the archive in question is a minfs image
        containing QEMU test results. If false, then the archive is assumed to
        be a tar file.
      directory (Path): The directory to extract the archive contents into.
    """
    self.m.file.ensure_directory('create test results dir', directory)
    if is_minfs:
      self.m.minfs.copy_image(
          step_name=step_name,
          image_path=archive_path,
          out_dir=directory,
      )
    else:
      self.m.tar.extract(
          step_name=step_name,
          path=archive_path,
          directory=directory,
      )

    # This is only needed for the recipe tests. file.listdir() doesn't mock the
    # existence of the paths it returns, so we must add it separately.
    # We add summary_path because we check for its existence in
    # FuchsiaTestResults._parse_summary().
    summary_path = directory.join(TEST_SUMMARY_JSON)
    outputs = self.m.file.listdir('get extracted files', directory)
    if summary_path in outputs:
      self.m.path.mock_add_paths(summary_path)

  def deprecated_test(self, *args, **kwargs):
    """Tests a Fuchsia build on the specified device with retries.

    Expects the build and artifacts to be at the same place they were at
    the end of the build.

    Args (see _launch_collect_process_funcs for other args):
      max_attempts (int): The tests will be run repeatedly until either
        max_attempts is hit or all tests pass.

    Returns:
      A `FuchsiaTestResults` object corresponding to the last test attempt.
    """
    # Ideally this method's arguments would look like
    # (self, *args, max_attempts=0, **kwargs)
    # but Python 2 doesn't allow default keyword args after variable-length
    # positional *args :(
    max_attempts = kwargs.pop('max_attempts', 0)
    if not max_attempts:
      max_attempts = self.m.swarming_retry.DEFAULT_MAX_ATTEMPTS

    launch, collect, process = self._launch_collect_process_funcs(
        *args, **kwargs)

    test_results = None
    final_exception = None

    # TODO(olivernewman): status='last' should cause this step to turn green as
    # long as the *last* test attempt is green, but this isn't working, at
    # least not for led jobs (if the first attempt fails at the second passes,
    # the build is marked as a failure). Figure out whether this will be
    # resolved by using luci_runner.
    with self.m.step.nest('run tests', status='last'):
      for i in range(max_attempts):
        with self.m.step.nest('attempt %d' % i) as attempt_presentation:
          task_result = collect(launch())
          try:
            test_results = process(
                task_result, presentation=attempt_presentation)
          except self.m.step.StepFailure as e:
            final_exception = e
          else:
            final_exception = None
            if test_results.passed:
              attempt_presentation.step_text = 'passed'
              break
            else:
              failed_count = len(test_results.failed_test_outputs)
              attempt_presentation.step_text = ('%d test(s) failed' %
                                                failed_count)

      if final_exception:
        raise final_exception  # pylint: disable=raising-bad-type

      return test_results

  def deprecated_test_async(self, *args, **kwargs):
    """Launches a swarming task to run Fuchsia tests.

    Returns:
      A function that, when invoked, waits for the tests to complete and
      returns a `FuchsiaTestResults` object representing the completed test.
    """
    launch, collect, process = self._launch_collect_process_funcs(
        *args, **kwargs)
    request_metadata = launch()
    return lambda: process(collect(request_metadata))

  def _launch_collect_process_funcs(
      self,
      debug_symbol_gcs_bucket,
      device_type,
      orchestration_inputs,
      overwrite_summary=True,
  ):
    """Returns 3-tuple of functions to launch Fuchsia tests, wait for them to
    complete, and process the results.

    Args:
      debug_symbol_gcs_bucket (str): A GCS bucket hosting debug symbols.
      device_type (str): Used as swarming device_type dimension.
      orchestration_inputs (TestOrchestrationInputs): the bits of data
          needed to orchestrate testing.
      overwrite_summary (bool): Whether to overwrite the name and label
        fields in summary.json based on tests.json. This should *only* be
        used by fuchsia_perf; do NOT add any new dependencies on this.
        TODO(fxb/10410): remove this entirely after fuchsia_perf is dead.

    Returns:
      A tuple of functions:
      - `launch`, which takes no arguments and launches a swarming task to
        run tests against the given build artifacts. Returns a
        `TaskRequestMetadata` object.
      - `collect`, which takes the `TaskRequestMetadata` object returned by
        launch` (and, optionally, a `StepPresentation` object to add logs
        to). It blocks until the task is complete and returns a swarming
        `TaskResult`.
      - `process`, which processes the results and returns a
        `FuchsiaTestResults` object representing the completed tests.
    """
    task = orchestration_inputs.shard_requests[0].task_request
    # This directory gets passed into `collect()`, but unfortunately the
    # `output_dir` attribute of the `TaskResult` returned by `collect()` is
    # a subdirectory of this output dir (to ensure that different tasks'
    # outputs do not collide when calling `api.swarming.collect()` with many
    # tasks). So we make this variable in-scope for all three functions so that
    # `process()` can use it as the output dir for the test results object.
    output_dir = self.m.path.mkdtemp('swarming')

    def launch():
      with self.m.context(infra_steps=True):
        return self.m.swarming.trigger(
            'trigger 1 task', [task])

    def collect(request_metadata):
      with self.m.context(infra_steps=True):
        results = self.m.swarming.collect(
            'collect', tasks=request_metadata, output_dir=output_dir)
        assert len(results) == 1, 'len(%s) != 1' % repr(results)
        return results[0]

    def process(task_result, presentation=None):
      symbolizer_output = output_dir.join(self.m.symbolize.LOG)
      symbolizer_json_output = self.m.path['cleanup'].join(
          self.m.symbolize.OUTPUT_JSON)
      with self.m.step.nest('task results'):
        self._analyze_task_result(
            result=task_result,
            debug_symbol_gcs_bucket=debug_symbol_gcs_bucket,
            symbolize_tool=orchestration_inputs.symbolize_tool,
            llvm_symbolizer=orchestration_inputs.llvm_symbolizer,
            symbolizer_output=symbolizer_output,
            symbolizer_json_output=symbolizer_json_output,
            presentation=presentation,
        )

      with self.m.context(infra_steps=True):
        # result.outputs contains the file outputs produced by the Swarming
        # task, returned via isolate. It's a mapping of the 'name' of the
        # output, represented as its relative path within the isolated it
        # was returned in, to a Path object pointing to its location on the
        # local disk. For each of the above tasks, there should be exactly
        # one output.
        serial_log_name = self.m.testing_requests.SERIAL_LOG_NAME
        if serial_log_name in task_result.outputs:
          serial_log = task_result.outputs.pop(serial_log_name)

          serial_log_contents = self.m.file.read_text(
              'read serial.txt', serial_log, test_data=[])
          serial_presentation = (
              presentation or self.m.step.active_result.presentation)
          serial_presentation.logs[serial_log_name] = (
              serial_log_contents.splitlines())
        assert len(task_result.outputs) == 1, 'len(%s) != 1' % repr(
            task_result.outputs)
        archive_name, archive_path = task_result.outputs.items()[0]

        test_results_dir = self.results_dir_on_host.join(
            'target', task_result.id)
        # _extract_test_results_archive needs minfs_path to be set.
        # This is kinda ugly. It'd be better to pass this in as an argument.
        self.m.minfs.minfs_path = orchestration_inputs.minfs
        self._extract_test_results_archive(
            step_name='extract results',
            is_minfs=self.m.emu.is_emulator_type(device_type),
            archive_path=archive_path,
            # Write test results to a subdirectory of |results_dir_on_host|
            # so as not to collide with host test results.
            directory=test_results_dir,
        )

        # Remove the archive file so it doesn't get uploaded to GCS.
        self.m.file.remove('remove %s' % archive_name, archive_path)

        test_list = self.m.file.read_json(
            'read tests.json', orchestration_inputs.tests_file, test_data=[])
        tests = [
            self.m.testsharder.Test.from_jsonish(t['test']) for t in test_list
        ]

      with self.m.step.nest('all test results'):
        test_results = self.FuchsiaTestResults(
            from_fuchsia=True,
            results_dir=test_results_dir,
            swarming_task_id=task_result.id,
            symbolizer_json_output=symbolizer_json_output,
            env_name=task_result.name,
            tests=tests,
            legacy_qemu=self.m.emu.is_emulator_type(device_type),
            api=self.m,
            symbolizer_output=symbolizer_output,
            output_dir=output_dir,
            overwrite_summary=overwrite_summary,
        )
        self._analyze_test_results(test_results, presentation=presentation)
        return test_results

    return launch, collect, process

  def final_results(self, tasks):
    """Returns the FuchsiaTestResults from the final attempt of each Task."""
    return [
        t.attempts[-1].test_results
        for t in tasks
        if t.attempts[-1].test_results
    ]

  def test_in_shards(self,
                     collect_timeout_secs,
                     debug_symbol_gcs_bucket,
                     max_attempts,
                     orchestration_inputs,
                     rerun_budget_secs=None):
    """Tests a Fuchsia build by sharding.

    Expects the build and artifacts to be at the same place they were at
    the end of the build.

    Args:
      collect_timeout_secs (int): Amount of time to wait for tasks to complete.
      debug_symbol_gcs_bucket (str): A GCS bucket hosting debug symbols.
      max_attempts (int): Maximum number of attempts before marking a shard
        as failed.
      orchestration_inputs (TestOrchestrationInputs): Build artifacts needed
        for testing.
      rerun_budget_secs (int): Will run tests repeatedly until this budget is consumed.
        If set, max_attempts is ignored.


    Returns:
      A list of FuchsiaTestResults objects representing the completed test
      tasks that were not subject to an infra failure.
    """
    # If no shards have been provided, then we have successfully run the empty
    # set of tests.
    if not orchestration_inputs.shard_requests:
      return []

    self.m.minfs.minfs_path = orchestration_inputs.minfs
    collect_timeout = None
    if collect_timeout_secs:
      collect_timeout = '%ds' % collect_timeout_secs

    self._test_runner = _ShardedTestRunner(
        self.m,
        collect_timeout=collect_timeout,
        debug_symbol_gcs_bucket=debug_symbol_gcs_bucket,
        llvm_symbolizer=orchestration_inputs.llvm_symbolizer,
        max_attempts=max_attempts,
        rerun_budget_secs=rerun_budget_secs,
        swarming_output_dir=self.m.path.mkdtemp('swarming'),
        symbolize_tool=orchestration_inputs.symbolize_tool,
        shard_requests=orchestration_inputs.shard_requests,
    )
    return self._test_runner.run_tests()

  def raise_failures(self):
    if self._test_runner:
      self._test_runner.raise_failures()

  def _analyze_task_result(
      self,
      result,
      symbolize_tool,
      llvm_symbolizer,
      debug_symbol_gcs_bucket,
      symbolizer_output,
      symbolizer_json_output,
      presentation=None,
  ):
    """Analyzes a swarming.TaskResult and reports results as a step.

    Args:
      result (api.swarming.TaskResult): The swarming task result to analyze.
      symbolize_tool (Path): The path to the symbolize tool.
      llvm_symbolizer (Path): The path to the llvm_symbolizer tool.
      debug_symbol_gcs_bucket (str): A GCS bucket hosting debug symbols.
      symbolizer_output (Path): A path to a file to write the symbolizer's
        stdout.
      symbolizer_json_output (Path): A path to a file to write the
        symbolizer's JSON output.
      presentation (StepPresentation or None): The step presentation to
        attach logs to. Defaults to `active_result.presentation`.

    Raises:
      A StepFailure if a kernel panic is detected or an InfraFailure if the
      swarming task failed for a different reason.
    """
    presentation = presentation or self.m.step.active_result.presentation
    if result.output:
      # Always symbolize the result output if present in this case.
      presentation.logs['symbolized log'] = self.m.symbolize(
          symbolize_tool=symbolize_tool,
          debug_symbol_gcs_bucket=debug_symbol_gcs_bucket,
          llvm_symbolizer=llvm_symbolizer,
          data=result.output,
          symbolizer_output=symbolizer_output,
          json_output=symbolizer_json_output,
          presentation=presentation)

    # A kernel panic may be present in the logs even if the task timed out, so
    # check for that first.
    if 'KERNEL PANIC' in result.output:
      presentation.step_text = 'kernel panic'
      presentation.status = self.m.step.FAILURE
      raise self.m.step.StepFailure(
          'Found kernel panic. See symbolized output for details.')

    if result.isolated_outputs:
      presentation.links['test outputs'] = result.isolated_outputs.url

    try:
      result.analyze()
    except self.m.step.StepFailure:
      self._present_task_errors(result, presentation)
      raise

  def _present_task_errors(self, task_result, presentation):
    """Updates text and status of the given step to reflect test task errors."""
    # If the task is in an unknown state or completed, but the executed command returned
    # a non-zero exit code, this points to a tooling failure.
    if task_result.state is None or task_result.state == self.m.swarming.TaskState.COMPLETED:
      text = 'tooling failure'  # pragma: no cover
    else:
      text = (task_result.state.name).replace('_', ' ').lower()
    presentation.step_text = text

    # Report timeouts as red, not purple, as it is likelier that the task is
    # timing out due to a bug in the system under test.
    if task_result.state == self.m.swarming.TaskState.TIMED_OUT:
      status = self.m.step.FAILURE  # pragma: no cover
    else:
      status = self.m.step.EXCEPTION
    presentation.status = status


def product(nums):
  return reduce(operator.mul, nums)


def nice_time(seconds):
  """Returns a human-readable duration given a number of seconds.

  For example: 3605 seconds => '1h 5s'
  """
  units = [
      ('d', 24),
      ('h', 60),
      ('m', 60),
      ('s', 1),
  ]

  result_parts = []
  divisor = product(unit_size for _, unit_size in units)
  for unit_abbrev, unit_size in units:
    count, seconds = divmod(seconds, divisor)
    if count > 0:
      result_parts.append('%d%s' % (count, unit_abbrev))
    divisor /= unit_size

  return ' '.join(result_parts)