/brz/remove-bazaar

# Copyright (C) 2006-2010 Canonical Ltd

#

# 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

"""BzrDir logic. The BzrDir is the basic control directory used by bzr.

At format 7 this was split out into Branch, Repository and Checkout control

directories.

Note: This module has a lot of ``open`` functions/methods that return

references to in-memory objects. As a rule, there are no matching ``close``

methods. To free any associated resources, simply stop referencing the

objects returned.

"""

# TODO: Move old formats into a plugin to make this file smaller.

import os

import sys

import warnings

from bzrlib.lazy_import import lazy_import

lazy_import(globals(), """

from stat import S_ISDIR

import textwrap

import bzrlib

from bzrlib import (

    branch,

    config,

    errors,

    graph,

    lockable_files,

    lockdir,

    osutils,

    remote,

    repository,

    revision as _mod_revision,

    ui,

    urlutils,

    versionedfile,

    win32utils,

    workingtree,

    workingtree_4,

    xml4,

    xml5,

    )

from bzrlib.osutils import (

    sha_string,

    )

from bzrlib.push import (

    PushResult,

    )

from bzrlib.repofmt import pack_repo

from bzrlib.smart.client import _SmartClient

from bzrlib.store.versioned import WeaveStore

from bzrlib.transactions import WriteTransaction

from bzrlib.transport import (

    do_catching_redirections,

    get_transport,

    local,

    )

from bzrlib.weave import Weave

""")

from bzrlib.trace import (

    mutter,

    note,

    warning,

    )

from bzrlib import (

    hooks,

    registry,

    symbol_versioning,

    )

