recipe_modules/builder_status/api.py

# Copyright 2023 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.
"""Check the status of builders."""

import dataclasses
import datetime
import functools
from typing import Optional, Sequence, Union

from PB.go.chromium.org.luci.buildbucket.proto import (
    build as build_pb2,
    builder_common as builder_common_pb2,
    common as common_pb2,
)
from recipe_engine import recipe_api
import RECIPE_MODULES.fuchsia.buildbucket_util.api as buildbucket_util


@dataclasses.dataclass
class BuilderStatus:
    _api: recipe_api.RecipeApi

    project: str
    bucket: str
    builder: str

    exists: bool
    builds: tuple

    incomplete_included: bool

    @property
    def link(self) -> str:
        self._api.buildbucket.builder_url(
            project=self.project,
            bucket=self.bucket,
            builder=self.builder,
        )


class BuilderStatusApi(recipe_api.RecipeApi):
    """Check the status of builders."""

    BuilderStatus = BuilderStatus

    @functools.cached_property
    def buildbucket_config(self):
        return self.m.luci_config.buildbucket()

    def _does_builder_exist(
        self,
        project: str,
        bucket: str,
        builder: str,
    ) -> bool:
        if project != self.m.buildbucket.build.builder.project:
            raise Exception("only the current project is supported")  # pragma: no cover

        # Most test cases won't hinge on whether a builder exists, so if no buckets
        # or builders are present in the buildbucket config just pretend any
        # requested builders exist.
        if self._test_data.enabled and not self.buildbucket_config.buckets:
            return True

        return any(
            bkt.name == bucket and b.name == builder
            for bkt in self.buildbucket_config.buckets
            for b in bkt.swarming.builders
        )

    # BuilderStatus is defined here but pylint thinks it isn't.
    def retrieve(
        self,
        project: Optional[str] = None,
        bucket: Optional[str] = None,
        builder: Optional[str] = None,
        n: int = 10,
        include_incomplete: bool = True,
        max_age: Optional[datetime.timedelta] = None,
        timeout: Union[datetime.timedelta, int, float] = (
            buildbucket_util.DEFAULT_SEARCH_TIMEOUT
        ),
    ) -> BuilderStatus:  # pylint: disable=undefined-variable
        this_builder: builder_common_pb2.BuilderID = self.m.buildbucket.build.builder
        project: str = project or this_builder.project
        bucket: str = bucket or this_builder.bucket
        builder: str = builder or this_builder.builder

        exists: bool = self._does_builder_exist(project, bucket, builder)

        builds: Sequence[build_pb2.Build] = ()
        if exists:
            status: Optional[common_pb2.Status] = common_pb2.ENDED_MASK
            if include_incomplete:
                status = None

            builds: Sequence[build_pb2.Build] = tuple(
                self.m.buildbucket_util.last_n_builds(
                    project=project,
                    bucket=bucket,
                    builder=builder,
                    status=status,
                    n=n,
                    fields=("status",),
                    max_age=max_age,
                    timeout=timeout,
                )
            )

        return BuilderStatus(
            _api=self.m,
            project=project,
            bucket=bucket,
            builder=builder,
            exists=exists,
            builds=builds,
            incomplete_included=include_incomplete,
        )

    def is_incomplete(self, status: BuilderStatus) -> bool:
        """Is a build running or scheduled?"""
        # This method doesn't make sense if incomplete builds weren't included
        # in the search.
        assert status.incomplete_included

        for build in status.builds:
            if build.status in (common_pb2.SCHEDULED, common_pb2.STARTED):
                return True
        return False

    def is_passing(self, status: BuilderStatus) -> bool:
        """Did the most recently completed build pass?"""
        for build in status.builds:
            if build.status in (common_pb2.SCHEDULED, common_pb2.STARTED):
                continue
            return build.status == common_pb2.SUCCESS
        return False

    def has_recently_passed(self, status: BuilderStatus) -> bool:
        """Has any recent build passed?"""
        # Most test cases want most builders to appear passing. Make that the default
        # in cases where test data was not supplied.
        if self._test_data.enabled and not status.builds:
            return True

        for build in status.builds:
            if build.status == common_pb2.SUCCESS:
                return True
        return False