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 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.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)

    def update(
        self,
        checkout: checkout_api.CheckoutContext,
        submodule_entries: Sequence[SubmoduleEntry],
    ) -> dict[config_types.Path, roll_util_api.Roll]:
        # Confirm the given path is actually a submodule.
        gitmodules = self.m.file.read_text(
            'read .gitmodules', checkout.root / '.gitmodules'
        )
        # 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))

        rolls: dict[config_types.Path, self.m.roll_util.Roll] = {}

        for entry in submodule_entries:
            submodule = Submodule(
                path=entry.path,
                name=entry.name or entry.path,
                branch=entry.branch,
            )
            submodule.dir = checkout.root / submodule.path

            with self.m.step.nest(submodule.name) as pres:
                section = f'submodule "{submodule.name}"'
                if not parser.has_section(section):
                    sections = parser.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 = parser.get(section, 'branch')
                    except configparser.NoOptionError:
                        submodule.branch = 'main'

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

                change = self.update_pin(
                    checkout,
                    submodule.dir,
                    submodule.branch,
                )

                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 self.m.roll_util.can_roll(direction):
                    rolls[submodule.path] = 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,
                    )

                else:
                    pres.step_summary_text = 'no roll required'
                    self.m.roll_util.skip_roll_step(
                        submodule.remote, change.old, change.new
                    )

        return rolls