/brz/remove-bazaar

# Copyright (C) 2007 Canonical Ltd

# Copyright (C) 2010-2018 Jelmer Vernooij

#

# 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 control dir and a Bazaar ControlDir."""

from __future__ import absolute_import

from .. import (

    branch as _mod_branch,

    errors as brz_errors,

    trace,

    osutils,

    urlutils,

    )

from ..sixish import (

    PY3,

    viewitems,

    )

from ..transport import (

    do_catching_redirections,

    get_transport_from_path,

    )

from ..controldir import (

    BranchReferenceLoop,

    ControlDir,

    ControlDirFormat,

    format_registry,

    RepositoryAcquisitionPolicy,

    )

from .push import (

    GitPushResult,

    )

from .transportgit import (

    OBJECTDIR,

    TransportObjectStore,

    )

class GitDirConfig(object):

    def get_default_stack_on(self):

        return None

    def set_default_stack_on(self, value):

        raise brz_errors.BzrError("Cannot set configuration")

class GitControlDirFormat(ControlDirFormat):

    colocated_branches = True

    fixed_components = True

    def __eq__(self, other):

        return type(self) == type(other)

    def is_supported(self):

        return True

    def network_name(self):

        return b"git"

class UseExistingRepository(RepositoryAcquisitionPolicy):

    """A policy of reusing an existing repository"""

    def __init__(self, repository, stack_on=None, stack_on_pwd=None,

                 require_stacking=False):

        """Constructor.

        :param repository: The repository to use.

        :param stack_on: A location to stack on

        :param stack_on_pwd: If stack_on is relative, the location it is

            relative to.

        """

        super(UseExistingRepository, self).__init__(

            stack_on, stack_on_pwd, require_stacking)

        self._repository = repository

    def acquire_repository(self, make_working_trees=None, shared=False,

                           possible_transports=None):

        """Implementation of RepositoryAcquisitionPolicy.acquire_repository

        Returns an existing repository to use.

        """

        return self._repository, False

