/brz/remove-bazaar

# Copyright (C) 2007,2012 Canonical Ltd

# Copyright (C) 2009-2018 Jelmer Vernooij <jelmer@jelmer.uk>

#

# This program is free software; you can redistribute it and/or modify

# it under the terms of the GNU General Public License as published by

# the Free Software Foundation; either version 2 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU General Public License for more details.

#

# You should have received a copy of the GNU General Public License

# along with this program; if not, write to the Free Software

# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA

"""An adapter between a Git Branch and a Bazaar Branch"""

import contextlib

from io import BytesIO

from collections import defaultdict

from dulwich.config import (

    ConfigFile as GitConfigFile,

    parse_submodules,

    )

from dulwich.objects import (

    NotCommitError,

    ZERO_SHA,

    )

from dulwich.repo import check_ref_format

from .. import (

    branch,

    config,

    controldir,

    errors,

    lock,

    repository as _mod_repository,

    revision,

    trace,

    transport,

    urlutils,

    )

from ..foreign import ForeignBranch

from ..revision import (

    NULL_REVISION,

    )

from ..tag import (

    Tags,

    InterTags,

    )

from ..trace import (

    is_quiet,

    mutter,

    warning,

    )

from .config import (

    GitBranchConfig,

    GitBranchStack,

    )

from .errors import (

    NoPushSupport,

    )

from .push import (

    remote_divergence,

    )

from .refs import (

    branch_name_to_ref,

    is_tag,

    ref_to_branch_name,

    ref_to_tag_name,

    remote_refs_dict_to_tag_refs,

    tag_name_to_ref,

    )

from .unpeel_map import (

    UnpeelMap,

    )

from .urls import (

    git_url_to_bzr_url,

    bzr_url_to_git_url,

    )

def _calculate_revnos(branch):

    if branch._format.stores_revno():

        return True

    config = branch.get_config_stack()

    return config.get('calculate_revnos')

class GitPullResult(branch.PullResult):

    """Result of a pull from a Git branch."""

    def _lookup_revno(self, revid):

        if not isinstance(revid, bytes):

            raise TypeError(revid)

        if not _calculate_revnos(self.target_branch):

            return None

        # Try in source branch first, it'll be faster

        with self.target_branch.lock_read():

            return self.target_branch.revision_id_to_revno(revid)

    @property

    def old_revno(self):

        return self._lookup_revno(self.old_revid)

    @property

    def new_revno(self):

        return self._lookup_revno(self.new_revid)

class InterTagsFromGitToRemoteGit(InterTags):

    @classmethod

    def is_compatible(klass, source, target):

        if not isinstance(source, GitTags):

            return False

        if not isinstance(target, GitTags):

            return False

        if getattr(target.branch.repository, "_git", None) is not None:

            return False

        return True

    def merge(self, overwrite=False, ignore_master=False, selector=None):

        if self.source.branch.repository.has_same_location(self.target.branch.repository):

            return {}, []

        updates = {}

        conflicts = []

        source_tag_refs = self.source.branch.get_tag_refs()

        def get_changed_refs(old_refs):

            ret = dict(old_refs)

            for ref_name, tag_name, peeled, unpeeled in (

                    source_tag_refs.iteritems()):

                if selector and not selector(tag_name):

                    continue

                if old_refs.get(ref_name) == unpeeled:

                    pass

                elif overwrite or ref_name not in old_refs:

                    ret[ref_name] = unpeeled

                    updates[tag_name] = self.target.branch.repository.lookup_foreign_revision_id(

                        peeled)

                    self.target.branch._tag_refs = None

                else:

                    conflicts.append(

                        (tag_name,

                         self.repository.lookup_foreign_revision_id(peeled),

                         self.target.branch.repository.lookup_foreign_revision_id(

                             old_refs[ref_name])))

            return ret

        self.target.branch.repository.controldir.send_pack(

            get_changed_refs, lambda have, want: [])

        return updates, set(conflicts)

