recipe_modules/buildbucket_util/api.py

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

from recipe_engine import recipe_api

from PB.go.chromium.org.luci.buildbucket.proto import common as common_pb2
from PB.go.chromium.org.luci.buildbucket.proto import rpc as rpc_pb2


class BuildbucketUtilApi(recipe_api.RecipeApi):
  """Module for high-level buildbucket utilities specific to Fuchsia infra."""

  @property
  def id(self):
    """A unique string identifier for the current build."""
    if self.m.led.launched_by_led:
      return self.m.led.run_id
    else:
      return str(self.m.buildbucket.build.id)

  @property
  def is_tryjob(self):
    """Whether the current build is running as a tryjob."""
    bucket_name = self.m.buildbucket.build.builder.bucket
    return bucket_name.endswith('try')

  def display_builds(self, step_name, builds, raise_on_failure=False):
    """
    Display build links and status for each input build.
    Optionally raise on build failure(s).

    Args:
      builds (seq[buildbucket.v2.Build]): buildbucket Build objects. See
        recipe_engine/buildbucket recipe module for more info.
      build_group_name (str): Name of build group to display in step name.
      raise_on_failure (bool): Raise InfraFailure or StepFailure on failure.

    Raises:
      InfraFailure: One or more input builds had infra failure. Takes priority
        over step failures.
      StepFailure: One or more of input builds failed.
    """
    infra_failed_builders = []
    failed_builders = []
    # Create per-build display steps.
    with self.m.step.nest(step_name):
      for build in builds:
        with self.m.step.nest(build.builder.builder) as display_step:
          step_links = display_step.presentation.links
          step_links[str(build.id)] = self.m.buildbucket.build_url(
              build_id=build.id)  # yapf: disable
          if build.status == common_pb2.SUCCESS:
            display_step.presentation.status = self.m.step.SUCCESS
          elif build.status == common_pb2.INFRA_FAILURE:
            display_step.presentation.status = self.m.step.EXCEPTION
            infra_failed_builders.append(build.builder.builder)
          elif build.status == common_pb2.FAILURE:
            display_step.presentation.status = self.m.step.FAILURE
            failed_builders.append(build.builder.builder)
          # For any other status, use warning color.
          else:
            display_step.presentation.status = self.m.step.WARNING

      if raise_on_failure:
        # Construct failure header and message. Include both types of failures,
        # regardless of whether we raise purple or red.
        failure_header = 'build(s) failed'
        failure_message = []
        if infra_failed_builders:
          failure_message.append(
              'infra failures: {infra_failed_builders}'.format(
                  infra_failed_builders=', '.join(infra_failed_builders)))
        if failed_builders:
          failure_message.append('step failures: {failed_builders}'.format(
              failed_builders=', '.join(failed_builders)))
        failure_message = ', '.join(failure_message)
        # If there were any infra failures, raise purple.
        if infra_failed_builders:
          self.m.python.infra_failing_step(
              failure_header,
              failure_message,
          )
        # Otherwise if there were any step failures, raise red.
        if failed_builders:
          self.m.python.failing_step(
              failure_header,
              failure_message,
          )

  def last_build(self, project, bucket, builder, fields=None, status=None):
    """Returns the build proto for a builder's most recent successful build.

    Returns None if no successful builds could be found.

    Args:
      project (str): LUCI project.
      bucket (str): BuildBucket bucket.
      builder (str): BuildBucket builder.
      fields (seq of str): BuildBucket Build proto message fields to include in
          the result. If None, defaults to api.buildbucket.DEFAULT_FIELDS.
      status (common_pb2.Status enum member): If set, will return the
          last build with this status.
    """
    predicate = rpc_pb2.BuildPredicate()
    predicate.builder.project = project
    predicate.builder.bucket = bucket
    predicate.builder.builder = builder
    if status is not None:
      predicate.status = status

    fields = fields or self.m.buildbucket.DEFAULT_FIELDS
    builds = self.m.buildbucket.search(predicate, limit=1, fields=fields)
    if not builds:
      return None

    assert len(builds) == 1
    return builds[0]