| # 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 = [ |
| "vpython3", |
| 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), |
| infra_step=True, |
| ) |
| |
| 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() |