recipe_modules/reported_step/api.py

# Copyright 2022 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 json
from recipe_engine import recipe_api
from PB.go.chromium.org.luci.resultdb.proto.v1.test_result import TestStatus

# step_name must include all parents steps separated by | (pipe), this is the
# default for `step.name` e.g. parent|nested|step
LOGDOG_LINK_FORMAT = (
    "https://cr-buildbucket.appspot.com/log/{build_id}/{step_name}?log={log_name}"
)


def _build_summary_html(logs):
    """
    input: A dict with the following shape
    {
        "log_name": "log_link.com",
        "log_name2": "log_link.com2",
        ...
    }
    """
    links_html = [
        f'<li><a href="{value}" target="_blank">{key}</a></li>'
        for key, value in logs.items()
    ]
    return "<ul>" + "\n".join(links_html) + "</ul>"


class PreparedStep:
    """Represents a step that will be uploaded to resultdb."""

    def __init__(self, api, test_id, step, resultdb_resource):
        self._api = api
        self._test_id = test_id
        self._step = step
        self._test_status = None
        self._resultdb_resource = resultdb_resource
        logs = list(x for x in step.presentation.logs.keys())

        # stdout is not always part of the `presentation.logs`.
        # This line is making sure to always include it.
        logs.append("stdout")
        self._logs = logs
        self._artifacts = {}

    def set_test_status(self, new_status):
        """
        Sets the test result status to the provided argument.
        If test status is set it will take priority over the step status.

        Args:
            new_status (TestStatus): Status for the test result.
        """
        self._test_status = new_status

    def add_log(self, log_name):
        """
        Adds a log that will link to logdog.

        The user of this method must
        ensure that the log being linked exists.

        Args:
            log_name (str)
        """
        self._logs.append(log_name)

    def add_artifact(self, artifact_name, artifact_content):
        """
        Adds an artifact and its contents to be uploaded as part of the reported
        step.

        Args:
            artifact_name (str): name of the artifact as it will be displayed in
            the test results ui

            artifact_content (str | list(str)): The content of the artifact
        """
        self._artifacts[artifact_name] = artifact_content

    def upload(self):
        """
        Uploads the prepared_step to resultdb.

        This operation creates a step at the current nesting level.
        Logs set in the prepared step are linked as part of the summary.
        """
        step_name = f"{self._step.name_tokens[-1]} - upload to resultdb"

        if not self._api.resultdb.enabled:
            self._api.step.empty(
                step_name,
                status=self._api.step.INFRA_FAILURE,
                step_text="ResultDB integration was not enabled for this build",
                raise_on_failure=False,
            )
            return

        status = (
            self._test_status
            if self._test_status is not None
            else self._status_to_test_result(self._step.presentation.status)
        )
        expected = status == TestStatus.PASS

        test_result = {
            "testId": self._test_id,
            "expected": expected,
            "summaryHtml": _build_summary_html(self._logs_to_links()),
            "status": TestStatus.Name(status),
        }

        cmd = [
            "vpython",
            self._resultdb_resource,
            json.dumps(test_result),
        ]

        if self._artifacts:
            for name, contents in self._artifacts.items():
                cmd.extend(["--artifact", name, self._artifact_to_raw_input(contents)])

        if not self._api.runtime.in_global_shutdown:
            self._api.step(
                step_name,
                self._api.resultdb.wrap(cmd),
            )

    def _logs_to_links(self):
        """Get the links to the logs uploaded to logdog.
        returns a new dict with the following shape:
        {
            "log_name_1": "https://link_to_log_name_1.com",
            "log_name_2": "https://link_to_log_name_2.com",
            ...
        }
        """

        return {
            log_name: LOGDOG_LINK_FORMAT.format(
                build_id=self._api.buildbucket_util.id,
                step_name=self._step.name,
                log_name=log_name,
            )
            for log_name in self._logs
        }

    def _status_to_test_result(self, step_status):
        status_map = {
            self._api.step.EXCEPTION: TestStatus.CRASH,
            self._api.step.INFRA_FAILURE: TestStatus.CRASH,
            self._api.step.FAILURE: TestStatus.FAIL,
            self._api.step.SUCCESS: TestStatus.PASS,
            # Note: copying from chromium/recipes-py/recipe_modules/step/api.py
            # Note: although recipes currently have this WARNING status, it's not
            # effecively hooked up to anything in the UI and is treated as SUCCESS.
            # crbug.com/854099
            self._api.step.WARNING: TestStatus.PASS,
        }
        return status_map.get(step_status, TestStatus.FAIL)

    def _artifact_to_raw_input(self, artifact_content):
        content = artifact_content
        if isinstance(artifact_content, list):
            content = "\n".join(artifact_content)

        return self._api.raw_io.input_text(content)


class ReportedStepApi(recipe_api.RecipeApi):
    """ReportedStepApi is a wrapper for steps that can be treated as tests,
    i.e. they verify the code on some criteria, emit logs, can fail the build,
    etc. And whose results should be uploaded to resultdb"""

    def prepare_step(self, test_id, step):
        """
        Use this function when you want to upload a step but need to attach
        logs or artifacts before it gets uploaded. e.g. Attach the failure
        reason of the build step.
        Args:
            test_id (str): test id to be used by resultdb.
            step (StepData): The step as returned by calling api.step.

        returns:
            PreparedStep
        """
        return PreparedStep(self.m, test_id, step, self.resource("resultdb.py"))

    def upload(self, test_id, step):
        """
        Uploads the step to resultdb.
        Logs set in the step are linked as part of the summary.

        Args:
            test_id (str): test id to be used by resultdb.
            step (StepData): The step as returned by calling api.step.
        """

        prepared_step = self.prepare_step(test_id, step)
        prepared_step.upload()