recipe_modules/lkg/api.py

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

from recipe_engine import recipe_api


class LKGApi(recipe_api.RecipeApi):
    """APIs for retrieving last-known-good entities."""

    def revision(
        self, step_name, builders, gitiles_ref=None, search_range=100, test_data=None
    ):
        """Get the last-known-good revision for a set of builders.

        Args:
          step_name (str): Name of the step.
          builders (seq(str)): A sequence of fully-qualified buildbucket
            builder names.
          gitiles_ref (str): If specified, filter by this gitiles ref, e.g.
            refs/heads/main.
          search_range (int): Number of builds per builder to search.

        Returns:
          str: Last-known-good revision.
        """
        args = ["revision", "-search-range", search_range]
        for builder in builders:
            args += ["-builder", builder]
        return self._run(
            step_name,
            args,
            gitiles_ref=gitiles_ref,
            step_test_data=lambda: self.m.raw_io.test_api.stream_output_text(test_data),
        ).stdout.strip()

    def build(self, step_name, builder, gitiles_ref=None, test_data=None):
        """Get the last-known-good buildbucket ID for a builder.

        Args:
          step_name (str): Name of the step.
          builder (str): A fully-qualified buildbucket builder name.
          gitiles_ref (str): If specified, filter by this gitiles ref, e.g.
            refs/heads/main.

        Returns:
          str: Last-known-good buildbucket build ID.
        """
        args = ["build", "-builder", builder]
        return self._run(
            step_name,
            args,
            gitiles_ref=gitiles_ref,
            step_test_data=lambda: self.m.raw_io.test_api.stream_output_text(test_data),
        ).stdout.strip()

    def _run(self, step_name, args, gitiles_ref, step_test_data=None):
        cmd = [self._lkg_tool]
        cmd += args
        if gitiles_ref:
            cmd += ["-gitiles-ref", gitiles_ref]
        return self.m.step(
            step_name,
            cmd,
            stdout=self.m.raw_io.output_text(),
            step_test_data=step_test_data,
        )

    @property
    def _lkg_tool(self):
        return self.m.ensure_tool("lkg", self.resource("tool_manifest.json"))