# 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.
"""Roll a project in an Android Repo Tool workspace."""

from __future__ import annotations

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

import attrs
from PB.recipes.pigweed.repo_roller import InputProperties
from PB.recipe_modules.pigweed.checkout.options import (
    Options as CheckoutOptions,
)
from recipe_engine import config_types, recipe_test_api
from RECIPE_MODULES.pigweed.roll_util import api as roll_util_api

DEPS = [
    'fuchsia/auto_roller',
    'fuchsia/sso',
    'pigweed/checkout',
    'pigweed/roll_util',
    'recipe_engine/buildbucket',
    'recipe_engine/file',
    'recipe_engine/path',
    'recipe_engine/properties',
    'recipe_engine/raw_io',
    'recipe_engine/step',
]

PROPERTIES = InputProperties


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)


def _is_branch(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(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)


@dataclasses.dataclass
class _RevisionChange:
    dir: config_types.Path
    old: str
    new: str
    direction: roll_util_api.Direction


def _update_project(api, checkout, path_to_update) -> _RevisionChange | None:
    new_revision = None

    # Try to get new_revision from the trigger.
    bb_remote = None
    commit = api.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(api.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'] = api.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'] == 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 api.step.StepFailure(
                    "repo paths don't match: {} from manifest "
                    "and {} from buildbucket".format(manifest_remote, bb_remote)
                )
            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 api.step.StepFailure(
            'cannot find "{}" in manifest (found {})'.format(
                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 _is_branch(proj_attrib['revision']):
        proj_branch = proj_attrib['revision']
    else:
        proj_branch = 'main'

    # If we still don't have a revision then it wasn't in the trigger. (Perhaps
    # this was manually triggered.) In this case we need to determine the
    # latest revision of this repository. The use_trigger flag should have no
    # effect but using it to be explicit. Checking out even if not needed so
    # commit messages can be collected later.
    proj_dir = api.path.start_dir / 'project'
    proj_checkout = api.checkout(
        CheckoutOptions(
            remote=manifest_remote, branch=proj_branch, use_trigger=False
        ),
        root=proj_dir,
    )
    if new_revision is None:
        new_revision = api.checkout.get_revision(proj_dir)

    assert new_revision
    new_revision = str(new_revision)

    # 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 _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 api.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

    direction = api.roll_util.Direction.FORWARD
    if not _is_branch(old_revision):
        direction = api.roll_util.get_roll_direction(
            proj_dir, old_revision, new_revision
        )

    if not api.roll_util.can_roll(direction):
        api.roll_util.skip_roll_step(
            manifest_remote, old_revision, new_revision
        )
        return

    _order_attributes(root)

    api.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 _RevisionChange(
        dir=proj_dir,
        old=old_revision,
        new=new_revision,
        direction=direction,
    )


def RunSteps(api, props):  # pylint: disable=invalid-name
    path_to_update = str(props.path_to_update)
    cc_authors_on_rolls = props.cc_authors_on_rolls
    cc_reviewers_on_rolls = props.cc_reviewers_on_rolls
    cc_domains = props.cc_domains
    always_cc = props.always_cc

    props.checkout_options.use_trigger = False
    checkout = api.checkout(props.checkout_options)

    change = _update_project(api, checkout, path_to_update)
    if not change:
        return

    roll = api.roll_util.create_roll(
        project_name=path_to_update,
        old_revision=change.old,
        new_revision=change.new,
        proj_dir=change.dir,
        direction=change.direction,
    )

    authors = api.roll_util.authors(roll)

    max_commits_for_ccing = props.max_commits_for_ccing or 10
    if len(roll.commits) <= max_commits_for_ccing:
        cc = set()
        if cc_authors_on_rolls:
            cc.update(authors)
        if cc_reviewers_on_rolls:
            cc.update(api.roll_util.reviewers(roll))

        def include_cc(email):
            return api.roll_util.include_cc(
                email, cc_domains, checkout.gerrit_host()
            )

        # include_cc() writes steps, so we want things sorted before calling it.
        cc = sorted(set(cc))
        cc_emails = [x.email for x in cc if include_cc(x)]

        if always_cc:
            props.auto_roller_options.cc_emails.extend(cc_emails)
        else:
            props.auto_roller_options.cc_on_failure_emails.extend(cc_emails)

    author_override = None
    with api.step.nest('authors') as pres:
        pres.step_summary_text = repr(authors)
    if len(authors) == 1 and props.forge_author:
        author_override = attrs.asdict(
            api.roll_util.fake_author(next(iter(authors)))
        )

    # merge auto_roller_options and override_auto_roller_options.
    complete_auto_roller_options = api.roll_util.merge_auto_roller_overrides(
        props.auto_roller_options, props.override_auto_roller_options
    )

    change = api.auto_roller.attempt_roll(
        complete_auto_roller_options,
        repo_dir=checkout.root,
        commit_message=api.roll_util.message(roll),
        author_override=author_override,
    )

    return api.auto_roller.raw_result(change)


def GenTests(api) -> Generator[recipe_test_api.TestData, None, None]:
    """Create tests."""

    manifest = 'https://host.googlesource.com/manifest'

    def properties(api, path, dry_run=True, equivalent_remotes=(), **kwargs):
        props = InputProperties(**kwargs)
        props.checkout_options.CopyFrom(
            api.checkout.git_options(
                remote=manifest,
                equivalent_remotes=equivalent_remotes,
                match_branch=True,
            )
        )
        props.forge_author = True
        props.path_to_update = path
        props.auto_roller_options.CopyFrom(
            api.auto_roller.Options(dry_run=dry_run, remote=manifest)
        )

        return api.properties(props)

    def read_step_data():
        return api.step_data(
            'read manifest',
            api.file.read_text(
                """
<?xml version="1.0" encoding="UTF-8"?>
<manifest>
  <!-- single-line comment -->
  <remote name="foo" fetch="sso://foo" review="sso://foo" />
  <remote name="bar" fetch="sso://bar" review="sso://bar" />
  <remote name="host" fetch=".." review="sso://host" />
  <remote name="dotdot-prefix" fetch="../prefix" review="sso://host/prefix" />
  <remote name="host-prefix" fetch="sso://host/prefix"
    review="sso://host/prefix" />
  <default remote="bar" />
  <project name="a" path="a1" remote="foo"
    revision="1111111111111111111111111111111111111111" upstream="main"/>
  <project name="b" path="b2"
    revision="2222222222222222222222222222222222222222" upstream="main"/>
  <!--
  multi
  line
  comment
  -->
  <project name="c" path="c3" revision="main"/>
  <project name="d" path="d4"
    revision="0000000000111111111122222222223333333333"/>
  <project name="e5" revision="refs/tags/e"/>
  <project name="f" path="f6" remote="host" revision="main"/>
  <project name="g" path="g7" remote="dotdot-prefix" revision="main"/>
  <project name="h" path="h8" remote="host-prefix" revision="main"/>
</manifest>
""".lstrip()
            ),
        )

    def commit_data(name):
        return api.roll_util.commit_data(
            name, api.roll_util.commit('a' * 40, 'foo\nbar')
        )

    yield api.test(
        'success',
        properties(api, path='a1', cc_authors_on_rolls=True, always_cc=True),
        api.checkout.ci_test_data(git_repo='https://foo.googlesource.com/a'),
        commit_data('a1'),
        read_step_data(),
        api.roll_util.forward_roll(),
        api.auto_roller.dry_run_success(),
    )

    yield api.test(
        'name-not-found',
        properties(api, path='missing'),
        api.checkout.ci_test_data(git_repo='https://bar.googlesource.com/b'),
        read_step_data(),
        status='FAILURE',
    )

    yield api.test(
        'equivalent',
        properties(
            api,
            path='b2',
            equivalent_remotes=(
                (
                    'https://equiv.googlesource.com/b',
                    'https://bar.googlesource.com/b',
                ),
            ),
            cc_reviewers_on_rolls=True,
        ),
        api.checkout.ci_test_data(git_repo='https://equiv.googlesource.com/b'),
        commit_data('b2'),
        read_step_data(),
        api.roll_util.forward_roll(),
        api.auto_roller.dry_run_success(),
    )

    yield api.test(
        'upstream-not-set',
        properties(api, path='c3'),
        api.checkout.ci_test_data(git_repo='https://bar.googlesource.com/c'),
        commit_data('c3'),
        read_step_data(),
        api.auto_roller.dry_run_success(),
    )

    yield api.test(
        'upstream-not-set-revision-not-branch',
        properties(api, path='d4'),
        api.checkout.ci_test_data(git_repo='https://bar.googlesource.com/c'),
        read_step_data(),
        status='FAILURE',
    )

    yield api.test(
        'no-trigger-with-upstream',
        properties(api, path='a1'),
        commit_data('a1'),
        read_step_data(),
        api.step_data(
            'git log',
            stdout=api.raw_io.output_text('hash-from-special-checkout'),
        ),
        api.roll_util.forward_roll(),
        api.auto_roller.dry_run_success(),
    )

    yield api.test(
        'no-trigger-with-revision-branch',
        properties(api, path='c3'),
        commit_data('c3'),
        read_step_data(),
        api.step_data(
            'git log',
            stdout=api.raw_io.output_text('hash-from-special-checkout'),
        ),
        api.auto_roller.dry_run_success(),
    )

    yield api.test(
        'no-trigger-with-revision-hash',
        properties(api, path='d4'),
        read_step_data(),
        api.step_data(
            'git log',
            stdout=api.raw_io.output_text('hash-from-special-checkout'),
        ),
        status='FAILURE',
    )

    yield api.test(
        'no-trigger-with-revision-tag',
        properties(api, path='e5'),
        read_step_data(),
        api.step_data(
            'git log',
            stdout=api.raw_io.output_text('hash-from-special-checkout'),
        ),
        status='FAILURE',
    )

    yield api.test(
        'backwards',
        properties(api, path='a1'),
        api.checkout.ci_test_data(git_repo='https://foo.googlesource.com/a'),
        read_step_data(),
        api.roll_util.backward_roll(),
    )

    yield api.test(
        'host-dot-dot',
        properties(api, path='f6'),
        api.checkout.ci_test_data(git_repo='https://host.googlesource.com/f'),
        commit_data('f6'),
        read_step_data(),
        api.auto_roller.dry_run_success(),
    )

    yield api.test(
        'dotdot-prefix',
        properties(api, path='g7'),
        api.checkout.ci_test_data(
            git_repo='https://host.googlesource.com/prefix/g'
        ),
        commit_data('g7'),
        read_step_data(),
        api.auto_roller.dry_run_success(),
    )

    yield api.test(
        'host-prefix',
        properties(api, path='h8'),
        api.checkout.ci_test_data(
            git_repo='https://host.googlesource.com/prefix/h'
        ),
        commit_data('h8'),
        read_step_data(),
        api.auto_roller.dry_run_success(),
    )