class BzrDir(object):

    """A .bzr control diretory.

    BzrDir instances let you create or open any of the things that can be

    found within .bzr - checkouts, branches and repositories.

    :ivar transport:

        the transport which this bzr dir is rooted at (i.e. file:///.../.bzr/)

    :ivar root_transport:

        a transport connected to the directory this bzr was opened from

        (i.e. the parent directory holding the .bzr directory).

    Everything in the bzrdir should have the same file permissions.

    :cvar hooks: An instance of BzrDirHooks.

    """

    def break_lock(self):

        """Invoke break_lock on the first object in the bzrdir.

        If there is a tree, the tree is opened and break_lock() called.

        Otherwise, branch is tried, and finally repository.

        """

        # XXX: This seems more like a UI function than something that really

        # belongs in this class.

        try:

            thing_to_unlock = self.open_workingtree()

        except (errors.NotLocalUrl, errors.NoWorkingTree):

            try:

                thing_to_unlock = self.open_branch()

            except errors.NotBranchError:

                try:

                    thing_to_unlock = self.open_repository()

                except errors.NoRepositoryPresent:

                    return

        thing_to_unlock.break_lock()

    def can_convert_format(self):

        """Return true if this bzrdir is one whose format we can convert from."""

        return True

    def check_conversion_target(self, target_format):

        """Check that a bzrdir as a whole can be converted to a new format."""

        # The only current restriction is that the repository content can be 

        # fetched compatibly with the target.

        target_repo_format = target_format.repository_format

        try:

            self.open_repository()._format.check_conversion_target(

                target_repo_format)

        except errors.NoRepositoryPresent:

            # No repo, no problem.

            pass

    @staticmethod

    def _check_supported(format, allow_unsupported,

        recommend_upgrade=True,

        basedir=None):

        """Give an error or warning on old formats.

        :param format: may be any kind of format - workingtree, branch,

        or repository.

        :param allow_unsupported: If true, allow opening

        formats that are strongly deprecated, and which may

        have limited functionality.

        :param recommend_upgrade: If true (default), warn

        the user through the ui object that they may wish

        to upgrade the object.

        """

        # TODO: perhaps move this into a base Format class; it's not BzrDir

        # specific. mbp 20070323

        if not allow_unsupported and not format.is_supported():

            # see open_downlevel to open legacy branches.

            raise errors.UnsupportedFormatError(format=format)

        if recommend_upgrade \

            and getattr(format, 'upgrade_recommended', False):

            ui.ui_factory.recommend_upgrade(

                format.get_format_description(),

                basedir)

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

              preserve_stacking=False):

        """Clone this bzrdir and its contents to url verbatim.

        :param url: The url create the clone at.  If url's last component does

            not exist, it will be created.

        :param revision_id: The tip revision-id to use for any branch or

            working tree.  If not None, then the clone operation may tune

            itself to download less data.

        :param force_new_repo: Do not use a shared repository for the target

                               even if one is available.

        :param preserve_stacking: When cloning a stacked branch, stack the

            new branch on top of the other branch's stacked-on branch.

        """

        return self.clone_on_transport(get_transport(url),

                                       revision_id=revision_id,

                                       force_new_repo=force_new_repo,

                                       preserve_stacking=preserve_stacking)

    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):

        """Clone this bzrdir and its contents to transport verbatim.

        :param transport: The transport for the location to produce the clone

            at.  If the target directory does not exist, it will be created.

        :param revision_id: The tip revision-id to use for any branch or

            working tree.  If not None, then the clone operation may tune

            itself to download less data.

        :param force_new_repo: Do not use a shared repository for the target,

                               even if one is available.

        :param preserve_stacking: When cloning a stacked branch, stack the

            new branch on top of the other branch's stacked-on branch.

        :param create_prefix: Create any missing directories leading up to

            to_transport.

        :param use_existing_dir: Use an existing directory if one exists.

        """

        # Overview: put together a broad description of what we want to end up

        # with; then make as few api calls as possible to do it.

        # We may want to create a repo/branch/tree, if we do so what format

        # would we want for each:

        require_stacking = (stacked_on is not None)

        format = self.cloning_metadir(require_stacking)

        # Figure out what objects we want:

        try:

            local_repo = self.find_repository()

        except errors.NoRepositoryPresent:

            local_repo = None

        try:

            local_branch = self.open_branch()

        except errors.NotBranchError:

            local_branch = None

        else:

            # enable fallbacks when branch is not a branch reference

            if local_branch.repository.has_same_location(local_repo):

                local_repo = local_branch.repository

            if preserve_stacking:

                try:

                    stacked_on = local_branch.get_stacked_on_url()

                except (errors.UnstackableBranchFormat,

                        errors.UnstackableRepositoryFormat,

                        errors.NotStacked):

                    pass

        # Bug: We create a metadir without knowing if it can support stacking,

        # we should look up the policy needs first, or just use it as a hint,

        # or something.

        if local_repo:

            make_working_trees = local_repo.make_working_trees()

            want_shared = local_repo.is_shared()

            repo_format_name = format.repository_format.network_name()

        else:

            make_working_trees = False

            want_shared = False

            repo_format_name = None

        result_repo, result, require_stacking, repository_policy = \

            format.initialize_on_transport_ex(transport,

            use_existing_dir=use_existing_dir, create_prefix=create_prefix,

            force_new_repo=force_new_repo, stacked_on=stacked_on,

            stack_on_pwd=self.root_transport.base,

            repo_format_name=repo_format_name,

            make_working_trees=make_working_trees, shared_repo=want_shared)

        if repo_format_name:

            try:

                # If the result repository is in the same place as the

                # resulting bzr dir, it will have no content, further if the

                # result is not stacked then we know all content should be

                # copied, and finally if we are copying up to a specific

                # revision_id then we can use the pending-ancestry-result which

                # does not require traversing all of history to describe it.

                if (result_repo.bzrdir.root_transport.base ==

                    result.root_transport.base and not require_stacking and

                    revision_id is not None):

                    fetch_spec = graph.PendingAncestryResult(

                        [revision_id], local_repo)

                    result_repo.fetch(local_repo, fetch_spec=fetch_spec)

                else:

                    result_repo.fetch(local_repo, revision_id=revision_id)

            finally:

                result_repo.unlock()

        else:

            if result_repo is not None:

                raise AssertionError('result_repo not None(%r)' % result_repo)

        # 1 if there is a branch present

        #   make sure its content is available in the target repository

        #   clone it.

        if local_branch is not None:

            result_branch = local_branch.clone(result, revision_id=revision_id,

                repository_policy=repository_policy)

        try:

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

            # the tree and fail.

            result.root_transport.local_abspath('.')

            if result_repo is None or result_repo.make_working_trees():

                self.open_workingtree().clone(result)

        except (errors.NoWorkingTree, errors.NotLocalUrl):

            pass

        return result

    # TODO: This should be given a Transport, and should chdir up; otherwise

    # this will open a new connection.

    def _make_tail(self, url):

        t = get_transport(url)

        t.ensure_base()

    @classmethod

    def create(cls, base, format=None, possible_transports=None):

        """Create a new BzrDir at the url 'base'.

        :param format: If supplied, the format of branch to create.  If not

            supplied, the default is used.

        :param possible_transports: If supplied, a list of transports that

            can be reused to share a remote connection.

        """

        if cls is not BzrDir:

            raise AssertionError("BzrDir.create always creates the default"

                " format, not one of %r" % cls)

        t = get_transport(base, possible_transports)

        t.ensure_base()

        if format is None:

            format = BzrDirFormat.get_default_format()

        return format.initialize_on_transport(t)

    @staticmethod

    def find_bzrdirs(transport, evaluate=None, list_current=None):

        """Find bzrdirs recursively from current location.

        This is intended primarily as a building block for more sophisticated

        functionality, like finding trees under a directory, or finding

        branches that use a given repository.

        :param evaluate: An optional callable that yields recurse, value,

            where recurse controls whether this bzrdir is recursed into

            and value is the value to yield.  By default, all bzrdirs

            are recursed into, and the return value is the bzrdir.

        :param list_current: if supplied, use this function to list the current

            directory, instead of Transport.list_dir

        :return: a generator of found bzrdirs, or whatever evaluate returns.

        """

        if list_current is None:

            def list_current(transport):

                return transport.list_dir('')

        if evaluate is None:

            def evaluate(bzrdir):

                return True, bzrdir

        pending = [transport]

        while len(pending) > 0:

            current_transport = pending.pop()

            recurse = True

            try:

                bzrdir = BzrDir.open_from_transport(current_transport)

            except errors.NotBranchError:

                pass

            else:

                recurse, value = evaluate(bzrdir)

                yield value

            try:

                subdirs = list_current(current_transport)

            except errors.NoSuchFile:

                continue

            if recurse:

                for subdir in sorted(subdirs, reverse=True):

                    pending.append(current_transport.clone(subdir))

    def list_branches(self):

        """Return a sequence of all branches local to this control directory.

        """

        try:

            return [self.open_branch()]

        except errors.NotBranchError:

            return []

    @staticmethod

    def find_branches(transport):

        """Find all branches under a transport.

        This will find all branches below the transport, including branches

        inside other branches.  Where possible, it will use

        Repository.find_branches.

        To list all the branches that use a particular Repository, see

        Repository.find_branches

        """

        def evaluate(bzrdir):

            try:

                repository = bzrdir.open_repository()

            except errors.NoRepositoryPresent:

                pass

            else:

                return False, ([], repository)

            return True, (bzrdir.list_branches(), None)

        ret = []

        for branches, repo in BzrDir.find_bzrdirs(transport,

                                                  evaluate=evaluate):

            if repo is not None:

                ret.extend(repo.find_branches())

            if branches is not None:

                ret.extend(branches)

        return ret

    def destroy_repository(self):

        """Destroy the repository in this BzrDir"""

        raise NotImplementedError(self.destroy_repository)

    def create_branch(self, name=None):

        """Create a branch in this BzrDir.

        :param name: Name of the colocated branch to create, None for

            the default branch.

        The bzrdir's format will control what branch format is created.

        For more control see BranchFormatXX.create(a_bzrdir).

        """

        raise NotImplementedError(self.create_branch)

    def destroy_branch(self, name=None):

        """Destroy a branch in this BzrDir.

        :param name: Name of the branch to destroy, None for the default 

            branch.

        """

        raise NotImplementedError(self.destroy_branch)

    @staticmethod

    def create_branch_and_repo(base, force_new_repo=False, format=None):

        """Create a new BzrDir, Branch and Repository at the url 'base'.

        This will use the current default BzrDirFormat unless one is

        specified, and use whatever

        repository format that that uses via bzrdir.create_branch and

        create_repository. If a shared repository is available that is used

        preferentially.

        The created Branch object is returned.

        :param base: The URL to create the branch at.

        :param force_new_repo: If True a new repository is always created.

        :param format: If supplied, the format of branch to create.  If not

            supplied, the default is used.

        """

        bzrdir = BzrDir.create(base, format)

        bzrdir._find_or_create_repository(force_new_repo)

        return bzrdir.create_branch()

    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.

        """

        def repository_policy(found_bzrdir):

            stack_on = None

            stack_on_pwd = None

            config = found_bzrdir.get_config()

            stop = False

            stack_on = config.get_default_stack_on()

            if stack_on is not None:

                stack_on_pwd = found_bzrdir.root_transport.base

                stop = True

            # does it have a repository ?

            try:

                repository = found_bzrdir.open_repository()

            except errors.NoRepositoryPresent:

                repository = None

            else:

                if ((found_bzrdir.root_transport.base !=

                     self.root_transport.base) and not repository.is_shared()):

                    # Don't look higher, can't use a higher shared repo.

                    repository = None

                    stop = True

                else:

                    stop = True

            if not stop:

                return None, False

            if repository:

                return UseExistingRepository(repository, stack_on,

                    stack_on_pwd, require_stacking=require_stacking), True

            else:

                return CreateRepository(self, stack_on, stack_on_pwd,

                    require_stacking=require_stacking), True

        if not force_new_repo:

            if stack_on is None:

                policy = self._find_containing(repository_policy)

                if policy is not None:

                    return policy

            else:

                try:

                    return UseExistingRepository(self.open_repository(),

                        stack_on, stack_on_pwd,

                        require_stacking=require_stacking)

                except errors.NoRepositoryPresent:

                    pass

        return CreateRepository(self, stack_on, stack_on_pwd,

                                require_stacking=require_stacking)

    def _find_or_create_repository(self, force_new_repo):

        """Create a new repository if needed, returning the repository."""

        policy = self.determine_repository_policy(force_new_repo)

        return policy.acquire_repository()[0]

    @staticmethod

    def create_branch_convenience(base, force_new_repo=False,

                                  force_new_tree=None, format=None,

                                  possible_transports=None):

        """Create a new BzrDir, Branch and Repository at the url 'base'.

        This is a convenience function - it will use an existing repository

        if possible, can be told explicitly whether to create a working tree or

        not.

        This will use the current default BzrDirFormat unless one is

        specified, and use whatever

        repository format that that uses via bzrdir.create_branch and

        create_repository. If a shared repository is available that is used

        preferentially. Whatever repository is used, its tree creation policy

        is followed.

        The created Branch object is returned.

        If a working tree cannot be made due to base not being a file:// url,

        no error is raised unless force_new_tree is True, in which case no

        data is created on disk and NotLocalUrl is raised.

        :param base: The URL to create the branch at.

        :param force_new_repo: If True a new repository is always created.

        :param force_new_tree: If True or False force creation of a tree or

                               prevent such creation respectively.

        :param format: Override for the bzrdir format to create.

        :param possible_transports: An optional reusable transports list.

        """

        if force_new_tree:

            # check for non local urls

            t = get_transport(base, possible_transports)

            if not isinstance(t, local.LocalTransport):

                raise errors.NotLocalUrl(base)

        bzrdir = BzrDir.create(base, format, possible_transports)

        repo = bzrdir._find_or_create_repository(force_new_repo)

        result = bzrdir.create_branch()

        if force_new_tree or (repo.make_working_trees() and

                              force_new_tree is None):

            try:

                bzrdir.create_workingtree()

            except errors.NotLocalUrl:

                pass

        return result

    @staticmethod

    def create_standalone_workingtree(base, format=None):

        """Create a new BzrDir, WorkingTree, Branch and Repository at 'base'.

        'base' must be a local path or a file:// url.

        This will use the current default BzrDirFormat unless one is

        specified, and use whatever

        repository format that that uses for bzrdirformat.create_workingtree,

        create_branch and create_repository.

        :param format: Override for the bzrdir format to create.

        :return: The WorkingTree object.

        """

        t = get_transport(base)

        if not isinstance(t, local.LocalTransport):

            raise errors.NotLocalUrl(base)

        bzrdir = BzrDir.create_branch_and_repo(base,

                                               force_new_repo=True,

                                               format=format).bzrdir

        return bzrdir.create_workingtree()

    def create_workingtree(self, revision_id=None, from_branch=None,

        accelerator_tree=None, hardlink=False):

        """Create a working tree at this BzrDir.

        :param revision_id: create it as of this revision id.

        :param from_branch: override bzrdir branch (for lightweight checkouts)

        :param accelerator_tree: A tree which can be used for retrieving file

            contents more quickly than the revision tree, i.e. a workingtree.

            The revision tree will be used for cases where accelerator_tree's

            content is different.

        """

        raise NotImplementedError(self.create_workingtree)

    def backup_bzrdir(self):

        """Backup this bzr control directory.

        :return: Tuple with old path name and new path name

        """

        def name_gen(base='backup.bzr'):

            counter = 1

            name = "%s.~%d~" % (base, counter)

            while self.root_transport.has(name):

                counter += 1

                name = "%s.~%d~" % (base, counter)

            return name

        backup_dir=name_gen()

        pb = ui.ui_factory.nested_progress_bar()

        try:

            # FIXME: bug 300001 -- the backup fails if the backup directory

            # already exists, but it should instead either remove it or make

            # a new backup directory.

            #

            # FIXME: bug 262450 -- the backup directory should have the same

            # permissions as the .bzr directory (probably a bug in copy_tree)

            old_path = self.root_transport.abspath('.bzr')

            new_path = self.root_transport.abspath(backup_dir)

            ui.ui_factory.note('making backup of %s\n  to %s' % (old_path, new_path,))

            self.root_transport.copy_tree('.bzr', backup_dir)

            return (old_path, new_path)

        finally:

            pb.finished()

    def retire_bzrdir(self, limit=10000):

        """Permanently disable the bzrdir.

        This is done by renaming it to give the user some ability to recover

        if there was a problem.

        This will have horrible consequences if anyone has anything locked or

        in use.

        :param limit: number of times to retry

        """

        i  = 0

        while True:

            try:

                to_path = '.bzr.retired.%d' % i

                self.root_transport.rename('.bzr', to_path)

                note("renamed %s to %s"

                    % (self.root_transport.abspath('.bzr'), to_path))

                return

            except (errors.TransportError, IOError, errors.PathError):

                i += 1

                if i > limit:

                    raise

                else:

                    pass

    def destroy_workingtree(self):

        """Destroy the working tree at this BzrDir.

        Formats that do not support this may raise UnsupportedOperation.

        """

        raise NotImplementedError(self.destroy_workingtree)

    def destroy_workingtree_metadata(self):

        """Destroy the control files for the working tree at this BzrDir.

        The contents of working tree files are not affected.

        Formats that do not support this may raise UnsupportedOperation.

        """

        raise NotImplementedError(self.destroy_workingtree_metadata)

    def _find_containing(self, evaluate):

        """Find something in a containing control directory.

        This method will scan containing control dirs, until it finds what

        it is looking for, decides that it will never find it, or runs out

        of containing control directories to check.

        It is used to implement find_repository and

        determine_repository_policy.

        :param evaluate: A function returning (value, stop).  If stop is True,

            the value will be returned.

        """

        found_bzrdir = self

        while True:

            result, stop = evaluate(found_bzrdir)

            if stop:

                return result

            next_transport = found_bzrdir.root_transport.clone('..')

            if (found_bzrdir.root_transport.base == next_transport.base):

                # top of the file system

                return None

            # find the next containing bzrdir

            try:

                found_bzrdir = BzrDir.open_containing_from_transport(

                    next_transport)[0]

            except errors.NotBranchError:

                return None

    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.

        """

        def usable_repository(found_bzrdir):

            # does it have a repository ?

            try:

                repository = found_bzrdir.open_repository()

            except errors.NoRepositoryPresent:

                return None, False

            if found_bzrdir.root_transport.base == self.root_transport.base:

                return repository, True

            elif repository.is_shared():

                return repository, True

            else:

                return None, True

        found_repo = self._find_containing(usable_repository)

        if found_repo is None:

            raise errors.NoRepositoryPresent(self)

        return found_repo