recipe_modules/repo_roll/api.py

# Copyright 2020 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.
"""Update an Android Repo Tool project."""

from __future__ import annotations

import collections
import dataclasses
import json
import re
from typing import Sequence, TYPE_CHECKING
import urllib
import xml.etree.ElementTree

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

if TYPE_CHECKING:  # pragma: no cover
    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 PB.recipe_modules.pigweed.repo_roll import Project


class _TreeBuilder(xml.etree.ElementTree.TreeBuilder):
    def comment(self, data):
        self.start(xml.etree.ElementTree.Comment, {})
        self.data(data)
        self.end(xml.etree.ElementTree.Comment)


class RepoRollApi(recipe_api.RecipeApi):
    """Update an Android Repo Tool project."""

    def is_branch(self, revision):
        """Return True if revision appears to be a branch name."""
        if re.search(r'^[0-9a-fA-F]{40}$', revision):
            return False
        return not revision.startswith('refs/tags/')

    # ElementTree orders attributes differently in Python 2 and 3, but if given
    # attributes in a specific order it preserves that order.
    def _order_attributes(self, root):
        for el in root.iter():
            if len(el.attrib) > 1:
                new_attrib = sorted(el.attrib.items())
                el.attrib.clear()
                el.attrib.update(new_attrib)

    def update_project(
        self,
        checkout: checkout_api.CheckoutContext,
        project: Project,
    ) -> list[git_roll_util_api.Roll]:
        new_revision = None

        # Try to get new_revision from the trigger.
        bb_remote = None
        commit = self.m.buildbucket.build.input.gitiles_commit
        if commit and commit.project:
            new_revision = commit.id
            host = commit.host
            bb_remote = f'https://{host}/{commit.project}'

        tree = _TreeBuilder()
        parser = xml.etree.ElementTree.XMLParser(target=tree)
        parser.feed(
            self.m.file.read_text('read manifest', checkout.manifest_path)
        )
        parser.close()
        root = tree.close()

        defaults = {}
        for default in root.findall('default'):
            defaults.update(default.attrib)

        remotes = {}
        for rem in root.findall('remote'):
            remotes[rem.attrib['name']] = rem.attrib

        # Search for path_to_update and check the repository it refers to
        # matches the triggering repository. This check is mostly a config check
        # and shouldn't fail in real runs unless somebody moves around project
        # paths in the manifest without updating the builder definition.
        paths_found = []
        proj = None
        proj_attrib = {}

        for proj in root.findall('project'):
            # Sometimes projects don't define something in which case we need to
            # fall back on the remote specified by the project or the default
            # for the entire manifest.
            proj_attrib = {}
            proj_attrib.update(defaults)
            remote_name = proj.attrib.get(
                'remote', defaults.get('remote', None)
            )
            assert remote_name
            proj_attrib.update(remotes[remote_name])
            proj_attrib.update(proj.attrib)
            proj_attrib['fetch'] = self.m.sso.sso_to_https(proj_attrib['fetch'])

            # Apparently if path is left off name is used. This is mildly
            # infuriating, but easy to work around.
            if 'path' not in proj_attrib:
                proj_attrib['path'] = proj_attrib['name']

            paths_found.append(proj_attrib['path'])
            if proj_attrib['path'] == project.path_to_update:
                fetch_host = proj_attrib['fetch'].strip('/')
                if fetch_host.startswith('..'):
                    fetch_host = urllib.parse.urljoin(
                        checkout.options.remote,
                        fetch_host,
                    )

                manifest_remote = (
                    fetch_host.rstrip('/')
                    + '/'
                    + proj_attrib['name'].strip('/')
                )
                if bb_remote and not checkout.remotes_equivalent(
                    manifest_remote, bb_remote
                ):
                    raise self.m.step.StepFailure(
                        f"repo paths don't match: {manifest_remote!r} from "
                        f'manifest and {bb_remote!r} from buildbucket'
                    )
                break

            # Reset proj_attrib to None if this entry wasn't a match, so if this
            # is the last iteration the condition a few lines down will work.
            proj_attrib = {}

        if not proj_attrib:
            raise self.m.step.StepFailure(
                'cannot find "{}" in manifest (found {})'.format(
                    project.path_to_update,
                    ', '.join('"{}"'.format(x) for x in paths_found),
                )
            )

        if 'upstream' in proj_attrib:
            proj_branch = proj_attrib['upstream']
        elif 'revision' in proj_attrib and self.is_branch(
            proj_attrib['revision']
        ):
            proj_branch = proj_attrib['revision']
        else:
            proj_branch = 'main'

        new_revision = self.m.git_roll_util.resolve_new_revision(
            manifest_remote,
            proj_branch,
            checkout.remotes_equivalent,
        )

        # If upstream not set we may be transitioning from tracking a branch to
        # rolling. In that case set upstream to be the revision, but only if the
        # revision appears to be a branch.
        if 'upstream' not in proj_attrib:
            if self.is_branch(proj_attrib['revision']):
                # Explicitly update both proj.attrib and proj_attrib to minimize
                # confusion if these statements are moved around later.
                proj.attrib['upstream'] = proj_attrib['revision']
                proj_attrib['upstream'] = proj.attrib['upstream']
            else:
                raise self.m.step.StepFailure(
                    'upstream not set and revision is not a branch, aborting'
                )

        old_revision = proj_attrib['revision']
        # Explicitly update both proj.attrib and proj_attrib to minimize
        # confusion.
        proj_attrib['revision'] = proj.attrib['revision'] = new_revision

        try:
            roll = self.m.git_roll_util.get_roll(
                repo_url=manifest_remote,
                repo_short_name=project.path_to_update,
                old_rev=old_revision,
                new_rev=new_revision,
            )

        except self.m.git_roll_util.BackwardsRollError:
            return []

        self._order_attributes(root)

        self.m.file.write_text(
            'write manifest',
            checkout.manifest_path,
            '<?xml version="1.0" encoding="UTF-8"?>\n{}\n'.format(
                xml.etree.ElementTree.tostring(root).decode(),
            ),
        )

        return [roll]