# Copyright 2019 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.
"""Test API for checkout."""

from __future__ import annotations

import urllib
from collections.abc import Sequence

from recipe_engine import post_process, recipe_test_api

from PB.go.chromium.org.luci.buildbucket.proto import common as common_pb
from PB.recipe_modules.pigweed.checkout import options as checkout_options

REPO = 'https://pigweed.googlesource.com/pigweed/pigweed'
MANIFEST_REPO = 'https://pigweed.googlesource.com/pigweed/manifest'

DEFAULT_MANIFEST = """
<?xml version="1.0" encoding="UTF-8"?>
<manifest>
  <remote
    name="default_remote"
    fetch="sso://default"
    />

  <remote
    name="pigweed_remote"
    revision="main"
    fetch=".."
    review="https://pigweed.googlesource.com"
    />

  <remote
    name="pigweed-internal_remote"
    revision="main"
    fetch="https://pigweed-internal.googlesource.com"
    review="https://pigweed-internal.googlesource.com"
    />

  <remote
    name="prefixed"
    fetch="https://foo.googlesource.com/prefix"
    review="https://foo.googlesource.com/prefix"
    />

  <default
    remote="default_remote"
    revision="main"
    />

  <project
    name="default_name"
    path="default_path"
    />

  <project
    name="pigweed_name"
    path="pigweed_path"
    remote="pigweed_remote"
    revision="main"
    />

  <project
    name="pigweed-internal_name"
    path="pigweed-internal_path"
    remote="pigweed-internal_remote"
    />

  <project
    name="pinned"
    path="pinned"
    remote="pigweed_remote"
    revision="0123456789012345678901234567890123456789"
    upstream="main"
    />

  <project
    name="suffix"
    path="prefix/suffix"
    remote="prefixed"
    />
</manifest>
""".strip()


def _project(
    project: str | None = None,
    remote: str | None = None,
    default: str | None = None,
):
    if project is not None:
        return project
    if remote is None:
        assert default
        return default

    assert remote
    project = urllib.parse.urlparse(remote).path.lstrip('/')
    if project.endswith('.git'):
        project = project[0 : -len('.git')]  # pragma: no cover
    return project


def _name(name, project):
    if name is not None:
        return name

    name = project
    if name.endswith('.git'):
        name = name[0 : -len('.git')]  # pragma: no cover
    parts = name.split('/')
    if parts[-1] == 'manifest':
        parts.pop()
    return parts[-1]


