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.

from __future__ import annotations

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

from recipe_engine import recipe_api

if TYPE_CHECKING:  # pragma: no cover
    from typing import Generator
    from recipe_engine import config_types
    from RECIPE_MODULES.fuchsia.git_roll_util import api as git_roll_util_api
    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 Submodule:
    path: str
    name: str
    branch: str
    remote: str = dataclasses.field(default=None)
    dir: config_types.Path = dataclasses.field(default=None)


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


class SubmoduleRollApi(recipe_api.RecipeApi):

    Submodule = Submodule
    RevisionChange = RevisionChange

    def update_pin(
        self,
        checkout: checkout_api.CheckoutContext,
        path: config_types.Path,
        new_revision: str,
    ) -> RevisionChange:
        with self.m.context(cwd=checkout.top):
            self.m.git.submodule_update(
                paths=(path,),
                timeout=checkout.options.submodule_timeout_sec,
            )

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

        with self.m.context(cwd=path):
            self.m.git('git fetch', 'fetch', 'origin', new_revision)
            self.m.git('git checkout', 'checkout', 'FETCH_HEAD')

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

        return RevisionChange(old=old_revision, new=new_revision)

    @functools.cache
    def read_gitmodules(self, path):
        # 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,
    ) -> git_roll_util_api.Roll:
        gitmodules = self.read_gitmodules(checkout.root / '.gitmodules')

        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) as pres:
            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,
            )

            direction = self.m.roll_util.get_roll_direction(
                submodule.dir, change.old, change.new
            )

            # If the primary roll is not necessary or is backwards we can
            # exit immediately and don't need to check deps.
            if not self.m.roll_util.can_roll(direction):
                pres.step_summary_text = 'no roll required'
                self.m.roll_util.skip_roll_step(
                    submodule.remote, change.old, change.new
                )
                return []

            return [
                self.m.roll_util.create_roll(
                    project_name=str(submodule.path),
                    old_revision=change.old,
                    new_revision=change.new,
                    proj_dir=submodule.dir,
                    direction=direction,
                    nest_steps=False,
                )
            ]