recipe_modules/ci_status/api.py

# Copyright 2021 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.
"""This module provides utilities for checking the status of CI builders.

The `CiStatusApi` class in this module defines methods for interacting with the
CI system to determine the health and status of builders. This can be used to
implement conditional logic in recipes, such as skipping certain steps if a
builder is already failing.
"""

from __future__ import annotations

from recipe_engine import recipe_api

from PB.go.chromium.org.luci.buildbucket.proto import common
from PB.recipe_engine import result as result_pb


class CiStatusApi(recipe_api.RecipeApi):
    """Calls to build code."""

    def transform_bucket_name(self, bucket: str | None = None) -> str:
        if bucket is None:
            bucket = self.m.buildbucket.build.builder.bucket
        parts = bucket.split('.')
        parts = ['ci' if x == 'try' else x for x in parts]
        parts = [x for x in parts if x != 'shadow']
        return '.'.join(parts)

    def exit_early_in_recipe_testing_if_failing(
        self,
    ) -> result_pb.RawResult | None:
        if not self.m.recipe_testing.enabled:
            return None

        with self.m.step.nest('checking CI status'):
            bucket = self.transform_bucket_name()

            all_props = self.m.properties.thaw()
            assume_existence = False
            if 'ci_builder_exists' in all_props:
                if all_props['ci_builder_exists']:
                    assume_existence = True
                else:
                    self.m.step('builder does not exist in CI', None)
                    return None

            status = self.m.builder_status.retrieve(
                bucket=bucket,
                include_incomplete=False,
                assume_existence=assume_existence,
                n=5,
            )
            if self.m.builder_status.has_recently_passed(status):
                self.m.step('builder recently passed in CI', None)
                return None

            full_name = '/'.join((status.project, bucket, status.builder))
            return result_pb.RawResult(
                summary_markdown=(
                    f'Exiting early since [{full_name}]({status.link}) has not '
                    'recently passed.'
                ),
                status=common.SUCCESS,
            )