class InterTagsFromGitToLocalGit(InterTags):

    @classmethod

    def is_compatible(klass, source, target):

        if not isinstance(source, GitTags):

            return False

        if not isinstance(target, GitTags):

            return False

        if getattr(target.branch.repository, "_git", None) is None:

            return False

        return True

    def merge(self, overwrite=False, ignore_master=False, selector=None):

        if self.source.branch.repository.has_same_location(self.target.branch.repository):

            return {}, []

        conflicts = []

        updates = {}

        source_tag_refs = self.source.branch.get_tag_refs()

        target_repo = self.target.branch.repository

        for ref_name, tag_name, peeled, unpeeled in source_tag_refs:

            if selector and not selector(tag_name):

                continue

            if target_repo._git.refs.get(ref_name) == unpeeled:

                pass

            elif overwrite or ref_name not in target_repo._git.refs:

                try:

                    updates[tag_name] = (

                        target_repo.lookup_foreign_revision_id(peeled))

                except KeyError:

                    trace.warning('%s does not point to a valid object',

                                  tag_name)

                    continue

                except NotCommitError:

                    trace.warning('%s points to a non-commit object',

                                  tag_name)

                    continue

                target_repo._git.refs[ref_name] = unpeeled or peeled

                self.target.branch._tag_refs = None

            else:

                try:

                    source_revid = self.source.branch.repository.lookup_foreign_revision_id(

                        peeled)

                    target_revid = target_repo.lookup_foreign_revision_id(

                        target_repo._git.refs[ref_name])

                except KeyError:

                    trace.warning('%s does not point to a valid object',

                                  ref_name)

                    continue

                except NotCommitError:

                    trace.warning('%s points to a non-commit object',

                                  tag_name)

                    continue

                conflicts.append((tag_name, source_revid, target_revid))

        return updates, set(conflicts)

class InterTagsFromGitToNonGit(InterTags):

    @classmethod

    def is_compatible(klass, source, target):

        if not isinstance(source, GitTags):

            return False

        if isinstance(target, GitTags):

            return False

        return True

    def merge(self, overwrite=False, ignore_master=False, selector=None):

        """See Tags.merge_to."""

        source_tag_refs = self.source.branch.get_tag_refs()

        if ignore_master:

            master = None

        else:

            master = self.target.branch.get_master_branch()

        with contextlib.ExitStack() as es:

            if master is not None:

                es.enter_context(master.lock_write())

            updates, conflicts = self._merge_to(

                self.target, source_tag_refs, overwrite=overwrite,

                selector=selector)

            if master is not None:

                extra_updates, extra_conflicts = self._merge_to(

                    master.tags, overwrite=overwrite,

                    source_tag_refs=source_tag_refs,

                    ignore_master=ignore_master, selector=selector)

                updates.update(extra_updates)

                conflicts.update(extra_conflicts)

            return updates, conflicts

    def _merge_to(self, to_tags, source_tag_refs, overwrite=False,

                  selector=None):

        unpeeled_map = defaultdict(set)

        conflicts = []

        updates = {}

        result = dict(to_tags.get_tag_dict())

        for ref_name, tag_name, peeled, unpeeled in source_tag_refs:

            if selector and not selector(tag_name):

                continue

            if unpeeled is not None:

                unpeeled_map[peeled].add(unpeeled)

            try:

                bzr_revid = self.source.branch.lookup_foreign_revision_id(peeled)

            except NotCommitError:

                continue

            if result.get(tag_name) == bzr_revid:

                pass

            elif tag_name not in result or overwrite:

                result[tag_name] = bzr_revid

                updates[tag_name] = bzr_revid

            else:

                conflicts.append((tag_name, bzr_revid, result[tag_name]))

        to_tags._set_tag_dict(result)

        if len(unpeeled_map) > 0:

            map_file = UnpeelMap.from_repository(to_tags.branch.repository)

            map_file.update(unpeeled_map)

            map_file.save_in_repository(to_tags.branch.repository)

        return updates, set(conflicts)

InterTags.register_optimiser(InterTagsFromGitToRemoteGit)

InterTags.register_optimiser(InterTagsFromGitToLocalGit)

InterTags.register_optimiser(InterTagsFromGitToNonGit)

class GitTags(Tags):

    """Ref-based tag dictionary."""

    def __init__(self, branch):

        self.branch = branch

        self.repository = branch.repository

    def get_tag_dict(self):

        ret = {}

        for (ref_name, tag_name, peeled, unpeeled) in (

                self.branch.get_tag_refs()):

            try:

                bzr_revid = self.branch.lookup_foreign_revision_id(peeled)

            except NotCommitError:

                continue

            else:

                ret[tag_name] = bzr_revid

        return ret

    def lookup_tag(self, tag_name):

        """Return the referent string of a tag"""

        # TODO(jelmer): Replace with something more efficient for local tags.

        td = self.get_tag_dict()

        try:

            return td[tag_name]

        except KeyError:

            raise errors.NoSuchTag(tag_name)