class GitDir(ControlDir):

    """An adapter to the '.git' dir used by git."""

    def is_supported(self):

        return True

    def can_convert_format(self):

        return False

    def break_lock(self):

        # There are no global locks, so nothing to break.

        raise NotImplementedError(self.break_lock)

    def cloning_metadir(self, stacked=False):

        return format_registry.make_controldir("git")

    def checkout_metadir(self, stacked=False):

        return format_registry.make_controldir("git")

    def _get_selected_ref(self, branch, ref=None):

        if ref is not None and branch is not None:

            raise brz_errors.BzrError("can't specify both ref and branch")

        if ref is not None:

            return ref

        if branch is not None:

            from .refs import branch_name_to_ref

            return branch_name_to_ref(branch)

        segment_parameters = getattr(

            self.user_transport, "get_segment_parameters", lambda: {})()

        ref = segment_parameters.get("ref")

        if ref is not None:

            return urlutils.unquote_to_bytes(ref)

        if branch is None and getattr(self, "_get_selected_branch", False):

            branch = self._get_selected_branch()

            if branch is not None:

                from .refs import branch_name_to_ref

                return branch_name_to_ref(branch)

        return b"HEAD"

    def get_config(self):

        return GitDirConfig()

    def _available_backup_name(self, base):

        return osutils.available_backup_name(base, self.root_transport.has)

    def sprout(self, url, revision_id=None, force_new_repo=False,

               recurse='down', possible_transports=None,

               accelerator_tree=None, hardlink=False, stacked=False,

               source_branch=None, create_tree_if_local=True):

        from ..repository import InterRepository

        from ..transport.local import LocalTransport

        from ..transport import get_transport

        target_transport = get_transport(url, possible_transports)

        target_transport.ensure_base()

        cloning_format = self.cloning_metadir()

        # Create/update the result branch

        try:

            result = ControlDir.open_from_transport(target_transport)

        except brz_errors.NotBranchError:

            result = cloning_format.initialize_on_transport(target_transport)

        source_branch = self.open_branch()

        source_repository = self.find_repository()

        try:

            result_repo = result.find_repository()

        except brz_errors.NoRepositoryPresent:

            result_repo = result.create_repository()

        if stacked:

            raise _mod_branch.UnstackableBranchFormat(

                self._format, self.user_url)

        interrepo = InterRepository.get(source_repository, result_repo)

        if revision_id is not None:

            determine_wants = interrepo.get_determine_wants_revids(

                [revision_id], include_tags=True)

        else:

            determine_wants = interrepo.determine_wants_all

        interrepo.fetch_objects(determine_wants=determine_wants,

                                mapping=source_branch.mapping)

        result_branch = source_branch.sprout(

            result, revision_id=revision_id, repository=result_repo)

        if (create_tree_if_local and

            result.open_branch(name="").name == result_branch.name and

            isinstance(target_transport, LocalTransport) and

                (result_repo is None or result_repo.make_working_trees())):

            result.create_workingtree(

                accelerator_tree=accelerator_tree,

                hardlink=hardlink, from_branch=result_branch)

        return result

    def clone_on_transport(self, transport, revision_id=None,

                           force_new_repo=False, preserve_stacking=False,

                           stacked_on=None, create_prefix=False,

                           use_existing_dir=True, no_tree=False):

        """See ControlDir.clone_on_transport."""

        from ..repository import InterRepository

        from .mapping import default_mapping

        if stacked_on is not None:

            raise _mod_branch.UnstackableBranchFormat(

                self._format, self.user_url)

        if no_tree:

            format = BareLocalGitControlDirFormat()

        else:

            format = LocalGitControlDirFormat()

        (target_repo, target_controldir, stacking,

         repo_policy) = format.initialize_on_transport_ex(

            transport, use_existing_dir=use_existing_dir,

            create_prefix=create_prefix,

            force_new_repo=force_new_repo)

        target_repo = target_controldir.find_repository()

        target_git_repo = target_repo._git

        source_repo = self.find_repository()

        interrepo = InterRepository.get(source_repo, target_repo)

        if revision_id is not None:

            determine_wants = interrepo.get_determine_wants_revids(

                [revision_id], include_tags=True)

        else:

            determine_wants = interrepo.determine_wants_all

        (pack_hint, _, refs) = interrepo.fetch_objects(determine_wants,

                                                       mapping=default_mapping)

        for name, val in viewitems(refs):

            target_git_repo.refs[name] = val

        result_dir = self.__class__(transport, target_git_repo, format)

        if revision_id is not None:

            result_dir.open_branch().set_last_revision(revision_id)

        try:

            # Cheaper to check if the target is not local, than to try making

            # the tree and fail.

            result_dir.root_transport.local_abspath('.')

            if result_dir.open_repository().make_working_trees():

                self.open_workingtree().clone(

                    result_dir, revision_id=revision_id)

        except (brz_errors.NoWorkingTree, brz_errors.NotLocalUrl):

            pass

        return result_dir

    def find_repository(self):

        """Find the repository that should be used.

        This does not require a branch as we use it to find the repo for

        new branches as well as to hook existing branches up to their

        repository.

        """

        return self._gitrepository_class(self._find_commondir())

    def get_refs_container(self):

        """Retrieve the refs container.

        """

        raise NotImplementedError(self.get_refs_container)

    def determine_repository_policy(self, force_new_repo=False, stack_on=None,

                                    stack_on_pwd=None, require_stacking=False):

        """Return an object representing a policy to use.

        This controls whether a new repository is created, and the format of

        that repository, or some existing shared repository used instead.

        If stack_on is supplied, will not seek a containing shared repo.

        :param force_new_repo: If True, require a new repository to be created.

        :param stack_on: If supplied, the location to stack on.  If not

            supplied, a default_stack_on location may be used.

        :param stack_on_pwd: If stack_on is relative, the location it is

            relative to.

        """

        return UseExistingRepository(self.find_repository())

    def get_branches(self):

        from .refs import ref_to_branch_name

        ret = {}

        for ref in self.get_refs_container().keys():

            try:

                branch_name = ref_to_branch_name(ref)

            except UnicodeDecodeError:

                trace.warning("Ignoring branch %r with unicode error ref", ref)

                continue

            except ValueError:

                continue

            ret[branch_name] = self.open_branch(ref=ref)

        return ret

    def list_branches(self):

        return list(self.get_branches().values())

    def push_branch(self, source, revision_id=None, overwrite=False,

                    remember=False, create_prefix=False, lossy=False,

                    name=None):

        """Push the source branch into this ControlDir."""

        push_result = GitPushResult()

        push_result.workingtree_updated = None

        push_result.master_branch = None

        push_result.source_branch = source

        push_result.stacked_on = None

        from .branch import GitBranch

        if isinstance(source, GitBranch) and lossy:

            raise brz_errors.LossyPushToSameVCS(source.controldir, self)

        target = self.open_branch(name, nascent_ok=True)

        push_result.branch_push_result = source.push(

            target, overwrite=overwrite, stop_revision=revision_id,

            lossy=lossy)

        push_result.new_revid = push_result.branch_push_result.new_revid

        push_result.old_revid = push_result.branch_push_result.old_revid

        try:

            wt = self.open_workingtree()

        except brz_errors.NoWorkingTree:

            push_result.workingtree_updated = None

        else:

            if self.open_branch(name="").name == target.name:

                wt._update_git_tree(

                    old_revision=push_result.old_revid,

                    new_revision=push_result.new_revid)

                push_result.workingtree_updated = True

            else:

                push_result.workingtree_updated = False

        push_result.target_branch = target

        if source.get_push_location() is None or remember:

            source.set_push_location(push_result.target_branch.base)

        return push_result

