# 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 attr
import collections

from recipe_engine import recipe_api
from recipe_engine.config_types import Path

from RECIPE_MODULES.fuchsia.swarming_retry import api as swarming_retry_api
from RECIPE_MODULES.fuchsia.utils import cached_property, nice_duration, pluralize

# Name of the file produced by a testing task that contains results and outputs
# of the tests that were run.
TEST_SUMMARY_JSON = "summary.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 have been moved
        into (out of task_output_dir).
      task_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_bot_id (str): The ID of the swarming bot that ran the task.
      swarming_task_id (str): The ID of the task that ran these tests.
      env_name (str): The name of the environment that these tags ran in.
      shard_name (str): The name of the task that ran these tests.
      api (RecipeApi): The api to use for accessing recipe modules from this
        object.
      summary_bytes (bytes): The contents of TEST_SUMMARY_JSON.
      is_multiplied (bool): Whether the test results are for a multiplier
        shard.
    """

    from_fuchsia = attr.ib(type=bool)
    results_dir = attr.ib(type=Path)
    task_output_dir = attr.ib(type=Path)
    swarming_bot_id = attr.ib(type=str)
    swarming_task_id = attr.ib(type=str)
    env_name = attr.ib(type=str)
    _shard_name = attr.ib(type=str)
    _api = attr.ib(type=recipe_api.RecipeApi)
    _summary_bytes = attr.ib(type=bytes)
    _is_multiplied = attr.ib(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 `_separate_tests_by_result()`, not a parameter to __init__.
    _pass_name_to_tests = attr.ib(factory=dict, init=False)
    _fail_name_to_tests = attr.ib(factory=dict, 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. Any other value is considered to
    # represent a failure.
    _TEST_RESULT_PASS = "PASS"
    _TEST_RESULT_FAIL = "FAIL"
    _TEST_RESULT_TIMEOUT = "ABORT"
    _TEST_RESULT_SKIP = "SKIP"

    @cached_property
    def summary(self):
        """The parsed summary file as a Dict or {} if missing."""
        if not self._summary_bytes:
            return {}

        try:
            return self._api.json.loads(self._summary_bytes)
        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])
            )

    @property
    def passed_tests(self):
        """All entries in |self._outputs| for tests that passed."""
        self._separate_tests_by_result()
        passed_tests = []
        for tests in self._pass_name_to_tests.values():
            passed_tests.extend(tests)
        return passed_tests

    @property
    def failed_tests(self):
        """All entries in |self._outputs| for tests that failed."""
        self._separate_tests_by_result()
        failed_tests = []
        for name, tests in self._fail_name_to_tests.items():
            if self._is_multiplied or name not in self._pass_name_to_tests:
                failed_tests.extend(tests)
        return failed_tests

    @property
    def flaked_tests(self):
        """All entries in |self._outputs| for tests that flaked."""
        self._separate_tests_by_result()
        flaked_tests = []
        for name, tests in self._fail_name_to_tests.items():
            if not self._is_multiplied and name in self._pass_name_to_tests:
                flaked_tests.extend(tests)
        return flaked_tests

    def _separate_tests_by_result(self):
        """Separates entries in |self._outputs| into maps based on test result.

        Passed tests will be stored in self._pass_name_to_tests and failed
        tests will be stored in self._fail_name_to_tests.
        """
        if not self._pass_name_to_tests or not self._fail_name_to_tests:
            # Using OrderedDict() will keep the tests in the same order as they
            # ran in. This doesn't really matter for displaying results, but it
            # keeps the expectation files from changing by maintaining the same
            # order of tests.
            self._pass_name_to_tests = collections.OrderedDict()
            self._fail_name_to_tests = collections.OrderedDict()
            for test in self.summary.get("tests") or ():
                if test["result"] == self._TEST_RESULT_PASS:
                    tests = self._pass_name_to_tests.get(test["name"], [])
                    tests.append(test)
                    self._pass_name_to_tests[test["name"]] = tests
                else:
                    tests = self._fail_name_to_tests.get(test["name"], [])
                    tests.append(test)
                    self._fail_name_to_tests[test["name"]] = tests

    def present_tests(self, show_failures_in_red, show_passed):
        def show_failed_tests(tests):
            for test in tests:
                if test["result"] == self._TEST_RESULT_TIMEOUT:
                    status = "timed out"
                elif test["result"] == self._TEST_RESULT_SKIP:
                    status = "skipped"
                else:
                    status = "failed"

                step = self._api.step.empty("%s: %s" % (status, test["name"]))
                for output_file in test["output_files"]:
                    # Failed test output files should have been read into
                    # self._outputs by Task.process_result().
                    output = self._outputs[output_file]
                    log_name = self._api.path.basename(output_file)
                    if not output.strip():
                        log_name += " (empty)"
                    step.presentation.logs[log_name] = output.splitlines()

                # If present, show the GN label to aid debugging.
                label = test.get("gn_label")
                if label:
                    step.presentation.step_summary_text = label
                if show_failures_in_red:
                    step.presentation.status = self._api.step.FAILURE

        show_failed_tests(self.failed_tests)
        show_failed_tests(self.flaked_tests)

        # 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_step.presentation.step_summary_text = pluralize(
                "passed test", self.passed_tests
            )
            if show_passed:
                passed_tests_step.presentation.step_text = self._passed_tests_text(
                    self.passed_tests
                )

    def _passed_tests_text(self, tests):
        """Format test names for presentation in Milo.

        If a single test ran many times in a row, we'll compress those runs
        into a single line with a run count, to improve readability.
        """
        test_names = [t["name"] for t in tests]
        lines = []
        current_test_count = 0
        for i, test_name in enumerate(test_names):
            current_test_count += 1
            # If we've reached the last test OR the next test has a different
            # name then we've reached the end of a repeated streak of running a
            # single test, so add a line of text for that streak.
            if i + 1 >= len(test_names) or test_names[i + 1] != test_name:
                # Surround test names in backticks so that Milo's markdown
                # renderer doesn't try to parse them as markdown.
                line = "`%s`" % test_name
                if current_test_count > 1:
                    line += " (%d runs)" % current_test_count
                lines.append(line)
                current_test_count = 0

        # Start with a newline to prevent the first test from showing up on
        # the same line as the step name.
        return "".join("\n" + l for l in lines)

    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),
                # We're returning the log contents to be attached to other
                # steps, so no need to log them here.
                include_log=False,
            )
        return self._outputs[output_path]

    def upload_results(
        self,
        gcs_bucket,
        upload_to_catapult,
        orchestration_inputs,
        resultdb_base_variant,
        resultdb_tags,
    ):
        """Upload select test results (e.g., coverage data) to a given GCS bucket."""
        assert gcs_bucket
        with self._api.context(infra_steps=True), self._api.step.nest(
            "upload %s test results" % self._shard_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.task_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
                )
                if self._api.resultdb.enabled and orchestration_inputs.resultdb:
                    self._upload_to_resultdb(
                        summary_path,
                        orchestration_inputs,
                        resultdb_base_variant,
                        resultdb_tags,
                    )
            snapshot_zip = self.results_dir.join(
                self._api.testing_requests.SNAPSHOT_NAME
            )
            self._api.path.mock_add_paths(snapshot_zip)
            if self._api.path.exists(snapshot_zip):
                self._api.file.move(
                    "move snapshot to output dir",
                    snapshot_zip,
                    self.task_output_dir.join(self._api.testing_requests.SNAPSHOT_NAME),
                )

            # If an output was important enough to present (meaning it's in _outputs),
            # we should upload it so it can be shown in Sponge, etc.
            with self._api.step.nest("move test outputs"):
                for output in sorted(self._outputs):
                    dest = self.task_output_dir.join(output)
                    self._api.file.ensure_directory(
                        "mkdir", self._api.path.dirname(dest)
                    )
                    self._api.file.move(
                        "mv",
                        self.results_dir.join(output),
                        dest,
                    )
            self._upload_outputs(gcs_bucket)

            if upload_to_catapult:
                self._api.catapult.upload_test_outputs(self.results_dir)

    def _upload_to_resultdb(
        self, summary_path, orchestration_inputs, base_variant, tags
    ):
        cmd = [orchestration_inputs.resultdb]
        base_variant = base_variant.copy()
        base_variant.update(
            {
                "bucket": self._api.buildbucket.build.builder.bucket,
                "builder": self._api.buildbucket.build.builder.builder,
            }
        )
        if self.results_dir:
            cmd.append("--output=%s" % self.results_dir)
        cmd.append("--summary=%s" % summary_path)
        for tag in tags:
            cmd.append("--tag=%s" % tag)
        try:
            self._api.step(
                "resultdb",
                self._api.resultdb.wrap(cmd, base_variant=base_variant, include=True),
            )
        except self._api.step.InfraFailure:  # pragma: no cover  # pylint: disable=try-except-raise
            # When there is a reliability issues with ResultDB, replace the
            # following statement from 'raise' to 'pass' so the step stays an
            # infra failure but the build is not affected:
            raise

    def _upload_outputs(self, gcs_bucket):
        self._api.gsutil.upload_namespaced_directory(
            source=self.task_output_dir,
            bucket=gcs_bucket,
            # Namespace to avoid collision across shards and attempts.
            subpath="%s/%s" % (self._shard_name, self.swarming_task_id),
            # Internal gsutil retries of transient failures can cause rsync to
            # fail when run by a service account that doesn't have permission
            # to delete objects.
            rsync=False,
        )

    def run_triage(self, triage_tool, triage_sources):
        snapshot_zip = self.results_dir.join(self._api.testing_requests.SNAPSHOT_NAME)
        self._api.path.mock_add_paths(snapshot_zip)
        if not self._api.path.exists(snapshot_zip):  # pragma: no cover
            return
        with self._api.step.nest("run triage"):
            snapshot_dir = self._api.path.mkdtemp("snapshot")
            self._api.tar.extract(
                "extract snapshot", path=snapshot_zip, directory=snapshot_dir
            )
            cmd = (
                [triage_tool]
                + ["--config=%s" % f for f in triage_sources]
                + ["--data", snapshot_dir]
            )
            return self._api.step(
                "triage",
                cmd,
                # The triage tool returns a non-zero exit code when it detects
                # any violations or errors. We don't want to fail in this case,
                # but just write the output to a file for analysis.
                ok_ret="any",
                stdout=self._api.raw_io.output_text(
                    # Write to the results_dir so it gets uploaded to ResultDB.
                    leak_to=self.results_dir.join("triage_output")
                ),
                step_test_data=lambda: self._api.raw_io.test_api.stream_output_text(
                    "triage info"
                ),
            )


class Task(swarming_retry_api.TriggeredTask):
    """Task processes and presents results of testing Swarming tasks."""

    def __init__(
        self,
        request,
        api,
        orchestration_inputs,
        debug_symbol_url,
        retry_task_on_test_failure=False,
        **kwargs
    ):
        super().__init__(request, api, **kwargs)
        self._orchestration_inputs = orchestration_inputs
        self._debug_symbol_url = debug_symbol_url
        self._retry_task_on_test_failure = retry_task_on_test_failure
        # 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 self.name.startswith("multiplied:"):
            self.max_attempts = 1
            self._is_multiplied = True
        else:
            self._is_multiplied = False
        # Abort on test failures that are likely due to the change being tested
        # to decrease true rejection time.
        self.abort_early_if_failed = (
            self.name.startswith("affected:") or self._is_multiplied
        )

    def launch(self, priority_boost_amount):
        """Launches a swarming task attempt.

        It also initializes the test_results field on the returned
        swarming_retry.Attempt to None so that the field will be set even if
        the attempt is not completed or the result processing fails.
        """
        attempt = super().launch(priority_boost_amount)
        attempt.test_results = None
        return attempt

    def process_result(self, attempt):
        """Processes the results produced by a test shard."""
        assert attempt.result
        result = attempt.result

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

        @attr.s
        class _LogToProcess(object):
            name = attr.ib(type=str)
            data = attr.ib(type=str)
            path = attr.ib(type=Path)

        swarming_summary_path = self._api.path.mkstemp(result.id)
        self._api.file.write_raw(
            "write swarming summary JSON",
            swarming_summary_path,
            self._api.json.dumps(result.raw),
        )

        tefmocheck_cmd = [
            self._orchestration_inputs.tefmocheck,
            "-swarming-summary-json",
            swarming_summary_path,
            "-swarming-host",
            attempt.host,
        ]

        to_process = [
            _LogToProcess(
                name=self._api.testing_requests.TEST_TASK_OUTPUT_FILE,
                path=result.output_dir.join(
                    self._api.testing_requests.TEST_TASK_OUTPUT_FILE
                ),
                data=result.output,
            )
        ]
        for log_name in (
            self._api.testing_requests.SYSLOG_NAME,
            self._api.testing_requests.SERIAL_LOG_NAME,
        ):
            if log_name in result.outputs:
                log_path = result.outputs[log_name]
                # TODO(crbug.com/1222835): Remove this check when bug is fixed.
                self._api.path.mock_add_paths(log_path)
                if self._api.path.isfile(log_path):
                    to_process.append(
                        _LogToProcess(
                            name=log_name,
                            path=log_path,
                            data=self._api.file.read_text(
                                "read %s" % log_name,
                                log_path,
                                test_data="extra log contents",
                            ),
                        )
                    )

        for log in to_process:
            # Flag names from
            # https://fuchsia.googlesource.com/fuchsia/+/main/tools/testing/tefmocheck/cmd/main.go
            tefmocheck_flag_name = {
                self._api.testing_requests.TEST_TASK_OUTPUT_FILE: "-swarming-output",
                self._api.testing_requests.SERIAL_LOG_NAME: "-serial-log",
                self._api.testing_requests.SYSLOG_NAME: "-syslog",
            }.get(log.name)
            if tefmocheck_flag_name:
                tefmocheck_cmd.extend((tefmocheck_flag_name, log.path))
            # Symbolize and overwrite so that tefmocheck and logdog see the
            # symbolized versions.
            if self._api.testing.task_targets_fuchsia(result):
                # 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.
                attempt.logs[log.name] = self._api.symbolize(
                    symbolizer_tool=self._orchestration_inputs.symbolizer_tool,
                    data=log.data,
                    name="symbolize %s" % log.name,
                    debug_symbol_url=self._debug_symbol_url,
                    symbolizer_output=log.path,
                )
            else:
                attempt.logs[log.name] = log.data
                if not self._api.path.exists(log.path):
                    # Ensure it exists on file system even if we didn't symbolize
                    # so that it gets uploaded to GCS later.
                    self._api.file.write_raw("write %s" % log.name, log.path, log.data)
            self._api.path.mock_add_paths(log.path)

        test_results_dir = self._api.testing.extract_test_results(
            step_name="extract results", task_result=result
        )

        # Copy syslog.txt, serial.test and infra_and_test_std_and_klog.txt to test_results_dir.
        # We look for artifacts from test_results_dir to upload to ResultDB.
        self._api.file.copy(
            "copy %s" % self._api.testing_requests.TEST_TASK_OUTPUT_FILE,
            result.output_dir.join(self._api.testing_requests.TEST_TASK_OUTPUT_FILE),
            test_results_dir.join(self._api.testing_requests.TEST_TASK_OUTPUT_FILE),
        )
        for log_name in (
            self._api.testing_requests.SYSLOG_NAME,
            self._api.testing_requests.SERIAL_LOG_NAME,
        ):
            # TODO(crbug.com/1222835): Remove the isfile check when bug is fixed.
            if log_name in result.outputs and self._api.path.isfile(
                result.outputs[log_name]
            ):
                self._api.file.copy(
                    "copy %s" % log_name,
                    result.outputs[log_name],
                    test_results_dir.join(log_name),
                )

        tefmocheck_cmd.extend(("-outputs-dir", test_results_dir))
        summary_json_path = test_results_dir.join(TEST_SUMMARY_JSON)

        if self._api.path.exists(summary_json_path):
            tefmocheck_cmd.extend(("-test-summary-json", summary_json_path))

        tefmocheck_cmd.extend(
            ("-json-output", self._api.raw_io.output_text(suffix="json"))
        )
        summary_bytes = self._api.step("tefmocheck", tefmocheck_cmd).raw_io.output_text
        attempt.logs[TEST_SUMMARY_JSON] = summary_bytes

        tags = {}
        for tag in self._request.tags:
            k, v = tag.split(":", 1)
            tags[k] = v

        test_results = FuchsiaTestResults(
            from_fuchsia=self._api.testing.task_targets_fuchsia(result),
            results_dir=test_results_dir,
            swarming_task_id=result.id,
            swarming_bot_id=result.bot_id,
            shard_name=result.name,
            env_name=tags[self._api.testing_requests.TEST_ENVIRONMENT_TAG_NAME],
            api=self._api,
            task_output_dir=result.output_dir,
            summary_bytes=summary_bytes,
            is_multiplied=self._is_multiplied,
        )
        attempt.test_results = test_results
        flaked_tests = test_results.flaked_tests
        failed_tests = test_results.failed_tests
        if not result.success:
            if result.state == self._api.swarming.TaskState.COMPLETED:
                # The task will have state COMPLETED even if it had a non-zero
                # exit code, so show a more helpful message than "completed".
                attempt.failure_reason = "task failed"
            else:
                # "BOT_DIED" -> "bot died"
                attempt.failure_reason = result.state.name.lower().replace("_", " ")
        else:
            if flaked_tests:
                attempt.has_flakes = True
            if failed_tests:
                attempt.failure_reason = "%s failed" % pluralize("test", failed_tests)
                # Only retry if all the failures are testing_failure_mode checks.
                # Sometimes, a problem can occur in the system that causes all tests
                # to fail after that point, in which case we should also retry.
                # The number of failed tests to check for in this case should match
                # that of the mass_test_failure_check here:
                # https://fuchsia.googlesource.com/fuchsia/+/e2f3065a52c5bd69090d0c2dcd8cbc76a75d601c/tools/testing/tefmocheck/cmd/main.go#125
                should_retry = (
                    all(t.get("is_testing_failure_mode") for t in failed_tests)
                    or len(failed_tests) > 5
                )
                if not self._retry_task_on_test_failure and not should_retry:
                    self.max_attempts = 1
        # 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", {}).items():
            attempt.logs[output_name] = test_results.get_output(output_path)
        for tests_by_name in [failed_tests] + [flaked_tests]:
            for test in tests_by_name:
                for output_file in test["output_files"]:
                    # Proactively read the log so that we can use an empty step
                    # that doesn't have "execution details" or "stdout" logs to
                    # present the failed test more clearly.
                    test_results.get_output(output_file)

        triage_step = test_results.run_triage(
            self._orchestration_inputs.triage,
            self._orchestration_inputs.triage_sources,
        )
        if triage_step:
            attempt.logs["triage_output"] = triage_step.stdout.splitlines()

    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" or category == "incomplete":
            show_failures_in_red = False

        attempt_status = "fail"
        if attempt.success:
            if attempt.has_flakes:
                attempt_status = "flake"
            else:
                attempt_status = "pass"
        name = "%s (%s)" % (attempt.name, attempt_status)
        with self._api.step.nest(name) as presentation:
            if show_failures_in_red and (not attempt.success or attempt.has_flakes):
                presentation.status = self._api.step.FAILURE

            if attempt.result and attempt.result.duration_secs:
                presentation.step_text = nice_duration(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.bot_ui_link:
                presentation.presentation.links[
                    "bot %s" % attempt.bot_id
                ] = attempt.bot_ui_link

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

            for log, data in sorted(attempt.logs.items()):
                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", {}).items():
                    presentation.logs[name] = test_results.get_output(path).split("\n")

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


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

    FuchsiaTestResults = FuchsiaTestResults

    TEST_SUMMARY_JSON = TEST_SUMMARY_JSON

    def task_targets_fuchsia(self, task_result):
        # Sharded tasks will have the OS environment in the task name.
        # Deprecated tasks do not have the OS in the name, but they always target
        # fuchsia, so this function should always return true.
        name = task_result.name.lower()
        return "linux" not in name and "mac" not in name

    def extract_test_results(self, step_name, task_result):
        """Extracts test results from task_result.

        Moves test results from the task_result's outputs into a new
        directory. We do this because we later want to be able to archive
        everything in the swarming task result's outputs, but we don't want
        to archive most of the test result files.

        Args:
            step_name (str): The name of the step.
            task_result (api.swarming.TaskResult): The task result from which to extract
                the test results.

        Returns:
            The Path to the directory into which test results were extracted.
        """
        # Use a directory that doesn't yet exist so that `api.file.move()`
        # moves the source directory to the location of this directory, instead
        # of *into* this directory.
        test_results_dir = self.m.path.mkdtemp(task_result.id).join("test_results")

        # The outputs should be in TEST_RESULTS_DIR_NAME.
        test_results_path = None
        for relative_path in sorted(task_result.outputs.keys()):
            if relative_path.startswith(
                self.m.testing_requests.TEST_RESULTS_DIR_NAME + "/"
            ):
                test_results_path = str(
                    task_result.output_dir.join(
                        self.m.testing_requests.TEST_RESULTS_DIR_NAME
                    )
                )
                break

        # If the swarming task failed, there may not be any test results.
        if test_results_path is None and not task_result.success:
            self.m.file.ensure_directory("create test results dir", test_results_dir)
            return test_results_dir

        assert (
            test_results_path
        ), "test results not found amongst outputs of task %s: %s" % (
            task_result.name,
            task_result.outputs,
        )
        self.m.file.move(step_name, source=test_results_path, dest=test_results_dir)

        # 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 = test_results_dir.join(TEST_SUMMARY_JSON)
        outputs = self.m.file.listdir("get extracted files", test_results_dir)
        if summary_path in outputs:
            self.m.path.mock_add_paths(summary_path)
        return test_results_dir

    def run_test_tasks(
        self,
        debug_symbol_url,
        max_attempts,
        orchestration_inputs,
        rerun_budget_secs=None,
        runs_per_shard=1,
        retry_task_on_test_failure=False,
    ):
        """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:
            debug_symbol_url (str): A GCS URL 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.
            runs_per_shard (int): Number of times to run each shard.
            retry_task_on_test_failure (bool): Retry tasks on all test
                failures. If false, retry whole tasks on tefmocheck failures
                but not on regular test failures.

        Returns:
            A list of swarming_retry.Tasks 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.task_requests:
            self.m.step.empty("no tests to run")
            return []

        launch_deadline_time = None
        if rerun_budget_secs:
            # If we have a rerun deadline, we want to run the tests as many times as
            # possible within that time regardless of whether the tasks pass or fail,
            # so we don't care about setting a max number of attempts.
            max_attempts = float("inf")
            launch_deadline_time = self.m.time.time() + rerun_budget_secs

        tasks = []
        for task_request in orchestration_inputs.task_requests:
            if rerun_budget_secs:
                # Rerun tasks should not block other tasks.
                task_request = task_request.with_priority(task_request.priority + 1)
            tasks.append(
                Task(
                    request=task_request,
                    api=self.m,
                    orchestration_inputs=orchestration_inputs,
                    debug_symbol_url=debug_symbol_url,
                    retry_task_on_test_failure=retry_task_on_test_failure,
                )
            )

        tasks = self.m.swarming_retry.run_tasks(
            tasks=tasks,
            collect_output_dir=self.m.path.mkdtemp("swarming"),
            max_attempts=max_attempts,
            launch_deadline_time=launch_deadline_time,
            run_count=runs_per_shard,
        )

        self.m.swarming_retry.present_tasks(tasks=tasks)

        return tasks

    def raise_failures(self, tasks):
        failed_tasks = []
        failed_tests = []
        failed_tests_no_tefmo = []
        for task in tasks:
            if task.failed:
                failed_tasks.append(task.name)
                # Report the test failures from the last failed attempt.
                attempt = task.get_failed_attempts()[-1]
                if attempt.test_results:
                    failed_tests.extend(
                        t["name"] for t in attempt.test_results.failed_tests
                    )
                    failed_tests_no_tefmo.extend(
                        t["name"]
                        for t in attempt.test_results.failed_tests
                        if not t.get("is_testing_failure_mode")
                    )

        # Expose these so tools can be used to re-run these failed tests.
        with self.m.step.nest("record failed_test_names") as presentation:
            presentation.properties["failed_test_names"] = sorted(
                set(failed_tests_no_tefmo)
            )

        if not failed_tests:
            if failed_tasks:
                raise self.m.step.StepFailure(
                    "failed to process test results for %s: %s"
                    % (
                        pluralize("task", failed_tasks),
                        ", ".join(sorted(failed_tasks)),
                    )
                )
            return

        def _test_description(test, failure_count):
            count_info = ""
            if failure_count > 1:
                count_info = " (%d failures)" % failure_count
            return "%s%s" % (test, count_info)

        failure_counts = collections.Counter(failed_tests)
        header = "%s failed:" % pluralize("test", failure_counts)
        if len(failure_counts) == 1:
            test, failure_count = failure_counts.popitem()
            raise self.m.step.StepFailure(
                "%s %s" % (header, _test_description(test, failure_count))
            )

        error_lines = [
            header,
            # Milo's markdown renderer requires a blank line before the
            # start of a bulleted list. Otherwise the list lines will all
            # be rendered on one hyphen-separated line.
            "",
        ]

        # Present up to 10 tests and show a count of the rest. But if there are
        # 11-15 failed tests, still show all of the tests because it's not
        # worth saving lines by hiding only a couple tests.
        max_tests_to_present = 15
        min_tests_to_hide = 5
        if len(failure_counts) > max_tests_to_present:
            tests_to_present = max_tests_to_present - min_tests_to_hide
        else:
            tests_to_present = len(failure_counts)

        counts_to_present = sorted(failure_counts.items())[:tests_to_present]
        for test, failure_count in counts_to_present:
            error_lines.append("- %s" % _test_description(test, failure_count))

        hidden_tests = len(failure_counts) - tests_to_present
        if hidden_tests > 0:
            error_lines.append("- (%d more)" % hidden_tests)

        raise self.m.step.StepFailure("\n".join(error_lines))