class LocalGitTagDict(GitTags):

    """Dictionary with tags in a local repository."""

    def __init__(self, branch):

        super(LocalGitTagDict, self).__init__(branch)

        self.refs = self.repository.controldir._git.refs

    def _set_tag_dict(self, to_dict):

        extra = set(self.refs.allkeys())

        for k, revid in to_dict.items():

            name = tag_name_to_ref(k)

            if name in extra:

                extra.remove(name)

            try:

                self.set_tag(k, revid)

            except errors.GhostTagsNotSupported:

                pass

        for name in extra:

            if is_tag(name):

                del self.repository._git[name]

    def set_tag(self, name, revid):

        try:

            git_sha, mapping = self.branch.lookup_bzr_revision_id(revid)

        except errors.NoSuchRevision:

            raise errors.GhostTagsNotSupported(self)

        self.refs[tag_name_to_ref(name)] = git_sha

        self.branch._tag_refs = None

    def delete_tag(self, name):

        ref = tag_name_to_ref(name)

        if ref not in self.refs:

            raise errors.NoSuchTag(name)

        del self.refs[ref]

        self.branch._tag_refs = None

class GitBranchFormat(branch.BranchFormat):

    def network_name(self):

        return b"git"

    def supports_tags(self):

        return True

    def supports_leaving_lock(self):

        return False

    def supports_tags_referencing_ghosts(self):

        return False

    def tags_are_versioned(self):

        return False

    def get_foreign_tests_branch_factory(self):

        from .tests.test_branch import ForeignTestsBranchFactory

        return ForeignTestsBranchFactory()

    def make_tags(self, branch):

        try:

            return branch.tags

        except AttributeError:

            pass

        if getattr(branch.repository, "_git", None) is None:

            from .remote import RemoteGitTagDict

            return RemoteGitTagDict(branch)

        else:

            return LocalGitTagDict(branch)

    def initialize(self, a_controldir, name=None, repository=None,

                   append_revisions_only=None):

        raise NotImplementedError(self.initialize)

    def get_reference(self, controldir, name=None):

        return controldir.get_branch_reference(name=name)

    def set_reference(self, controldir, name, target):

        return controldir.set_branch_reference(target, name)

    def stores_revno(self):

        """True if this branch format store revision numbers."""

        return False

    supports_reference_locations = False

class LocalGitBranchFormat(GitBranchFormat):

    def get_format_description(self):

        return 'Local Git Branch'

    @property

    def _matchingcontroldir(self):

        from .dir import LocalGitControlDirFormat

        return LocalGitControlDirFormat()

    def initialize(self, a_controldir, name=None, repository=None,

                   append_revisions_only=None):

        from .dir import LocalGitDir

        if not isinstance(a_controldir, LocalGitDir):

            raise errors.IncompatibleFormat(self, a_controldir._format)

        return a_controldir.create_branch(

            repository=repository, name=name,

            append_revisions_only=append_revisions_only)

