# 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]:
        with self.m.step.nest(project.path_to_update):
            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]