recipe_modules/buildsetlookup/api.py

# Copyright 2018 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 BuildSetLookupApi(recipe_api.RecipeApi):
    """APIs for retrieving the build ID for a given buildset and builder ID."""

    def __call__(
        self,
        step_name,
        builder,
        buildset,
        build_status=None,
        leak_to=None,
        retries=0,
        interval=0,
    ):
        """Retrieves the build ID for a given buildset and builder ID.

        Args:
          step_name (str): The name of the step produced.
          builder (str): A fully-qualified buildbucket v2 builder ID,
            consisting of <project>/<project-namespaced bucket>/<builder name>. For example:
            'fuchsia/ci/core.x64-release'.
          buildset (str): A fully-qualified buildbucket V2 buildset tag,
            consisting of commit/gitiles/<host>/<repo>/+/<commit ID>.  For example:
            'commit/gitiles/fuchsia.googlesource.com/foo/+/e3127e0bd6d57da7a5959ee70eb0a396590e6d53'.
          build_status (int): The expected buildbucket status code of the build. If
            None, use tool's default (12, SUCCESS).
          leak_to (Path): If leak_to is provided, it must be a Path object. This path will be used in
            place of a random temporary file, and the file will not be deleted at the end of the step.
          retries (int): Number of times to retry lookup, in case builds are not yet
            triggered.
          interval (int): Interval between retries in seconds.
        """
        step_args = [
            self._buildsetlookup_client,
            "-builder-id",
            builder,
            "-build-set",
            buildset,
        ]
        if build_status is not None:
            step_args.extend(["-build-status", build_status])

        with self.m.step.nest(step_name):
            for attempt in range(retries + 1):
                step = self.m.step(
                    "attempt %d" % attempt,
                    step_args,
                    stdout=self.m.raw_io.output_text(leak_to=leak_to),
                    ok_ret=(0, 1),
                )
                if step.retcode:
                    if attempt < retries:
                        self.m.time.sleep(interval)
                        continue
                    else:
                        step.presentation.status = self.m.step.EXCEPTION
                        raise self.m.step.InfraFailure(
                            "could not find build for %s, buildset %s"
                            % (builder, buildset),
                        )
                return step.stdout

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