class GitBranch(ForeignBranch):

    """An adapter to git repositories for bzr Branch objects."""

    @property

    def control_transport(self):

        return self._control_transport

    @property

    def user_transport(self):

        return self._user_transport

    def __init__(self, controldir, repository, ref, format):

        self.repository = repository

        self._format = format

        self.controldir = controldir

        self._lock_mode = None

        self._lock_count = 0

        super(GitBranch, self).__init__(repository.get_mapping())

        self.ref = ref

        self._head = None

        self._user_transport = controldir.user_transport.clone('.')

        self._control_transport = controldir.control_transport.clone('.')

        self._tag_refs = None

        params = {}

        try:

            self.name = ref_to_branch_name(ref)

        except ValueError:

            self.name = None

            if self.ref is not None:

                params = {"ref": urlutils.escape(self.ref)}

        else:

            if self.name != "":

                params = {"branch": urlutils.escape(self.name)}

        for k, v in params.items():

            self._user_transport.set_segment_parameter(k, v)

            self._control_transport.set_segment_parameter(k, v)

        self.base = controldir.user_transport.base

    def _get_checkout_format(self, lightweight=False):

        """Return the most suitable metadir for a checkout of this branch.

        Weaves are used if this branch's repository uses weaves.

        """

        if lightweight:

            return controldir.format_registry.make_controldir("git")

        else:

            return controldir.format_registry.make_controldir("default")

    def get_child_submit_format(self):

        """Return the preferred format of submissions to this branch."""

        ret = self.get_config_stack().get("child_submit_format")

        if ret is not None:

            return ret

        return "git"

    def get_config(self):

        return GitBranchConfig(self)

    def get_config_stack(self):

        return GitBranchStack(self)

    def _get_nick(self, local=False, possible_master_transports=None):

        """Find the nick name for this branch.

        :return: Branch nick

        """

        if getattr(self.repository, '_git', None):

            cs = self.repository._git.get_config_stack()

            try:

                return cs.get((b"branch", self.name.encode('utf-8')),

                              b"nick").decode("utf-8")

            except KeyError:

                pass

        return self.name or u"HEAD"

    def _set_nick(self, nick):

        cf = self.repository._git.get_config()

        cf.set((b"branch", self.name.encode('utf-8')),

               b"nick", nick.encode("utf-8"))

        f = BytesIO()

        cf.write_to_file(f)

        self.repository._git._put_named_file('config', f.getvalue())

    nick = property(_get_nick, _set_nick)

    def __repr__(self):

        return "<%s(%r, %r)>" % (self.__class__.__name__, self.repository.base,

                                 self.name)

    def generate_revision_history(self, revid, last_rev=None,

                                  other_branch=None):

        if last_rev is not None:

            graph = self.repository.get_graph()

            if not graph.is_ancestor(last_rev, revid):

                # our previous tip is not merged into stop_revision

                raise errors.DivergedBranches(self, other_branch)

        self.set_last_revision(revid)

    def lock_write(self, token=None):

        if token is not None:

            raise errors.TokenLockingNotSupported(self)

        if self._lock_mode:

            if self._lock_mode == 'r':

                raise errors.ReadOnlyError(self)

            self._lock_count += 1

        else:

            self._lock_ref()

            self._lock_mode = 'w'

            self._lock_count = 1

        self.repository.lock_write()

        return lock.LogicalLockResult(self.unlock)

    def leave_lock_in_place(self):

        raise NotImplementedError(self.leave_lock_in_place)

    def dont_leave_lock_in_place(self):

        raise NotImplementedError(self.dont_leave_lock_in_place)

    def get_stacked_on_url(self):

        # Git doesn't do stacking (yet...)

        raise branch.UnstackableBranchFormat(self._format, self.base)

    def _get_push_origin(self, cs):

        """Get the name for the push origin.

        The exact behaviour is documented in the git-config(1) manpage.

        """

        try:

            return cs.get((b'branch', self.name.encode('utf-8')), b'pushRemote')

        except KeyError:

            try:

                return cs.get((b'branch', ), b'remote')

            except KeyError:

                try:

                    return cs.get((b'branch', self.name.encode('utf-8')), b'remote')

                except KeyError:

                    return b'origin'

    def _get_origin(self, cs):

        try:

            return cs.get((b'branch', self.name.encode('utf-8')), b'remote')

        except KeyError:

            return b'origin'

    def _get_related_push_branch(self, cs):

        remote = self._get_push_origin(cs)

        try:

            location = cs.get((b"remote", remote), b"url")

        except KeyError:

            return None

        return git_url_to_bzr_url(location.decode('utf-8'), ref=self.ref)

    def _get_related_merge_branch(self, cs):

        remote = self._get_origin(cs)

        try:

            location = cs.get((b"remote", remote), b"url")

        except KeyError:

            return None

        try:

            ref = cs.get((b"branch", remote), b"merge")

        except KeyError:

            ref = self.ref

        return git_url_to_bzr_url(location.decode('utf-8'), ref=ref)

    def _get_parent_location(self):

        """See Branch.get_parent()."""

        cs = self.repository._git.get_config_stack()

        return self._get_related_merge_branch(cs)

    def _write_git_config(self, cs):

        f = BytesIO()

        cs.write_to_file(f)

        self.repository._git._put_named_file('config', f.getvalue())

    def set_parent(self, location):

        cs = self.repository._git.get_config()

        remote = self._get_origin(cs)

        this_url = urlutils.strip_segment_parameters(self.user_url)

        target_url, branch, ref = bzr_url_to_git_url(location)

        location = urlutils.relative_url(this_url, target_url)

        cs.set((b"remote", remote), b"url", location)