/brz/remove-bazaar

# Copyright (C) 2006-2011 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.

"""

import sys

from bzrlib.lazy_import import lazy_import

lazy_import(globals(), """

import bzrlib

from bzrlib import (

    branch as _mod_branch,

    cleanup,

    errors,

    fetch,

    graph,

    lockable_files,

    lockdir,

    osutils,

    pyutils,

    remote,

    repository,

    revision as _mod_revision,

    transport as _mod_transport,

    ui,

    urlutils,

    win32utils,

    workingtree_3,

    workingtree_4,

    )

from bzrlib.repofmt import knitpack_repo

from bzrlib.transport import (

    do_catching_redirections,

    local,

    )

from bzrlib.i18n import gettext

""")

from bzrlib.trace import (

    mutter,

    note,

    )

from bzrlib import (

    config,

    controldir,

    registry,

    )

from bzrlib.symbol_versioning import (

    deprecated_in,

    deprecated_method,

    )

class BzrDir(controldir.ControlDir):

    """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 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

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

        """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.

        :param no_tree: If set to true prevents creation of a working tree.

        """

        # 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() and not no_tree

            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.user_url == result.user_url

                    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, revision_id=revision_id)

        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 = _mod_transport.get_transport(url)

        t.ensure_base()

    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.user_url

                stop = True

            # does it have a repository ?

            try:

                repository = found_bzrdir.open_repository()

            except errors.NoRepositoryPresent:

                repository = None

            else:

                if (found_bzrdir.user_url != self.user_url 

                    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]

    def _find_source_repo(self, add_cleanup, source_branch):

        """Find the source branch and repo for a sprout operation.

        This is helper intended for use by _sprout.

        :returns: (source_branch, source_repository).  Either or both may be

            None.  If not None, they will be read-locked (and their unlock(s)

            scheduled via the add_cleanup param).

        """

        if source_branch is not None:

            add_cleanup(source_branch.lock_read().unlock)

            return source_branch, source_branch.repository

        try:

            source_branch = self.open_branch()

            source_repository = source_branch.repository

        except errors.NotBranchError:

            source_branch = None

            try:

                source_repository = self.open_repository()

            except errors.NoRepositoryPresent:

                source_repository = None

            else:

                add_cleanup(source_repository.lock_read().unlock)

        else:

            add_cleanup(source_branch.lock_read().unlock)

        return source_branch, source_repository

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

        """Create a copy of this controldir prepared for use as a new line of

        development.

        If url's last component does not exist, it will be created.

        Attributes related to the identity of the source branch like

        branch nickname will be cleaned, a working tree is created

        whether one existed before or not; and a local branch is always

        created.

        if revision_id is not None, then the clone operation may tune

            itself to download less data.

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

        :param hardlink: If true, hard-link files from accelerator_tree,

            where possible.

        :param stacked: If true, create a stacked branch referring to the

            location of this control directory.

        :param create_tree_if_local: If true, a working-tree will be created

            when working locally.

        """

        operation = cleanup.OperationWithCleanups(self._sprout)

        return operation.run(url, revision_id=revision_id,

            force_new_repo=force_new_repo, recurse=recurse,

            possible_transports=possible_transports,

            accelerator_tree=accelerator_tree, hardlink=hardlink,

            stacked=stacked, source_branch=source_branch,

            create_tree_if_local=create_tree_if_local)

    def _sprout(self, op, 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):

        add_cleanup = op.add_cleanup

        fetch_spec_factory = fetch.FetchSpecFactory()

        if revision_id is not None:

            fetch_spec_factory.add_revision_ids([revision_id])

            fetch_spec_factory.source_branch_stop_revision_id = revision_id

        target_transport = _mod_transport.get_transport(url,

            possible_transports)

        target_transport.ensure_base()

        cloning_format = self.cloning_metadir(stacked)

        # Create/update the result branch

        result = cloning_format.initialize_on_transport(target_transport)

        source_branch, source_repository = self._find_source_repo(

            add_cleanup, source_branch)

        fetch_spec_factory.source_branch = source_branch

        # if a stacked branch wasn't requested, we don't create one

        # even if the origin was stacked

        if stacked and source_branch is not None:

            stacked_branch_url = self.root_transport.base

        else:

            stacked_branch_url = None

        repository_policy = result.determine_repository_policy(

            force_new_repo, stacked_branch_url, require_stacking=stacked)

        result_repo, is_new_repo = repository_policy.acquire_repository()

        add_cleanup(result_repo.lock_write().unlock)

        fetch_spec_factory.source_repo = source_repository

        fetch_spec_factory.target_repo = result_repo

        if stacked or (len(result_repo._fallback_repositories) != 0):

            target_repo_kind = fetch.TargetRepoKinds.STACKED

        elif is_new_repo:

            target_repo_kind = fetch.TargetRepoKinds.EMPTY

        else:

            target_repo_kind = fetch.TargetRepoKinds.PREEXISTING

        fetch_spec_factory.target_repo_kind = target_repo_kind

        if source_repository is not None:

            fetch_spec = fetch_spec_factory.make_fetch_spec()

            result_repo.fetch(source_repository, fetch_spec=fetch_spec)

        if source_branch is None:

            # this is for sprouting a controldir without a branch; is that

            # actually useful?

            # Not especially, but it's part of the contract.

            result_branch = result.create_branch()

        else:

            result_branch = source_branch.sprout(result,

                revision_id=revision_id, repository_policy=repository_policy,

                repository=result_repo)

        mutter("created new branch %r" % (result_branch,))

        # Create/update the result working tree

        if (create_tree_if_local and

            isinstance(target_transport, local.LocalTransport) and

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

            wt = result.create_workingtree(accelerator_tree=accelerator_tree,

                hardlink=hardlink, from_branch=result_branch)

            wt.lock_write()

            try:

                if wt.path2id('') is None:

                    try:

                        wt.set_root_id(self.open_workingtree.get_root_id())

                    except errors.NoWorkingTree:

                        pass

            finally:

                wt.unlock()

        else:

            wt = None

        if recurse == 'down':

            basis = None

            if wt is not None:

                basis = wt.basis_tree()

            elif result_branch is not None:

                basis = result_branch.basis_tree()

            elif source_branch is not None:

                basis = source_branch.basis_tree()

            if basis is not None:

                add_cleanup(basis.lock_read().unlock)

                subtrees = basis.iter_references()

            else:

                subtrees = []

            for path, file_id in subtrees:

                target = urlutils.join(url, urlutils.escape(path))

                sublocation = source_branch.reference_parent(file_id, path)

                sublocation.bzrdir.sprout(target,

                    basis.get_reference_revision(file_id, path),

                    force_new_repo=force_new_repo, recurse=recurse,

                    stacked=stacked)

        return result

    @deprecated_method(deprecated_in((2, 3, 0)))

    def generate_backup_name(self, base):

        return self._available_backup_name(base)

    def _available_backup_name(self, base):

        """Find a non-existing backup file name based on base.

        See bzrlib.osutils.available_backup_name about race conditions.

        """

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

    def backup_bzrdir(self):

        """Backup this bzr control directory.

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

        """

        pb = ui.ui_factory.nested_progress_bar()

        try:

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

            backup_dir = self._available_backup_name('backup.bzr')

            new_path = self.root_transport.abspath(backup_dir)

            ui.ui_factory.note(gettext('making backup of {0}\n  to {1}').format(

                urlutils.unescape_for_display(old_path, 'utf-8'),

                urlutils.unescape_for_display(new_path, 'utf-8')))

            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(gettext("renamed {0} to {1}").format(

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

                return

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

                i += 1

                if i > limit:

                    raise

                else:

                    pass

    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.user_url == next_transport.base):

                # top of the file system

                return None

            # find the next containing bzrdir

            try:

                found_bzrdir = self.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.user_url == self.user_url:

                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

    def _find_creation_modes(self):

        """Determine the appropriate modes for files and directories.

        They're always set to be consistent with the base directory,

        assuming that this transport allows setting modes.

        """

        # TODO: Do we need or want an option (maybe a config setting) to turn

        # this off or override it for particular locations? -- mbp 20080512

        if self._mode_check_done:

            return

        self._mode_check_done = True

        try:

            st = self.transport.stat('.')

        except errors.TransportNotPossible:

            self._dir_mode = None

            self._file_mode = None

        else:

            # Check the directory mode, but also make sure the created

            # directories and files are read-write for this user. This is

            # mostly a workaround for filesystems which lie about being able to

            # write to a directory (cygwin & win32)

            if (st.st_mode & 07777 == 00000):

                # FTP allows stat but does not return dir/file modes

                self._dir_mode = None

                self._file_mode = None

            else:

                self._dir_mode = (st.st_mode & 07777) | 00700

                # Remove the sticky and execute bits for files

                self._file_mode = self._dir_mode & ~07111

    def _get_file_mode(self):

        """Return Unix mode for newly created files, or None.

        """

        if not self._mode_check_done:

            self._find_creation_modes()

        return self._file_mode

    def _get_dir_mode(self):

        """Return Unix mode for newly created directories, or None.

        """

        if not self._mode_check_done:

            self._find_creation_modes()

        return self._dir_mode

    def get_config(self):

        """Get configuration for this BzrDir."""

        return config.BzrDirConfig(self)

    def _get_config(self):

        """By default, no configuration is available."""

        return None

    def __init__(self, _transport, _format):

        """Initialize a Bzr control dir object.

        Only really common logic should reside here, concrete classes should be

        made with varying behaviours.

        :param _format: the format that is creating this BzrDir instance.

        :param _transport: the transport this dir is based at.

        """

        self._format = _format

        # these are also under the more standard names of 

        # control_transport and user_transport

        self.transport = _transport.clone('.bzr')

        self.root_transport = _transport

        self._mode_check_done = False

    @property 

    def user_transport(self):

        return self.root_transport

    @property

    def control_transport(self):

        return self.transport

    def is_control_filename(self, filename):

        """True if filename is the name of a path which is reserved for bzrdir's.

        :param filename: A filename within the root transport of this bzrdir.

        This is true IF and ONLY IF the filename is part of the namespace reserved

        for bzr control dirs. Currently this is the '.bzr' directory in the root

        of the root_transport. 

        """

        # this might be better on the BzrDirFormat class because it refers to

        # all the possible bzrdir disk formats.

        # This method is tested via the workingtree is_control_filename tests-

        # it was extracted from WorkingTree.is_control_filename. If the method's

        # contract is extended beyond the current trivial implementation, please

        # add new tests for it to the appropriate place.

        return filename == '.bzr' or filename.startswith('.bzr/')

    def _cloning_metadir(self):

        """Produce a metadir suitable for cloning with.

        :returns: (destination_bzrdir_format, source_repository)

        """

        result_format = self._format.__class__()

        try:

            try:

                branch = self.open_branch(ignore_fallbacks=True)

                source_repository = branch.repository

                result_format._branch_format = branch._format

            except errors.NotBranchError:

                source_branch = None

                source_repository = self.open_repository()

        except errors.NoRepositoryPresent:

            source_repository = None

        else:

            # XXX TODO: This isinstance is here because we have not implemented

            # the fix recommended in bug # 103195 - to delegate this choice the

            # repository itself.

            repo_format = source_repository._format

            if isinstance(repo_format, remote.RemoteRepositoryFormat):

                source_repository._ensure_real()

                repo_format = source_repository._real_repository._format

            result_format.repository_format = repo_format

        try:

            # TODO: Couldn't we just probe for the format in these cases,

            # rather than opening the whole tree?  It would be a little

            # faster. mbp 20070401

            tree = self.open_workingtree(recommend_upgrade=False)

        except (errors.NoWorkingTree, errors.NotLocalUrl):

            result_format.workingtree_format = None

        else:

            result_format.workingtree_format = tree._format.__class__()

        return result_format, source_repository

    def cloning_metadir(self, require_stacking=False):

        """Produce a metadir suitable for cloning or sprouting with.

        These operations may produce workingtrees (yes, even though they're

        "cloning" something that doesn't have a tree), so a viable workingtree

        format must be selected.