recipe_modules/submodule_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.
"""Recipe module for updating Git submodules.

This module provides the `SubmoduleRollApi` class, designed to automate the
process of rolling Git submodules to their latest revisions. It handles:
  - Reading the `.gitmodules` file to determine submodule properties like
    URL, path, and tracked branch.
  - Resolving the latest commit hash for a submodule from its remote repository
    and specified branch.
  - Updating the submodule's pin in the parent repository to this new commit.
  - Performing necessary Git operations like `git submodule update`, `git fetch`,
    and `git checkout` within the submodule directory.
  - Generating `Roll` objects suitable for creating standardized commit messages
    for submodule updates.
"""

from __future__ import annotations

import configparser
import dataclasses
import functools
import io
import re
from collections.abc import Callable

from recipe_engine import config_types, recipe_api

from PB.recipe_modules.pigweed.submodule_roll.submodule import SubmoduleEntry

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 Submodule:
    """Represents a Git submodule with its relevant properties.

    Attributes:
        path: The path to the submodule within the main repository.
        name: The logical name of the submodule (often same as path).
        branch: The branch to track for updates in the submodule.
        remote: The remote URL of the submodule's repository.
        dir: The absolute path to the submodule's directory on disk.
    """

    path: str
    name: str
    branch: str
    remote: str = dataclasses.field(default=None)
    dir: config_types.Path = dataclasses.field(default=None)


@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 commands to update the
                  submodule's pin in the main repository.
    """

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


class SubmoduleRollApi(recipe_api.RecipeApi):
    """API for updating Git submodules to their latest revisions."""

    Submodule = Submodule
    RevisionChange = RevisionChange

    def update_pin(
        self,
        checkout: checkout_api.CheckoutContext,
        path: config_types.Path,
        new_revision: str,
        submodule_timeout_sec: int,
    ) -> RevisionChange:
        """Prepares and returns a RevisionChange object for a submodule update.

        This method first ensures the submodule is initialized and updated locally.
        It then gets the current (old) revision of the submodule.
        A `finalize` function is created that, when called, will fetch the
        `new_revision` into the submodule and check it out.

        Args:
            checkout: The checkout context of the main repository.
            path: The path to the submodule directory.
            new_revision: The new commit hash to update the submodule to.
            submodule_timeout_sec: Timeout in seconds for 'git submodule update'.

        Returns:
            A `RevisionChange` object containing the old and new revisions,
            and a `finalize` callable to apply the update.
        """
        with self.m.context(cwd=checkout.top):
            self.m.git.submodule_update(
                paths=(path,),
                timeout=submodule_timeout_sec,
            )

        old_revision = self.m.checkout.get_revision(
            path, 'get old revision', test_data='1' * 40
        )

        def finalize() -> None:
            with self.m.context(cwd=path):
                self.m.git('git fetch', 'fetch', 'origin', new_revision)
                self.m.git('git checkout', 'checkout', 'FETCH_HEAD')

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

    @functools.cache
    def read_gitmodules(
        self, path: config_types.Path
    ) -> configparser.RawConfigParser:
        """Reads and parses the .gitmodules file.

        The result is cached to avoid redundant file reads and parsing.

        Args:
            path: The path to the .gitmodules file.

        Returns:
            A `configparser.RawConfigParser` instance loaded with the
            contents of the .gitmodules file.
        """
        # Confirm the given path is actually a submodule.
        gitmodules = self.m.file.read_text('read .gitmodules', path)
        # Example .gitmodules file:
        # [submodule "third_party/pigweed"]
        #   path = third_party/pigweed
        #   url = https://pigweed.googlesource.com/pigweed/pigweed

        # configparser doesn't like leading whitespace on lines, despite what
        # its documentation says.
        gitmodules = re.sub(r'\n\s+', '\n', gitmodules)
        parser = configparser.RawConfigParser()
        parser.readfp(io.StringIO(gitmodules))
        return parser

    def update(
        self,
        checkout: checkout_api.CheckoutContext,
        submodule_entry: SubmoduleEntry,
    ) -> list[git_roll_util_api.Roll]:
        """Updates a Git submodule to the latest commit on its tracked branch.

        This method reads the .gitmodules file to find the submodule's remote URL
        and branch. It then resolves the latest commit on that branch and uses
        `update_pin` to prepare the update. If an update is performed, it
        generates a `Roll` object for commit message formatting.

        Args:
            checkout: The checkout context of the main repository.
            submodule_entry: A protobuf message defining the submodule to update,
                             including its path, name, and optionally branch.

        Returns:
            A list containing a `Roll` object if the submodule was updated,
            or an empty list if no update was needed (e.g., already up-to-date
            or a backwards roll).

        Raises:
            StepFailure: If the submodule specified in `submodule_entry` is not
                         found in the .gitmodules file.
        """
        submodule = Submodule(
            path=submodule_entry.path,
            name=submodule_entry.name or submodule_entry.path,
            branch=submodule_entry.branch,
        )
        submodule.dir = checkout.root / submodule.path

        with self.m.step.nest(submodule.name):
            gitmodules = self.read_gitmodules(checkout.root / '.gitmodules')

            section = f'submodule "{submodule.name}"'
            if not gitmodules.has_section(section):
                sections = gitmodules.sections()
                submodules = sorted(
                    re.sub(r'^.*"(.*)"$', r'\1', x) for x in sections
                )
                raise self.m.step.StepFailure(
                    'no submodule "{}" (submodules: {})'.format(
                        submodule.name,
                        ', '.join('"{}"'.format(x) for x in submodules),
                    )
                )

            if not submodule.branch:
                try:
                    submodule.branch = gitmodules.get(section, 'branch')
                except configparser.NoOptionError:
                    submodule.branch = 'main'

            submodule.remote = self.m.git_roll_util.normalize_remote(
                gitmodules.get(section, 'url'),
                checkout.options.remote,
            )

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

            change = self.update_pin(
                checkout,
                submodule.dir,
                new_revision,
                submodule_entry.timeout_sec or 10 * 60,
            )

            try:
                roll = self.m.git_roll_util.get_roll(
                    repo_url=submodule.remote,
                    repo_short_name=submodule.path,
                    old_rev=change.old,
                    new_rev=change.new,
                )

                change.finalize()

                self.m.bazel_roll.update_module_bazel_lock(checkout)

                return [roll]

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

                return []