class LocalGitControlDirFormat(GitControlDirFormat):

    """The .git directory control format."""

    bare = False

    @classmethod

    def _known_formats(self):

        return set([LocalGitControlDirFormat()])

    @property

    def repository_format(self):

        from .repository import GitRepositoryFormat

        return GitRepositoryFormat()

    @property

    def workingtree_format(self):

        from .workingtree import GitWorkingTreeFormat

        return GitWorkingTreeFormat()

    def get_branch_format(self):

        from .branch import LocalGitBranchFormat

        return LocalGitBranchFormat()

    def open(self, transport, _found=None):

        """Open this directory.

        """

        from .transportgit import TransportRepo

        def _open(transport):

            try:

                return TransportRepo(transport, self.bare,

                                     refs_text=getattr(self, "_refs_text", None))

            except ValueError as e:

                if e.args == ('Expected file to start with \'gitdir: \'', ):

                    raise brz_errors.NotBranchError(path=transport.base)

                raise

        def redirected(transport, e, redirection_notice):

            trace.note(redirection_notice)

            return transport._redirected_to(e.source, e.target)

        gitrepo = do_catching_redirections(_open, transport, redirected)

        if not gitrepo._controltransport.has('HEAD'):

            raise brz_errors.NotBranchError(path=transport.base)

        return LocalGitDir(transport, gitrepo, self)

    def get_format_description(self):

        return "Local Git Repository"

    def initialize_on_transport(self, transport):

        from .transportgit import TransportRepo

        git_repo = TransportRepo.init(transport, bare=self.bare)

        return LocalGitDir(transport, git_repo, self)

    def initialize_on_transport_ex(self, transport, use_existing_dir=False,

                                   create_prefix=False, force_new_repo=False,

                                   stacked_on=None,

                                   stack_on_pwd=None, repo_format_name=None,

                                   make_working_trees=None,

                                   shared_repo=False, vfs_only=False):

        def make_directory(transport):

            transport.mkdir('.')

            return transport

        def redirected(transport, e, redirection_notice):

            trace.note(redirection_notice)

            return transport._redirected_to(e.source, e.target)

        try:

            transport = do_catching_redirections(

                make_directory, transport, redirected)

        except brz_errors.FileExists:

            if not use_existing_dir:

                raise

        except brz_errors.NoSuchFile:

            if not create_prefix:

                raise

            transport.create_prefix()

        controldir = self.initialize_on_transport(transport)

        if repo_format_name:

            result_repo = controldir.find_repository()

            repository_policy = UseExistingRepository(result_repo)

            result_repo.lock_write()

        else:

            result_repo = None

            repository_policy = None

        return (result_repo, controldir, False,

                repository_policy)

    def is_supported(self):

        return True

    def supports_transport(self, transport):

        try:

            external_url = transport.external_url()

        except brz_errors.InProcessTransport:

            raise brz_errors.NotBranchError(path=transport.base)

        return external_url.startswith("file:")

    def is_control_filename(self, filename):

        return (filename == '.git'

                or filename.startswith('.git/')

                or filename.startswith('.git\\'))

class BareLocalGitControlDirFormat(LocalGitControlDirFormat):

    bare = True

    supports_workingtrees = False

    def get_format_description(self):

        return "Local Git Repository (bare)"

    def is_control_filename(self, filename):

        return False

