recipe_modules/builder_status/api.py

# Copyright 2023 The Pigweed Authors
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may not
# use this file except in compliance with the License. You may obtain a copy of
# the License at
#
#     https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations under
# the License.
"""Check the status of builders."""

import dataclasses
from typing import Optional

from PB.go.chromium.org.luci.buildbucket.proto import common as common_pb2
from recipe_engine import recipe_api


@dataclasses.dataclass
class BuilderStatus:
    project: str
    bucket: str
    builder: str

    builds: tuple


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

    def retrieve(
        self,
        project: Optional[str] = None,
        bucket: Optional[str] = None,
        builder: Optional[str] = None,
        n: int = 10,
    ) -> BuilderStatus:
        return BuilderStatus(
            project=project,
            bucket=bucket,
            builder=builder,
            builds=tuple(
                self.m.buildbucket_util.last_n_builds(
                    project=project, bucket=bucket, builder=builder, n=n,
                )
            ),
        )

    def is_incomplete(self, status: BuilderStatus) -> bool:
        """Is a build running or scheduled?"""
        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?"""
        for build in status.builds:
            if build.status == common_pb2.SUCCESS:
                return True
        return False