recipe_modules/merge_roll/api.py

# Copyright 2026 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.
"""Recipe module for merging into a Git repository."""

from __future__ import annotations

import dataclasses
from collections.abc import Callable

from recipe_engine import recipe_api

from PB.recipe_modules.pigweed.merge_roll.merge import MergeEntry

from RECIPE_MODULES.fuchsia.git_roll_util import api as git_roll_util_api
from RECIPE_MODULES.pigweed.checkout import api as checkout_api


@dataclasses.dataclass
class RevisionChange:
    """Represents a change in a submodule's pinned revision.

    Attributes:
        old: The old commit hash the submodule was pinned to.
        new: The new commit hash the submodule will be pinned to.
        finalize: A callable that executes the Git merge command.
    """

    new: str
    old: str
    finalize: Callable[[], None]


class MergeRollApi(recipe_api.RecipeApi):
    """API for merging branches."""

    RevisionChange = RevisionChange

    def prepare_merge(
        self,
        checkout: checkout_api.CheckoutContext,
        new_revision: str,
    ) -> RevisionChange | None:
        """Prepares and returns a RevisionChange object for a merge.

        A `finalize` function is created that, when called, will merge the
        `new_revision`.

        Args:
            checkout: The checkout context of the main repository.
            new_revision: The new commit hash to merge.

        Returns:
            A `RevisionChange` object containing the old and new revisions,
            and a `finalize` callable to apply the update, or None if there's
            nothing to merge.
        """

        with self.m.context(cwd=checkout.top):
            self.m.git.fetch(checkout.options.remote, new_revision)

            new_commits: list[str] = []
            for pair in self.m.git.cherry(
                source=new_revision,
                destination='HEAD',
            ):
                if not pair.has_equivalent:
                    new_commits.append(pair.commit)

            if not new_commits:
                return None  # pragma: no cover

            old_revision = self.m.git.rev_parse(f'{new_commits[0]}~1')

        def finalize() -> None:
            with self.m.context(cwd=checkout.top):
                self.m.git.merge(ref=new_revision, commit=False)

        return RevisionChange(
            new=new_revision,
            old=old_revision,
            finalize=finalize,
        )

    def merge(
        self,
        checkout: checkout_api.CheckoutContext,
        merge_entry: MergeEntry,
    ) -> list[git_roll_util_api.Roll]:
        """Merges a branch into the checked out branch."""
        with self.m.step.nest(merge_entry.branch):
            assert merge_entry.branch

            new_revision = self.m.git_roll_util.resolve_new_revision(
                checkout.options.remote,
                merge_entry.branch,
                checkout.remotes_equivalent,
            )

            change = self.prepare_merge(
                checkout,
                new_revision,
            )

            if not change:
                return []  # pragma: no cover

            try:
                roll = self.m.git_roll_util.get_roll(
                    repo_url=checkout.options.remote,
                    repo_short_name=merge_entry.branch,
                    new_rev=change.new,
                    old_rev=change.old,
                )

                change.finalize()

                return [roll]

            except self.m.git_roll_util.BackwardsRollError:
                props = self.m.step.empty('output property').presentation
                props.properties[merge_entry.branch] = {
                    'remote': checkout.options.remote,
                    'revision': change.new,
                }

                return []