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 json
import re
from typing import TYPE_CHECKING

import attrs
from google.protobuf import json_format
from recipe_engine import recipe_api

if TYPE_CHECKING:  # pragma: no cover
    from typing import Any, Sequence
    from PB.go.chromium.org.luci.buildbucket.proto import (
        build as build_pb2,
        common as common_pb2,
    )


@attrs.define
class ChangeWithComments:
    change: common_pb2.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_pb2.Build.Input = self.m.buildbucket.build.input
        change: common_pb2.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
            ],
        }