class CheckoutTestApi(recipe_test_api.RecipeTestApi):
    """Test API for checkout."""

    @property
    def pigweed_repo(self) -> str:
        return REPO

    @property
    def pigweed_repo_dot_git(self) -> str:
        return f'{REPO}.git'

    @property
    def manifest_repo(self) -> str:
        return MANIFEST_REPO

    def git_options(
        self,
        *,
        remote: str = REPO,
        branch: str = 'main',
        use_trigger: bool = True,
        use_repo: bool = False,
        equivalent_remotes: Sequence[Sequence[str]] = (),
        force_no_rebase: bool = False,
        match_branch: bool = True,
        manifest_file: str = '',
        root_subdirectory: str = '',
        initialize_submodules: bool = True,
        included_submodules: Sequence[str] = (),
        excluded_submodules: Sequence[str] = (),
        manifest_groups: Sequence[str] = (),
        rewrites: Sequence[tuple[str, str]] = (),
        eligible_workspace_paths: Sequence[str] = (),
        use_packfiles: bool = True,
        repo_depth: int | None = None,
        do_not_cache: bool = False,
    ) -> checkout_options.Options:
        props = checkout_options.Options()
        props.remote = remote
        props.branch = branch or 'main'
        props.use_trigger = use_trigger
        props.use_repo = use_repo
        for remotes in equivalent_remotes:
            remote_group = checkout_options.EquivalentRemotes()
            remote_group.remotes.extend(remotes)
            props.equivalent_remotes.extend([remote_group])
        props.force_no_rebase = force_no_rebase
        props.match_branch = match_branch
        props.manifest_file = manifest_file
        props.root_subdirectory = root_subdirectory
        props.initialize_submodules = initialize_submodules
        props.included_submodules.extend(included_submodules)
        props.excluded_submodules.extend(excluded_submodules)
        props.manifest_groups.extend(manifest_groups)
        props.rewrites.extend(
            checkout_options.Rewrite(original=x, final=y) for x, y in rewrites
        )
        props.eligible_workspace_paths.extend(eligible_workspace_paths)
        props.do_not_use_packfiles = not use_packfiles
        if repo_depth:
            props.repo_depth = repo_depth
        props.do_not_cache = do_not_cache
        return props

    def repo_options(
        self,
        remote: str = MANIFEST_REPO,
        use_repo: bool = True,
        manifest_file: str = 'default.xml',
        **kwargs,
    ) -> checkout_options.Options:
        return self.git_options(
            remote=remote,
            use_repo=use_repo,
            manifest_file=manifest_file,
            **kwargs,
        )

    def ci_test_data(
        self,
        git_repo: str = REPO,
        name: str | None = None,
        branch: str | None = None,
        **kwargs,
    ) -> recipe_test_api.StepTestData:
        project = _project(remote=git_repo)
        if branch:
            kwargs['git_ref'] = f'refs/heads/{branch}'
        ret = self.m.buildbucket.ci_build(git_repo=git_repo, **kwargs)
        if branch:
            name = _name(name, project)

            ret += self.override_step_data(
                f'checkout {name}.change data.process gitiles commit.number',
                self.m.json.output(
                    [
                        {
                            '_number': '1234',
                            'branch': branch,
                            'project': project or name,
                        }
                    ]
                ),
            )
        return ret

    def cl(self, host, project, change, patchset=1):
        return common_pb.GerritChange(
            host=host, project=project, change=change, patchset=patchset
        )

    def try_test_data(self, git_repo=REPO, **kwargs):
        return self.m.buildbucket.try_build(git_repo=git_repo, **kwargs)

    def manifest_test_data(self, name='pigweed', raw_xml=DEFAULT_MANIFEST):
        return self.step_data(
            f'checkout {name}.read manifest.read file',
            self.m.file.read_text(raw_xml),
        )

    def cl_branch_parents(
        self,
        branch: str = 'main',
        num_parents: int = 1,
        index: int = 0,
        name: str | None = None,
        message: str = '',
        project: str | None = None,
        remote: str | None = None,
        uploader: str = 'nobody@google.com',
        submit_requirements: dict[str, str] | None = None,
    ):
        project = _project(project, remote, 'pigweed/pigweed')
        name = _name(name, project)

        return self.override_step_data(
            f'checkout {name}.change data.process gerrit changes.{index}.details',
            self.m.json.output(
                {
                    'branch': branch,
                    'current_revision': 'f' * 40,
                    'revisions': {
                        'f' * 40: {
                            '_number': 4,
                            'commit': {
                                'parents': [None for _ in range(num_parents)],
                                'message': message,
                            },
                            'uploader': {'email': uploader},
                        },
                    },
                    'project': project or name,
                    'submit_requirements': [
                        {
                            'name': name,
                            'status': status,
                        }
                        for name, status in (submit_requirements or {}).items()
                    ],
                },
            ),
        )

    def manifest_has_matching_branch(self, branch, name='pigweed'):
        # The contents of the returned JSON data don't matter--they just need to
        # evaluate to True when converted to bool. Gitiles returns an empty
        # dictionary when the branch does not exist.
        return self.m.git.get_remote_branch_head(
            f'checkout {name}.manifest has branch.git ls-remote {branch}',
            'a' * 40,
        )

    def submodule(
        self,
        path,
        remote,
        status='',
        *,
        hash='a' * 40,
        initialized=True,
        branch='main',
        name=None,
        describe='',
        modified=False,
        conflict=False,
        url=None,
        update=None,
        ignore=None,
        shallow=False,
        fetchRecurseSubmodules=None,
    ):
        result = {}
        result['path'] = path
        result['remote'] = remote
        result['hash'] = hash
        result['initialized'] = initialized
        result['branch'] = branch
        result['name'] = name or path
        result['describe'] = describe
        result['modified'] = modified
        result['conflict'] = conflict
        result['url'] = url or remote
        result['update'] = update
        result['ignore'] = ignore
        result['shallow'] = shallow
        result['fetchRecurseSubmodules'] = fetchRecurseSubmodules

        if status == '-':
            result['initialized'] = True
            result['modified'] = False
            result['conflict'] = False
        elif status == '+':
            result['initialized'] = True
            result['modified'] = True
            result['conflict'] = False
        elif status == 'U':  # pragma: no cover
            # Including for completeness but this is not expected to be used.
            result['initialized'] = True
            result['modified'] = True
            result['conflict'] = True
        elif status == ' ':
            result['initialized'] = False
            result['modified'] = False
            result['conflict'] = False

        return result

    def submodules(
        self,
        *submodules,
        prefix='checkout pigweed.',
    ):
        sub_data = {sub['path']: sub for sub in submodules}

        res = self.step_data(
            f'{prefix}submodule status',
            self.m.json.output(sub_data),
        )

        return res

    def all_changes_applied(self):
        return self.some_changes_applied() + self.post_process(
            post_process.DoesNotRunRE, '.*some changes were not applied.*'
        )

    def some_changes_applied(self):
        return self.post_process(
            post_process.DoesNotRunRE, '.*no changes were applied.*'
        )

    def no_changes_applied(self):
        return self.post_process(
            post_process.MustRunRE, '.*no changes were applied.*'
        )

    def change_applied(self, name):
        return self.post_process(
            post_process.DoesNotRunRE, f'.*failed to apply {name}.*'
        ) + self.post_process(
            post_process.MustRunRE, r'.*apply {}.git.*'.format(name)
        )

    def change_not_applied(self, name):
        return self.post_process(
            post_process.MustRunRE, f'.*failed to apply {name}.*'
        ) + self.post_process(
            post_process.DoesNotRunRE, r'.*apply {}.git.*'.format(name)
        )

    def included_submodule(self, name):
        return self.post_process(
            post_process.MustRunRE, r'.*including submodule {}'.format(name)
        ) + self.post_process(
            post_process.DoesNotRunRE, r'.*excluding submodule {}'.format(name)
        )

    def excluded_submodule(self, name):
        return self.post_process(
            post_process.DoesNotRunRE, r'.*including submodule {}'.format(name)
        ) + self.post_process(
            post_process.MustRunRE, r'.*excluding submodule {}'.format(name)
        )

    def multiple_changes_for_same_repo(self):
        return self.post_process(
            post_process.MustRunRE,
            r'.*multiple changes for same repo',
        )