recipe_modules/util/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.
"""Utility functions common to multiple recipes that don't fit elsewhere."""

from __future__ import annotations

import dataclasses
import json
import re
from collections.abc import Sequence
from typing import Any

from google.protobuf import json_format

from recipe_engine import recipe_api

from PB.go.chromium.org.luci.buildbucket.proto import build as build_pb
from PB.go.chromium.org.luci.buildbucket.proto import common as common_pb


@dataclasses.dataclass
class ChangeWithComments:
    change: common_pb.GerritChange
    details: dict[str, Any]
    commit_message: str
    comments: Sequence[str]


class UtilApi(recipe_api.RecipeApi):
    ChangeWithComments = ChangeWithComments

    def get_change_with_comments(self) -> ChangeWithComments:
        input_: build_pb.Build.Input = self.m.buildbucket.build.input
        change: common_pb.GerritChange = input_.gerrit_changes[0]
        change_id: str = str(change.change)
        details: dict[str, Any] = self.m.gerrit.change_details(
            'change details',
            change_id=change_id,
            host=input_.gerrit_changes[0].host,
            query_params=['ALL_COMMITS', 'ALL_REVISIONS', 'ALL_FILES'],
            test_data=self.m.json.test_api.output(
                {
                    'owner': {
                        'email': 'coder@example.com',
                    },
                    'current_revision': 'a' * 40,
                    'revisions': {
                        'a' * 40: {
                            'files': [],
                            'commit': {
                                'message': '',
                            },
                            'description': 'description',
                        }
                    },
                    'revert_of': 0,
                }
            ),
        ).json.output

        current_revision: str = details['revisions'][
            details['current_revision']
        ]
        commit_message: str = current_revision['commit']['message']

        comments: list[str] = []

        for revision in details['revisions'].values():
            if revision.get('description'):
                comments.append(revision['description'])

        comments_result: dict[str, Any] = self.m.gerrit.list_change_comments(
            'list change comments',
            change_id,
            test_data=self.m.json.test_api.output(
                {
                    '/PATCHSET_LEVEL': [{'message': ''}],
                }
            ),
        ).json.output

        for _, comment_data in comments_result.items():
            comments.extend(x['message'] for x in comment_data)

        return ChangeWithComments(change, details, commit_message, comments)

    def find_matching_comment(
        self, rx: re.Pattern, comments: Sequence[str]
    ) -> str | None:
        """Find a comment in comments that matches regex object rx."""
        result: str | None = None
        with self.m.step.nest('checking comments'):
            for i, comment in enumerate(comments):
                with self.m.step.nest(f'comment ({i})') as pres:
                    pres.step_summary_text = comment
                    match: re.Match = re.search(rx, comment)
                    if match:
                        pres.step_summary_text = f'MATCH: {comment}'
                        result: str = match.group(0)
                        break

            if result:
                with self.m.step.nest('found'):
                    pass

        return result

    def build_metadata(self) -> dict[str, Any]:
        return {
            'bb_id': self.m.buildbucket.build.id,
            'swarming_id': self.m.swarming.task_id,
            'builder': self.m.buildbucket_util.full_builder_name(),
            'url': self.m.buildbucket.build_url(),
            'triggers': [
                json.loads(json_format.MessageToJson(x))
                for x in self.m.scheduler.triggers
            ],
        }

    def on_last_iteration(
        self,
        poll_interval_secs: int,
        *,
        debug: bool = False,
    ) -> bool:
        remaining = self._secs_remaining_in_build()
        if debug:
            pres = self.m.step.empty('secs remaining in build').presentation
            pres.step_summary_text = f'{remaining}'

            pres = self.m.step.empty('poll interval secs').presentation
            pres.step_summary_text = f'{poll_interval_secs}'

        return remaining < poll_interval_secs * 3 // 2

    def _secs_remaining_in_build(self) -> float:
        current_time = self.m.time.time()
        elapsed_time = (
            current_time - self.m.buildbucket.build.start_time.seconds
        )
        remaining = (
            self.m.buildbucket.build.execution_timeout.seconds - elapsed_time
        )

        # If a test isn't fully specified we might get some inconsistent results.
        # Ignore those cases.
        if remaining < -1337000000:
            remaining = 10000

        return remaining