recipe_modules/bazel_roll/api.py

# Copyright 2024 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.

from __future__ import annotations

import configparser
import dataclasses
import io
import re
from typing import TYPE_CHECKING

from PB.recipe_modules.pigweed.checkout.options import (
    Options as CheckoutOptions,
)
from recipe_engine import recipe_api

if TYPE_CHECKING:  # pragma: no cover
    from typing import Generator
    from PB.recipe_modules.pigweed.bazel_roll.git_repository import (
        GitRepository,
    )
    from recipe_engine import config_types
    from RECIPE_MODULES.pigweed.checkout import api as checkout_api
    from RECIPE_MODULES.pigweed.roll_util import api as roll_util_api


@dataclasses.dataclass
class RevisionChange:
    old: str
    new: str


class BazelRollApi(recipe_api.RecipeApi):

    RevisionChange = RevisionChange

    def workspace_path(
        self,
        root: config_types.Path,
        value: str,
    ) -> config_types.Path:
        """Figure out the location of the WORKSPACE or MODULE.bazel file.

        If value is '', look for root / 'WORKSPACE' and root / 'MODULE.bazel'.
        If exactly one exists, return it. If not, error out.

        If root / value is a file, return it.

        If root / value is a directory, set root = root / value and apply the
        above logic. This enables applying this logic to subdirectories.

        Args:
            api: Recipe API object.
            root: Checkout root.
            value: Relative path specified in properties.

        Returns:
            Path to the WORKSPACE or MODULE.bazel file.
        """
        if value:
            value_path = root / value
            self.m.path.mock_add_file(value_path)
            if self.m.path.isfile(value_path):
                return value_path
            elif self.m.path.isdir(value_path):  # pragma: no cover
                root = value_path
            else:
                self.m.step.empty(  # pragma: no cover
                    f'{value_path} does not exist',
                    status='FAILURE',
                )

        workspace = root / 'WORKSPACE'
        module_bazel = root / 'MODULE.bazel'

        self.m.path.mock_add_file(workspace)

        if self.m.path.isfile(module_bazel) and self.m.path.isfile(workspace):
            self.m.step.empty(  # pragma: no cover
                f'{module_bazel} and {workspace} both exist',
                status='FAILURE',
            )

        if self.m.path.isfile(module_bazel):
            return module_bazel  # pragma: no cover

        if self.m.path.isfile(workspace):
            return workspace

        self.m.step.empty(  # pragma: no cover
            'no WORKSPACE or MODULE.bazel file found',
            status='FAILURE',
        )

    def update_git_repository(
        self,
        checkout: checkout_self.m.CheckoutContext,
        git_repository: GitRepository,
    ) -> dict[str, self.m.roll_util.Roll]:
        workspace_path = self.workspace_path(
            checkout.root,
            git_repository.workspace_path,
        )
        git_repository.branch = git_repository.branch or 'main'

        new_revision: Optional[str] = None

        # First, try to get new_revision from the trigger.
        bb_remote: Optional[str] = None
        commit: common_pb2.GitilesCommit = (
            self.m.buildbucket.build.input.gitiles_commit
        )
        if commit and commit.project:
            new_revision = commit.id
            host: str = commit.host
            bb_remote: str = f'https://{host}/{commit.project}'

        # If we still don't have a revision then it wasn't in the trigger.
        # (Perhaps this was manually triggered.) In this case we update to the
        # property-specified branch HEAD.
        if new_revision is None:
            new_revision = git_repository.branch

        # If this was triggered by a gitiles poller, check that the triggering
        # repository matches project_remote. Exception: allow a trigger to be
        # for the top-level project instead.
        use_trigger_for_project = True

        if bb_remote:
            if checkout.remotes_equivalent(checkout.options.remote, bb_remote):
                use_trigger_for_project = False

            elif not checkout.remotes_equivalent(
                git_repository.remote,
                bb_remote,
            ):
                self.m.step.empty(
                    'triggering repository ({}) does not match project remote '
                    '({})'.format(bb_remote, git_repository.remote),
                    status='FAILURE',
                )

        project_dir = self.m.path.start_dir / 'project'

        project_checkout = self.m.checkout(
            CheckoutOptions(
                remote=git_repository.remote,
                branch=git_repository.branch,
                use_trigger=use_trigger_for_project,
            ),
            root=project_dir,
        )

        # In case new_revision is a branch name we need to retrieve the hash it
        # resolves to.
        if not re.search(r'^[0-9a-f]{40}$', new_revision):
            new_revision = self.m.checkout.get_revision(
                project_dir, 'get new revision', test_data='2' * 40
            )

        update_result = self.m.bazel.update_commit_hash(
            checkout=checkout,
            project_remote=git_repository.remote,
            new_revision=new_revision,
            path=workspace_path,
        )
        if not update_result:
            self.m.step.empty('failed to update commit hash', status='FAILURE')

        direction = self.m.roll_util.get_roll_direction(
            project_dir, update_result.old_revision, new_revision
        )

        # If the primary roll is not necessary or is backwards we can exit
        # immediately.
        if not self.m.roll_util.can_roll(direction):
            self.m.roll_util.skip_roll_step(
                git_repository.remote, update_result.old_revision, new_revision
            )
            return

        project_name = git_repository.name or update_result.project_name
        if not project_name:
            self.m.step.empty(
                f'could not find name line in {workspace_path}',
                status='FAILURE',
            )

        rolls: dict[str, self.m.roll_util.Roll] = {
            workspace_path: self.m.roll_util.create_roll(
                project_name=project_name,
                old_revision=update_result.old_revision,
                new_revision=new_revision,
                proj_dir=project_dir,
                direction=direction,
            ),
        }

        return rolls