recipe_modules/raw_result/api.py

# Copyright 2025 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.
"""Insert additional annotations to RawResult messages."""

from __future__ import annotations

import functools
from collections.abc import Callable

from recipe_engine import recipe_api

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


def _get_reasons(*exceptions: Exception) -> list[str]:
    reasons = []

    for exception in exceptions:
        if isinstance(exception, ExceptionGroup):
            for exc in exception.exceptions:
                reasons.extend(_get_reasons(exc))

        elif isinstance(exception, recipe_api.StepFailure):
            reasons.append(exception.reason)

        else:
            reasons.append(str(exception))

    return reasons


def wrap_run_steps(f: Callable[..., result_pb.RawResult | None]):
    @functools.wraps(f)
    def _wrapper(
        api: recipe_api.RecipeScriptApi,
        *args,
        **kwargs,
    ) -> result_pb.RawResult:
        # If we're not doing bisection, then there's nothing to add to the
        # result, so directly pass through the return value.
        if not api.raw_result.bisector.enabled:
            return f(api, *args, **kwargs)

        infra_failure: list[Exception] = []
        step_failure: list[Exception] = []
        other_exception: list[Exception] = []

        try:
            result = f(api, *args, **kwargs)

            if result is None:
                return result_pb.RawResult(
                    summary_markdown=api.raw_result.bisector.summary(''),
                    status=common_pb.SUCCESS,
                )

            return result_pb.RawResult(
                summary_markdown=api.raw_result.bisector.summary(
                    result.summary_markdown,
                ),
                status=result.status,
            )

        except* recipe_api.InfraFailure as exc:
            infra_failure.append(exc)

        except* recipe_api.StepFailure as exc:
            step_failure.append(exc)

        except* Exception as exc:
            other_exception.append(exc)

        all_exceptions = [*infra_failure, *step_failure, *other_exception]

        summary_markdown = api.raw_result.bisector.summary(
            '\n\n'.join(_get_reasons(*all_exceptions)),
        )

        status = common_pb.FAILURE
        if other_exception or infra_failure:
            status = common_pb.INFRA_FAILURE
        if 'LUCI CV no longer needs this Tryjob' in summary_markdown:
            status = common_pb.CANCELED

        return result_pb.RawResult(
            summary_markdown=summary_markdown,
            status=status,
        )

    return _wrapper


class RawResultApi(recipe_api.RecipeApi):
    @property
    def bisector(self):
        # Don't make all recipes using this module also explicitly depend on
        # the bisector.
        return self.m.bisector