class LocalGitDir(GitDir):

    """An adapter to the '.git' dir used by git."""

    def _get_gitrepository_class(self):

        from .repository import LocalGitRepository

        return LocalGitRepository

    def __repr__(self):

        return "<%s at %r>" % (

            self.__class__.__name__, self.root_transport.base)

    _gitrepository_class = property(_get_gitrepository_class)

    @property

    def user_transport(self):

        return self.root_transport

    @property

    def control_transport(self):

        return self._git._controltransport

    def __init__(self, transport, gitrepo, format):

        self._format = format

        self.root_transport = transport

        self._mode_check_done = False

        self._git = gitrepo

        if gitrepo.bare:

            self.transport = transport

        else:

            self.transport = transport.clone('.git')

        self._mode_check_done = None

    def _get_symref(self, ref):

        ref_chain, unused_sha = self._git.refs.follow(ref)

        if len(ref_chain) == 1:

            return None

        return ref_chain[1]

    def set_branch_reference(self, target_branch, name=None):

        ref = self._get_selected_ref(name)

        target_transport = target_branch.controldir.control_transport

        if self.control_transport.base == target_transport.base:

            if ref == target_branch.ref:

                raise BranchReferenceLoop(target_branch)

            self._git.refs.set_symbolic_ref(ref, target_branch.ref)

        else:

            try:

                target_path = (

                    target_branch.controldir.control_transport.local_abspath(

                        '.'))

            except brz_errors.NotLocalUrl:

                raise brz_errors.IncompatibleFormat(

                    target_branch._format, self._format)

            # TODO(jelmer): Do some consistency checking across branches..

            self.control_transport.put_bytes(

                'commondir', target_path.encode('utf-8'))

            # TODO(jelmer): Urgh, avoid mucking about with internals.

            self._git._commontransport = (

                target_branch.repository._git._commontransport.clone())

            self._git.object_store = TransportObjectStore(

                self._git._commontransport.clone(OBJECTDIR))

            self._git.refs.transport = self._git._commontransport

            target_ref_chain, unused_sha = (

                target_branch.controldir._git.refs.follow(target_branch.ref))

            for target_ref in target_ref_chain:

                if target_ref == b'HEAD':

                    continue

                break

            else:

                # Can't create a reference to something that is not a in a repository.

                raise brz_errors.IncompatibleFormat(

                    self.set_branch_reference, self)

            self._git.refs.set_symbolic_ref(ref, target_ref)

    def get_branch_reference(self, name=None):

        ref = self._get_selected_ref(name)

        target_ref = self._get_symref(ref)

        if target_ref is not None:

            from .refs import ref_to_branch_name

            try:

                branch_name = ref_to_branch_name(target_ref)

            except ValueError:

                params = {'ref': urlutils.quote(

                    target_ref.decode('utf-8'), '')}

            else:

                if branch_name != '':

                    params = {'branch': urlutils.quote(branch_name, '')}

                else:

                    params = {}

            try:

                commondir = self.control_transport.get_bytes('commondir')

            except brz_errors.NoSuchFile:

                base_url = self.user_url.rstrip('/')

            else:

                base_url = urlutils.local_path_to_url(

                    commondir.decode(osutils._fs_enc)).rstrip('/.git/') + '/'

            if not PY3:

                params = {k: v.encode('utf-8') for (k, v) in viewitems(params)}

            return urlutils.join_segment_parameters(base_url, params)

        return None

    def find_branch_format(self, name=None):

        from .branch import (

            LocalGitBranchFormat,

            )

        return LocalGitBranchFormat()

    def get_branch_transport(self, branch_format, name=None):

        if branch_format is None:

            return self.transport

        if isinstance(branch_format, LocalGitControlDirFormat):

            return self.transport

        raise brz_errors.IncompatibleFormat(branch_format, self._format)

    def get_repository_transport(self, format):

        if format is None:

            return self.transport

        if isinstance(format, LocalGitControlDirFormat):

            return self.transport

        raise brz_errors.IncompatibleFormat(format, self._format)

    def get_workingtree_transport(self, format):

        if format is None:

            return self.transport

        if isinstance(format, LocalGitControlDirFormat):

            return self.transport

        raise brz_errors.IncompatibleFormat(format, self._format)

    def open_branch(self, name=None, unsupported=False, ignore_fallbacks=None,

                    ref=None, possible_transports=None, nascent_ok=False):

        """'create' a branch for this dir."""

        repo = self.find_repository()

        from .branch import LocalGitBranch

        ref = self._get_selected_ref(name, ref)

        if not nascent_ok and ref not in self._git.refs:

            raise brz_errors.NotBranchError(

                self.root_transport.base, controldir=self)

        ref_chain, unused_sha = self._git.refs.follow(ref)

        if ref_chain[-1] == b'HEAD':

            controldir = self

        else:

            controldir = self._find_commondir()

        return LocalGitBranch(controldir, repo, ref_chain[-1])

    def destroy_branch(self, name=None):

        refname = self._get_selected_ref(name)

        if refname == b'HEAD':

            # HEAD can't be removed

            raise brz_errors.UnsupportedOperation(

                self.destroy_branch, self)

        try:

            del self._git.refs[refname]

        except KeyError:

            raise brz_errors.NotBranchError(

                self.root_transport.base, controldir=self)

    def destroy_repository(self):

        raise brz_errors.UnsupportedOperation(self.destroy_repository, self)

    def destroy_workingtree(self):

        raise brz_errors.UnsupportedOperation(self.destroy_workingtree, self)

    def destroy_workingtree_metadata(self):

        raise brz_errors.UnsupportedOperation(

            self.destroy_workingtree_metadata, self)

    def needs_format_conversion(self, format=None):

        return not isinstance(self._format, format.__class__)

    def open_repository(self):

        """'open' a repository for this dir."""

        if self.control_transport.has('commondir'):