/brz/remove-bazaar

# Copyright (C) 2005, 2006 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., 59 Temple Place, Suite 330, Boston, MA  02111-1307  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.

"""

# TODO: remove unittest dependency; put that stuff inside the test suite

from copy import deepcopy

from cStringIO import StringIO

import os

from stat import S_ISDIR

from unittest import TestSuite

import bzrlib

import bzrlib.errors as errors

from bzrlib.lockable_files import LockableFiles, TransportLock

from bzrlib.lockdir import LockDir

from bzrlib.osutils import (

                            abspath,

                            pathjoin,

                            safe_unicode,

                            sha_strings,

                            sha_string,

                            )

from bzrlib.store.revision.text import TextRevisionStore

from bzrlib.store.text import TextStore

from bzrlib.store.versioned import WeaveStore

from bzrlib.trace import mutter

from bzrlib.transactions import WriteTransaction

from bzrlib.transport import get_transport

from bzrlib.transport.local import LocalTransport

import bzrlib.urlutils as urlutils

from bzrlib.weave import Weave

from bzrlib.xml4 import serializer_v4

import bzrlib.xml5

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.

    transport

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

    root_transport

        a transport connected to the directory this bzr was opened from.

    """

    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.

        """

        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

    @staticmethod

    def _check_supported(format, allow_unsupported):

        """Check whether format is a supported format.

        If allow_unsupported is True, this is a no-op.

        """

        if not allow_unsupported and not format.is_supported():

            # see open_downlevel to open legacy branches.

            raise errors.UnsupportedFormatError(format=format)

    def clone(self, url, revision_id=None, basis=None, force_new_repo=False):

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

        If urls last component does not exist, it will be created.

        if revision_id is 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.

        """

        self._make_tail(url)

        basis_repo, basis_branch, basis_tree = self._get_basis_components(basis)

        result = self._format.initialize(url)

        try:

            local_repo = self.find_repository()

        except errors.NoRepositoryPresent:

            local_repo = None

        if local_repo:

            # may need to copy content in

            if force_new_repo:

                result_repo = local_repo.clone(

                    result,

                    revision_id=revision_id,

                    basis=basis_repo)

                result_repo.set_make_working_trees(local_repo.make_working_trees())

            else:

                try:

                    result_repo = result.find_repository()

                    # fetch content this dir needs.

                    if basis_repo:

                        # XXX FIXME RBC 20060214 need tests for this when the basis

                        # is incomplete

                        result_repo.fetch(basis_repo, revision_id=revision_id)

                    result_repo.fetch(local_repo, revision_id=revision_id)

                except errors.NoRepositoryPresent:

                    # needed to make one anyway.

                    result_repo = local_repo.clone(

                        result,

                        revision_id=revision_id,

                        basis=basis_repo)

                    result_repo.set_make_working_trees(local_repo.make_working_trees())

        # 1 if there is a branch present

        #   make sure its content is available in the target repository

        #   clone it.

        try:

            self.open_branch().clone(result, revision_id=revision_id)

        except errors.NotBranchError:

            pass

        try:

            self.open_workingtree().clone(result, basis=basis_tree)

        except (errors.NoWorkingTree, errors.NotLocalUrl):

            pass

        return result

    def _get_basis_components(self, basis):

        """Retrieve the basis components that are available at basis."""

        if basis is None:

            return None, None, None

        try:

            basis_tree = basis.open_workingtree()

            basis_branch = basis_tree.branch

            basis_repo = basis_branch.repository

        except (errors.NoWorkingTree, errors.NotLocalUrl):

            basis_tree = None

            try:

                basis_branch = basis.open_branch()

                basis_repo = basis_branch.repository

            except errors.NotBranchError:

                basis_branch = None

                try:

                    basis_repo = basis.open_repository()

                except errors.NoRepositoryPresent:

                    basis_repo = None

        return basis_repo, basis_branch, basis_tree

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

    # this will open a new connection.

    def _make_tail(self, url):

        head, tail = urlutils.split(url)

        if tail and tail != '.':

            t = bzrlib.transport.get_transport(head)

            try:

                t.mkdir(tail)

            except errors.FileExists:

                pass

    # TODO: Should take a Transport

    @classmethod

    def create(cls, base):

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

        This will call the current default formats initialize with base

        as the only parameter.

        If you need a specific format, consider creating an instance

        of that and calling initialize().

        """

        if cls is not BzrDir:

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

                    "not one of %r" % cls)

        head, tail = urlutils.split(base)

        if tail and tail != '.':

            t = bzrlib.transport.get_transport(head)

            try:

                t.mkdir(tail)

            except errors.FileExists:

                pass

        return BzrDirFormat.get_default_format().initialize(safe_unicode(base))

    def create_branch(self):

        """Create a branch in this BzrDir.

        The bzrdirs format will control what branch format is created.

        For more control see BranchFormatXX.create(a_bzrdir).

        """

        raise NotImplementedError(self.create_branch)

    @staticmethod

    def create_branch_and_repo(base, force_new_repo=False):

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

        This will use the current default BzrDirFormat, 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.

        """

        bzrdir = BzrDir.create(base)

        bzrdir._find_or_create_repository(force_new_repo)

        return bzrdir.create_branch()

    def _find_or_create_repository(self, force_new_repo):

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

        if force_new_repo:

            return self.create_repository()

        try:

            return self.find_repository()

        except errors.NoRepositoryPresent:

            return self.create_repository()

    @staticmethod

    def create_branch_convenience(base, force_new_repo=False,

                                  force_new_tree=None, format=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, 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 for the bzrdir format to create

        """

        if force_new_tree:

            # check for non local urls

            t = get_transport(safe_unicode(base))

            if not isinstance(t, LocalTransport):

                raise errors.NotLocalUrl(base)

        if format is None:

            bzrdir = BzrDir.create(base)

        else:

            bzrdir = format.initialize(base)

        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_repository(base, shared=False):

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

        This will use the current default BzrDirFormat, and use whatever 

        repository format that that uses for bzrdirformat.create_repository.

        ;param shared: Create a shared repository rather than a standalone

                       repository.

        The Repository object is returned.

        This must be overridden as an instance method in child classes, where

        it should take no parameters and construct whatever repository format

        that child class desires.

        """

        bzrdir = BzrDir.create(base)

        return bzrdir.create_repository(shared)

    @staticmethod

    def create_standalone_workingtree(base):

        """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, and use whatever 

        repository format that that uses for bzrdirformat.create_workingtree,

        create_branch and create_repository.

        The WorkingTree object is returned.

        """

        t = get_transport(safe_unicode(base))

        if not isinstance(t, LocalTransport):

            raise errors.NotLocalUrl(base)

        bzrdir = BzrDir.create_branch_and_repo(safe_unicode(base),

                                               force_new_repo=True).bzrdir

        return bzrdir.create_workingtree()

    def create_workingtree(self, revision_id=None):

        """Create a working tree at this BzrDir.

        revision_id: create it as of this revision id.

        """

        raise NotImplementedError(self.create_workingtree)

    def find_repository(self):

        """Find the repository that should be used for a_bzrdir.

        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.

        """

        try:

            return self.open_repository()

        except errors.NoRepositoryPresent:

            pass

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

        while True:

            # find the next containing bzrdir

            try:

                found_bzrdir = BzrDir.open_containing_from_transport(

                    next_transport)[0]

            except errors.NotBranchError:

                # none found

                raise errors.NoRepositoryPresent(self)

            # does it have a repository ?

            try:

                repository = found_bzrdir.open_repository()

            except errors.NoRepositoryPresent:

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

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

                    # top of the file system

                    break

                else:

                    continue

            if ((found_bzrdir.root_transport.base == 

                 self.root_transport.base) or repository.is_shared()):

                return repository

            else:

                raise errors.NoRepositoryPresent(self)

        raise errors.NoRepositoryPresent(self)

    def get_branch_transport(self, branch_format):

        """Get the transport for use by branch format in this BzrDir.

        Note that bzr dirs that do not support format strings will raise

        IncompatibleFormat if the branch format they are given has

        a format string, and vice versa.

        If branch_format is None, the transport is returned with no 

        checking. if it is not None, then the returned transport is

        guaranteed to point to an existing directory ready for use.

        """

        raise NotImplementedError(self.get_branch_transport)

    def get_repository_transport(self, repository_format):

        """Get the transport for use by repository format in this BzrDir.

        Note that bzr dirs that do not support format strings will raise

        IncompatibleFormat if the repository format they are given has

        a format string, and vice versa.

        If repository_format is None, the transport is returned with no 

        checking. if it is not None, then the returned transport is

        guaranteed to point to an existing directory ready for use.

        """

        raise NotImplementedError(self.get_repository_transport)

    def get_workingtree_transport(self, tree_format):

        """Get the transport for use by workingtree format in this BzrDir.

        Note that bzr dirs that do not support format strings will raise

        IncompatibleFormat if the workingtree format they are given has

        a format string, and vice versa.

        If workingtree_format is None, the transport is returned with no 

        checking. if it is not None, then the returned transport is

        guaranteed to point to an existing directory ready for use.

        """

        raise NotImplementedError(self.get_workingtree_transport)

    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

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

        self.root_transport = _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. it is expected that plugins will need to extend

        this in the future - for instance to make bzr talk with svn working

        trees.

        """

        # 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 methods

        # 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 needs_format_conversion(self, format=None):

        """Return true if this bzrdir needs convert_format run on it.

        For instance, if the repository format is out of date but the 

        branch and working tree are not, this should return True.

        :param format: Optional parameter indicating a specific desired

                       format we plan to arrive at.

        """

        raise NotImplementedError(self.needs_format_conversion)

    @staticmethod

    def open_unsupported(base):

        """Open a branch which is not supported."""

        return BzrDir.open(base, _unsupported=True)

    @staticmethod

    def open(base, _unsupported=False):

        """Open an existing bzrdir, rooted at 'base' (url)

        _unsupported is a private parameter to the BzrDir class.

        """

        t = get_transport(base)

        mutter("trying to open %r with transport %r", base, t)

        format = BzrDirFormat.find_format(t)

        BzrDir._check_supported(format, _unsupported)

        return format.open(t, _found=True)

    def open_branch(self, unsupported=False):

        """Open the branch object at this BzrDir if one is present.

        If unsupported is True, then no longer supported branch formats can

        still be opened.

        TODO: static convenience version of this?

        """

        raise NotImplementedError(self.open_branch)

    @staticmethod

    def open_containing(url):

        """Open an existing branch which contains url.

        :param url: url to search from.

        See open_containing_from_transport for more detail.

        """

        return BzrDir.open_containing_from_transport(get_transport(url))

    @staticmethod

    def open_containing_from_transport(a_transport):

        """Open an existing branch which contains a_transport.base

        This probes for a branch at a_transport, and searches upwards from there.

        Basically we keep looking up until we find the control directory or

        run into the root.  If there isn't one, raises NotBranchError.

        If there is one and it is either an unrecognised format or an unsupported 

        format, UnknownFormatError or UnsupportedFormatError are raised.

        If there is one, it is returned, along with the unused portion of url.

        :return: The BzrDir that contains the path, and a Unicode path 

                for the rest of the URL.

        """

        # this gets the normalised url back. I.e. '.' -> the full path.

        url = a_transport.base

        while True:

            try:

                format = BzrDirFormat.find_format(a_transport)

                BzrDir._check_supported(format, False)

                return format.open(a_transport), urlutils.unescape(a_transport.relpath(url))

            except errors.NotBranchError, e:

                ## mutter('not a branch in: %r %s', a_transport.base, e)

                pass

            new_t = a_transport.clone('..')

            if new_t.base == a_transport.base:

                # reached the root, whatever that may be

                raise errors.NotBranchError(path=url)

            a_transport = new_t

    def open_repository(self, _unsupported=False):

        """Open the repository object at this BzrDir if one is present.

        This will not follow the Branch object pointer - its strictly a direct

        open facility. Most client code should use open_branch().repository to

        get at a repository.

        _unsupported is a private parameter, not part of the api.

        TODO: static convenience version of this?

        """

        raise NotImplementedError(self.open_repository)

    def open_workingtree(self, _unsupported=False):

        """Open the workingtree object at this BzrDir if one is present.

        TODO: static convenience version of this?

        """

        raise NotImplementedError(self.open_workingtree)

    def has_branch(self):

        """Tell if this bzrdir contains a branch.

        Note: if you're going to open the branch, you should just go ahead

        and try, and not ask permission first.  (This method just opens the 

        branch and discards it, and that's somewhat expensive.) 

        """

        try:

            self.open_branch()

            return True

        except errors.NotBranchError:

            return False

    def has_workingtree(self):

        """Tell if this bzrdir contains a working tree.

        This will still raise an exception if the bzrdir has a workingtree that

        is remote & inaccessible.

        Note: if you're going to open the working tree, you should just go ahead

        and try, and not ask permission first.  (This method just opens the 

        workingtree and discards it, and that's somewhat expensive.) 

        """

        try:

            self.open_workingtree()

            return True

        except errors.NoWorkingTree:

            return False

    def sprout(self, url, revision_id=None, basis=None, force_new_repo=False):

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

        development.

        If urls 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.

        """

        self._make_tail(url)

        result = self._format.initialize(url)

        basis_repo, basis_branch, basis_tree = self._get_basis_components(basis)

        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:

                # copy the entire basis one if there is one

                # but there is no repository.

                source_repository = basis_repo

        if force_new_repo:

            result_repo = None

        else:

            try:

                result_repo = result.find_repository()

            except errors.NoRepositoryPresent:

                result_repo = None

        if source_repository is None and result_repo is not None:

            pass

        elif source_repository is None and result_repo is None:

            # no repo available, make a new one

            result.create_repository()

        elif source_repository is not None and result_repo is None:

            # have source, and want to make a new target repo

            # we don't clone the repo because that preserves attributes

            # like is_shared(), and we have not yet implemented a 

            # repository sprout().

            result_repo = result.create_repository()

        if result_repo is not None:

            # fetch needed content into target.

            if basis_repo:

                # XXX FIXME RBC 20060214 need tests for this when the basis

                # is incomplete

                result_repo.fetch(basis_repo, revision_id=revision_id)

            result_repo.fetch(source_repository, revision_id=revision_id)

        if source_branch is not None:

            source_branch.sprout(result, revision_id=revision_id)

        else:

            result.create_branch()

        # TODO: jam 20060426 we probably need a test in here in the

        #       case that the newly sprouted branch is a remote one

        if result_repo is None or result_repo.make_working_trees():

            result.create_workingtree()

        return result

class BzrDirPreSplitOut(BzrDir):

    """A common class for the all-in-one formats."""

    def __init__(self, _transport, _format):

        """See BzrDir.__init__."""

        super(BzrDirPreSplitOut, self).__init__(_transport, _format)

        assert self._format._lock_class == TransportLock

        assert self._format._lock_file_name == 'branch-lock'

        self._control_files = LockableFiles(self.get_branch_transport(None),

                                            self._format._lock_file_name,

                                            self._format._lock_class)

    def break_lock(self):

        """Pre-splitout bzrdirs do not suffer from stale locks."""

        raise NotImplementedError(self.break_lock)

    def clone(self, url, revision_id=None, basis=None, force_new_repo=False):

        """See BzrDir.clone()."""

        from bzrlib.workingtree import WorkingTreeFormat2

        self._make_tail(url)

        result = self._format._initialize_for_clone(url)

        basis_repo, basis_branch, basis_tree = self._get_basis_components(basis)

        self.open_repository().clone(result, revision_id=revision_id, basis=basis_repo)

        from_branch = self.open_branch()

        from_branch.clone(result, revision_id=revision_id)

        try:

            self.open_workingtree().clone(result, basis=basis_tree)

        except errors.NotLocalUrl:

            # make a new one, this format always has to have one.

            try:

                WorkingTreeFormat2().initialize(result)

            except errors.NotLocalUrl:

                # but we cannot do it for remote trees.

                to_branch = result.open_branch()

                WorkingTreeFormat2().stub_initialize_remote(to_branch.control_files)

        return result

    def create_branch(self):

        """See BzrDir.create_branch."""

        return self.open_branch()

    def create_repository(self, shared=False):

        """See BzrDir.create_repository."""

        if shared:

            raise errors.IncompatibleFormat('shared repository', self._format)

        return self.open_repository()

    def create_workingtree(self, revision_id=None):

        """See BzrDir.create_workingtree."""

        # this looks buggy but is not -really-

        # clone and sprout will have set the revision_id

        # and that will have set it for us, its only

        # specific uses of create_workingtree in isolation

        # that can do wonky stuff here, and that only

        # happens for creating checkouts, which cannot be 

        # done on this format anyway. So - acceptable wart.

        result = self.open_workingtree()

        if revision_id is not None:

            result.set_parent_trees([(revision_id,

                result.branch.repository.revision_tree(revision_id))])

        return result

    def get_branch_transport(self, branch_format):

        """See BzrDir.get_branch_transport()."""

        if branch_format is None:

            return self.transport

        try:

            branch_format.get_format_string()

        except NotImplementedError:

            return self.transport

        raise errors.IncompatibleFormat(branch_format, self._format)

    def get_repository_transport(self, repository_format):

        """See BzrDir.get_repository_transport()."""

        if repository_format is None:

            return self.transport

        try:

            repository_format.get_format_string()

        except NotImplementedError:

            return self.transport

        raise errors.IncompatibleFormat(repository_format, self._format)

    def get_workingtree_transport(self, workingtree_format):

        """See BzrDir.get_workingtree_transport()."""

        if workingtree_format is None:

            return self.transport

        try:

            workingtree_format.get_format_string()

        except NotImplementedError:

            return self.transport

        raise errors.IncompatibleFormat(workingtree_format, self._format)

    def needs_format_conversion(self, format=None):

        """See BzrDir.needs_format_conversion()."""

        # if the format is not the same as the system default,

        # an upgrade is needed.

        if format is None:

            format = BzrDirFormat.get_default_format()

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

    def open_branch(self, unsupported=False):

        """See BzrDir.open_branch."""

        from bzrlib.branch import BzrBranchFormat4

        format = BzrBranchFormat4()

        self._check_supported(format, unsupported)

        return format.open(self, _found=True)