~bzr-pqm/bzr/bzr.dev

# 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

"""Tests for control directory implementations - tests a controldir format."""

from itertools import izip

import bzrlib.branch

from bzrlib import (

    bzrdir,

    check,

    controldir,

    errors,

    gpg,

    osutils,

    transport,

    ui,

    urlutils,

    workingtree,

    )

import bzrlib.revision

from bzrlib.tests import (

    fixtures,

    ChrootedTestCase,

    TestNotApplicable,

    TestSkipped,

    )

from bzrlib.tests.per_controldir import TestCaseWithControlDir

from bzrlib.transport.local import LocalTransport

from bzrlib.ui import (

    CannedInputUIFactory,

    )

from bzrlib.remote import (

    RemoteBzrDir,

    RemoteBzrDirFormat,

    RemoteRepository,

    )

class TestControlDir(TestCaseWithControlDir):

    def skipIfNoWorkingTree(self, a_bzrdir):

        """Raises TestSkipped if a_bzrdir doesn't have a working tree.

        If the bzrdir does have a workingtree, this is a no-op.

        """

        try:

            a_bzrdir.open_workingtree()

        except (errors.NotLocalUrl, errors.NoWorkingTree):

            raise TestSkipped("bzrdir on transport %r has no working tree"

                              % a_bzrdir.transport)

    def openWorkingTreeIfLocal(self, a_bzrdir):

        """If a_bzrdir is on a local transport, call open_workingtree() on it.

        """

        if not isinstance(a_bzrdir.root_transport, LocalTransport):

            # it's not local, but that's ok

            return

        a_bzrdir.open_workingtree()

    def createWorkingTreeOrSkip(self, a_bzrdir):

        """Create a working tree on a_bzrdir, or raise TestSkipped.

        A simple wrapper for create_workingtree that translates NotLocalUrl into

        TestSkipped.  Returns the newly created working tree.

        """

        try:

            return a_bzrdir.create_workingtree()

        except (errors.NotLocalUrl, errors.UnsupportedOperation):

            raise TestSkipped("cannot make working tree with transport %r"

                              % a_bzrdir.transport)

    def sproutOrSkip(self, from_bzrdir, to_url, revision_id=None,

                     force_new_repo=False, accelerator_tree=None,

                     create_tree_if_local=True):

        """Sprout from_bzrdir into to_url, or raise TestSkipped.

        A simple wrapper for from_bzrdir.sprout that translates NotLocalUrl into

        TestSkipped.  Returns the newly sprouted bzrdir.

        """

        to_transport = transport.get_transport(to_url)

        if not isinstance(to_transport, LocalTransport):

            raise TestSkipped('Cannot sprout to remote bzrdirs.')

        target = from_bzrdir.sprout(to_url, revision_id=revision_id,

                                    force_new_repo=force_new_repo,

                                    possible_transports=[to_transport],

                                    accelerator_tree=accelerator_tree,

                                    create_tree_if_local=create_tree_if_local)

        return target

    def test_create_null_workingtree(self):

        dir = self.make_bzrdir('dir1')

        dir.create_repository()

        dir.create_branch()

        try:

            wt = dir.create_workingtree(revision_id=bzrlib.revision.NULL_REVISION)

        except (errors.NotLocalUrl, errors.UnsupportedOperation):

            raise TestSkipped("cannot make working tree with transport %r"

                              % dir.transport)

        self.assertEqual([], wt.get_parent_ids())

    def test_destroy_workingtree(self):

        tree = self.make_branch_and_tree('tree')

        self.build_tree(['tree/file'])

        tree.add('file')

        tree.commit('first commit')

        bzrdir = tree.bzrdir

        try:

            bzrdir.destroy_workingtree()

        except errors.UnsupportedOperation:

            raise TestSkipped('Format does not support destroying tree')

        self.failIfExists('tree/file')

        self.assertRaises(errors.NoWorkingTree, bzrdir.open_workingtree)

        bzrdir.create_workingtree()

        self.failUnlessExists('tree/file')

        bzrdir.destroy_workingtree_metadata()

        self.failUnlessExists('tree/file')

        self.assertRaises(errors.NoWorkingTree, bzrdir.open_workingtree)

    def test_destroy_branch(self):

        branch = self.make_branch('branch')

        bzrdir = branch.bzrdir

        try:

            bzrdir.destroy_branch()

        except (errors.UnsupportedOperation, errors.TransportNotPossible):

            raise TestNotApplicable('Format does not support destroying branch')

        self.assertRaises(errors.NotBranchError, bzrdir.open_branch)

        bzrdir.create_branch()

        bzrdir.open_branch()

    def test_destroy_repository(self):

        repo = self.make_repository('repository')

        bzrdir = repo.bzrdir

        try:

            bzrdir.destroy_repository()

        except (errors.UnsupportedOperation, errors.TransportNotPossible):

            raise TestNotApplicable('Format does not support destroying'

                                    ' repository')

        self.assertRaises(errors.NoRepositoryPresent, bzrdir.open_repository)

        bzrdir.create_repository()

        bzrdir.open_repository()

    def test_open_workingtree_raises_no_working_tree(self):

        """ControlDir.open_workingtree() should raise NoWorkingTree (rather than

        e.g. NotLocalUrl) if there is no working tree.

        """

        dir = self.make_bzrdir('source')

        vfs_dir = bzrdir.BzrDir.open(self.get_vfs_only_url('source'))

        if vfs_dir.has_workingtree():

            # This ControlDir format doesn't support ControlDirs without

            # working trees, so this test is irrelevant.

            return

        self.assertRaises(errors.NoWorkingTree, dir.open_workingtree)

    def test_clone_bzrdir_repository_under_shared(self):

        tree = self.make_branch_and_tree('commit_tree')

        self.build_tree(['foo'], transport=tree.bzrdir.transport.clone('..'))

        tree.add('foo')

        tree.commit('revision 1', rev_id='1')

        dir = self.make_bzrdir('source')

        repo = dir.create_repository()

        repo.fetch(tree.branch.repository)

        self.assertTrue(repo.has_revision('1'))

        try:

            self.make_repository('target', shared=True)

        except errors.IncompatibleFormat:

            return

        target = dir.clone(self.get_url('target/child'))

        self.assertNotEqual(dir.transport.base, target.transport.base)

        self.assertRaises(errors.NoRepositoryPresent, target.open_repository)

    def test_clone_bzrdir_repository_branch_both_under_shared(self):

        # Create a shared repository

        try:

            shared_repo = self.make_repository('shared', shared=True)

        except errors.IncompatibleFormat:

            return

        # Make a branch, 'commit_tree', and working tree outside of the shared

        # repository, and commit some revisions to it.

        tree = self.make_branch_and_tree('commit_tree')

        self.build_tree(['foo'], transport=tree.bzrdir.root_transport)

        tree.add('foo')

        tree.commit('revision 1', rev_id='1')

        tree.bzrdir.open_branch().set_revision_history([])

        tree.set_parent_trees([])

        tree.commit('revision 2', rev_id='2')

        # Copy the content (i.e. revisions) from the 'commit_tree' branch's

        # repository into the shared repository.

        tree.branch.repository.copy_content_into(shared_repo)

        # Make a branch 'source' inside the shared repository.

        dir = self.make_bzrdir('shared/source')

        dir.create_branch()

        # Clone 'source' to 'target', also inside the shared repository.

        target = dir.clone(self.get_url('shared/target'))

        # 'source', 'target', and the shared repo all have distinct bzrdirs.

        self.assertNotEqual(dir.transport.base, target.transport.base)

        self.assertNotEqual(dir.transport.base, shared_repo.bzrdir.transport.base)

        # The shared repository will contain revisions from the 'commit_tree'

        # repository, even revisions that are not part of the history of the

        # 'commit_tree' branch.

        self.assertTrue(shared_repo.has_revision('1'))

    def test_clone_bzrdir_repository_branch_only_source_under_shared(self):

        try:

            shared_repo = self.make_repository('shared', shared=True)

        except errors.IncompatibleFormat:

            return

        tree = self.make_branch_and_tree('commit_tree')

        self.build_tree(['commit_tree/foo'])

        tree.add('foo')

        tree.commit('revision 1', rev_id='1')

        tree.branch.bzrdir.open_branch().set_revision_history([])

        tree.set_parent_trees([])

        tree.commit('revision 2', rev_id='2')

        tree.branch.repository.copy_content_into(shared_repo)

        if shared_repo.make_working_trees():

            shared_repo.set_make_working_trees(False)

            self.assertFalse(shared_repo.make_working_trees())

        self.assertTrue(shared_repo.has_revision('1'))

        dir = self.make_bzrdir('shared/source')

        dir.create_branch()

        target = dir.clone(self.get_url('target'))

        self.assertNotEqual(dir.transport.base, target.transport.base)

        self.assertNotEqual(dir.transport.base, shared_repo.bzrdir.transport.base)

        branch = target.open_branch()

        self.assertTrue(branch.repository.has_revision('1'))

        self.assertFalse(branch.repository.make_working_trees())

        self.assertTrue(branch.repository.is_shared())

    def test_clone_bzrdir_repository_revision(self):

        # test for revision limiting, [smoke test, not corner case checks].

        # make a repository with some revisions,

        # and clone it with a revision limit.

        #

        tree = self.make_branch_and_tree('commit_tree')

        self.build_tree(['commit_tree/foo'])

        tree.add('foo')

        tree.commit('revision 1', rev_id='1')

        tree.branch.bzrdir.open_branch().set_revision_history([])

        tree.set_parent_trees([])

        tree.commit('revision 2', rev_id='2')

        source = self.make_repository('source')

        tree.branch.repository.copy_content_into(source)

        dir = source.bzrdir

        target = dir.clone(self.get_url('target'), revision_id='2')

        raise TestSkipped('revision limiting not strict yet')

    def test_clone_bzrdir_branch_and_repo_fixed_user_id(self):

        # Bug #430868 is about an email containing '.sig'

        self.overrideEnv('BZR_EMAIL', 'murphy@host.sighup.org')

        tree = self.make_branch_and_tree('commit_tree')

        self.build_tree(['commit_tree/foo'])

        tree.add('foo')

        rev1 = tree.commit('revision 1')

        tree_repo = tree.branch.repository

        tree_repo.lock_write()

        tree_repo.start_write_group()

        tree_repo.sign_revision(rev1, gpg.LoopbackGPGStrategy(None))

        tree_repo.commit_write_group()

        tree_repo.unlock()

        target = self.make_branch('target')

        tree.branch.repository.copy_content_into(target.repository)

        tree.branch.copy_content_into(target)

        self.assertTrue(target.repository.has_revision(rev1))

        self.assertEqual(

            tree_repo.get_signature_text(rev1),

            target.repository.get_signature_text(rev1))

    def test_clone_bzrdir_branch_and_repo_into_shared_repo(self):

        # by default cloning into a shared repo uses the shared repo.

        tree = self.make_branch_and_tree('commit_tree')

        self.build_tree(['commit_tree/foo'])

        tree.add('foo')

        tree.commit('revision 1')

        source = self.make_branch('source')

        tree.branch.repository.copy_content_into(source.repository)

        tree.branch.copy_content_into(source)

        try:

            self.make_repository('target', shared=True)

        except errors.IncompatibleFormat:

            return

        dir = source.bzrdir

        target = dir.clone(self.get_url('target/child'))

        self.assertNotEqual(dir.transport.base, target.transport.base)

        self.assertRaises(errors.NoRepositoryPresent, target.open_repository)

        self.assertEqual(source.revision_history(),

                         target.open_branch().revision_history())

    def test_clone_bzrdir_branch_revision(self):

        # test for revision limiting, [smoke test, not corner case checks].

        # make a branch with some revisions,

        # and clone it with a revision limit.

        #

        tree = self.make_branch_and_tree('commit_tree')

        self.build_tree(['commit_tree/foo'])

        tree.add('foo')

        tree.commit('revision 1', rev_id='1')

        tree.commit('revision 2', rev_id='2', allow_pointless=True)

        source = self.make_branch('source')

        tree.branch.repository.copy_content_into(source.repository)

        tree.branch.copy_content_into(source)

        dir = source.bzrdir

        target = dir.clone(self.get_url('target'), revision_id='1')

        self.assertEqual('1', target.open_branch().last_revision())

    def test_clone_on_transport_preserves_repo_format(self):

        if self.bzrdir_format == bzrdir.format_registry.make_bzrdir('default'):

            format = 'knit'

        else:

            format = None

        source_branch = self.make_branch('source', format=format)

        # Ensure no format data is cached

        a_dir = bzrlib.branch.Branch.open_from_transport(

            self.get_transport('source')).bzrdir

        target_transport = self.get_transport('target')

        target_bzrdir = a_dir.clone_on_transport(target_transport)

        target_repo = target_bzrdir.open_repository()

        source_branch = bzrlib.branch.Branch.open(

            self.get_vfs_only_url('source'))

        if isinstance(target_repo, RemoteRepository):

            target_repo._ensure_real()

            target_repo = target_repo._real_repository

        self.assertEqual(target_repo._format, source_branch.repository._format)

    def test_clone_bzrdir_tree_revision(self):

        # test for revision limiting, [smoke test, not corner case checks].

        # make a tree with a revision with a last-revision

        # and clone it with a revision limit.

        # This smoke test just checks the revision-id is right. Tree specific

        # tests will check corner cases.

        tree = self.make_branch_and_tree('source')

        self.build_tree(['source/foo'])

        tree.add('foo')

        tree.commit('revision 1', rev_id='1')

        tree.commit('revision 2', rev_id='2', allow_pointless=True)

        dir = tree.bzrdir

        target = dir.clone(self.get_url('target'), revision_id='1')

        self.skipIfNoWorkingTree(target)

        self.assertEqual(['1'], target.open_workingtree().get_parent_ids())

    def test_clone_bzrdir_into_notrees_repo(self):

        """Cloning into a no-trees repo should not create a working tree"""

        tree = self.make_branch_and_tree('source')

        self.build_tree(['source/foo'])

        tree.add('foo')

        tree.commit('revision 1')

        try:

            repo = self.make_repository('repo', shared=True)

        except errors.IncompatibleFormat:

            raise TestNotApplicable('must support shared repositories')

        if repo.make_working_trees():

            repo.set_make_working_trees(False)

            self.assertFalse(repo.make_working_trees())

        dir = tree.bzrdir

        a_dir = dir.clone(self.get_url('repo/a'))

        a_dir.open_branch()

        self.assertRaises(errors.NoWorkingTree, a_dir.open_workingtree)

    def test_clone_respects_stacked(self):

        branch = self.make_branch('parent')

        child_transport = self.get_transport('child')

        child = branch.bzrdir.clone_on_transport(child_transport,

                                                 stacked_on=branch.base)

        self.assertEqual(child.open_branch().get_stacked_on_url(), branch.base)

    def test_get_branch_reference_on_reference(self):

        """get_branch_reference should return the right url."""

        referenced_branch = self.make_branch('referenced')

        dir = self.make_bzrdir('source')

        try:

            reference = bzrlib.branch.BranchReferenceFormat().initialize(dir,

                target_branch=referenced_branch)

        except errors.IncompatibleFormat:

            # this is ok too, not all formats have to support references.

            return

        self.assertEqual(referenced_branch.bzrdir.root_transport.abspath('') + '/',

            dir.get_branch_reference())

    def test_get_branch_reference_on_non_reference(self):

        """get_branch_reference should return None for non-reference branches."""

        branch = self.make_branch('referenced')

        self.assertEqual(None, branch.bzrdir.get_branch_reference())

    def test_get_branch_reference_no_branch(self):

        """get_branch_reference should not mask NotBranchErrors."""

        dir = self.make_bzrdir('source')

        if dir.has_branch():

            # this format does not support branchless bzrdirs.

            return

        self.assertRaises(errors.NotBranchError, dir.get_branch_reference)

    def test_sprout_bzrdir_empty(self):

        dir = self.make_bzrdir('source')

        target = dir.sprout(self.get_url('target'))

        self.assertNotEqual(dir.transport.base, target.transport.base)

        # creates a new repository branch and tree

        target.open_repository()

        target.open_branch()

        self.openWorkingTreeIfLocal(target)

    def test_sprout_bzrdir_empty_under_shared_repo(self):

        # sprouting an empty dir into a repo uses the repo

        dir = self.make_bzrdir('source')

        try:

            self.make_repository('target', shared=True)

        except errors.IncompatibleFormat:

            return

        target = dir.sprout(self.get_url('target/child'))

        self.assertRaises(errors.NoRepositoryPresent, target.open_repository)

        target.open_branch()

        try:

            target.open_workingtree()

        except errors.NoWorkingTree:

            # Some bzrdirs can never have working trees.

            self.assertFalse(target._format.supports_workingtrees)

    def test_sprout_bzrdir_empty_under_shared_repo_force_new(self):

        # the force_new_repo parameter should force use of a new repo in an empty

        # bzrdir's sprout logic

        dir = self.make_bzrdir('source')

        try:

            self.make_repository('target', shared=True)

        except errors.IncompatibleFormat:

            return

        target = dir.sprout(self.get_url('target/child'), force_new_repo=True)

        target.open_repository()

        target.open_branch()

        self.openWorkingTreeIfLocal(target)

    def test_sprout_bzrdir_with_repository_to_shared(self):

        tree = self.make_branch_and_tree('commit_tree')

        self.build_tree(['commit_tree/foo'])

        tree.add('foo')

        tree.commit('revision 1', rev_id='1')

        tree.bzrdir.open_branch().set_revision_history([])

        tree.set_parent_trees([])

        tree.commit('revision 2', rev_id='2')

        source = self.make_repository('source')

        tree.branch.repository.copy_content_into(source)

        dir = source.bzrdir

        try:

            shared_repo = self.make_repository('target', shared=True)

        except errors.IncompatibleFormat:

            return

        target = dir.sprout(self.get_url('target/child'))

        self.assertNotEqual(dir.transport.base, target.transport.base)

        self.assertTrue(shared_repo.has_revision('1'))

    def test_sprout_bzrdir_repository_branch_both_under_shared(self):

        try:

            shared_repo = self.make_repository('shared', shared=True)

        except errors.IncompatibleFormat:

            return

        tree = self.make_branch_and_tree('commit_tree')

        self.build_tree(['commit_tree/foo'])

        tree.add('foo')

        tree.commit('revision 1', rev_id='1')

        tree.bzrdir.open_branch().set_revision_history([])

        tree.set_parent_trees([])

        tree.commit('revision 2', rev_id='2')

        tree.branch.repository.copy_content_into(shared_repo)

        dir = self.make_bzrdir('shared/source')

        dir.create_branch()

        target = dir.sprout(self.get_url('shared/target'))

        self.assertNotEqual(dir.transport.base, target.transport.base)

        self.assertNotEqual(dir.transport.base, shared_repo.bzrdir.transport.base)

        self.assertTrue(shared_repo.has_revision('1'))

    def test_sprout_bzrdir_repository_branch_only_source_under_shared(self):

        try:

            shared_repo = self.make_repository('shared', shared=True)

        except errors.IncompatibleFormat:

            return

        tree = self.make_branch_and_tree('commit_tree')

        self.build_tree(['commit_tree/foo'])

        tree.add('foo')

        tree.commit('revision 1', rev_id='1')

        tree.bzrdir.open_branch().set_revision_history([])

        tree.set_parent_trees([])

        tree.commit('revision 2', rev_id='2')

        tree.branch.repository.copy_content_into(shared_repo)

        if shared_repo.make_working_trees():

            shared_repo.set_make_working_trees(False)

            self.assertFalse(shared_repo.make_working_trees())

        self.assertTrue(shared_repo.has_revision('1'))

        dir = self.make_bzrdir('shared/source')

        dir.create_branch()

        target = dir.sprout(self.get_url('target'))

        self.assertNotEqual(dir.transport.base, target.transport.base)

        self.assertNotEqual(dir.transport.base, shared_repo.bzrdir.transport.base)

        branch = target.open_branch()

        # The sprouted bzrdir has a branch, so only revisions referenced by

        # that branch are copied, rather than the whole repository.  It's an

        # empty branch, so none are copied.

        self.assertEqual([], branch.repository.all_revision_ids())

        if branch.bzrdir._format.supports_workingtrees:

            self.assertTrue(branch.repository.make_working_trees())

        self.assertFalse(branch.repository.is_shared())

    def test_sprout_bzrdir_repository_under_shared_force_new_repo(self):

        tree = self.make_branch_and_tree('commit_tree')

        self.build_tree(['commit_tree/foo'])

        tree.add('foo')

        tree.commit('revision 1', rev_id='1')

        tree.bzrdir.open_branch().set_revision_history([])

        tree.set_parent_trees([])

        tree.commit('revision 2', rev_id='2')

        source = self.make_repository('source')

        tree.branch.repository.copy_content_into(source)

        dir = source.bzrdir

        try:

            shared_repo = self.make_repository('target', shared=True)

        except errors.IncompatibleFormat:

            return

        target = dir.sprout(self.get_url('target/child'), force_new_repo=True)

        self.assertNotEqual(dir.transport.base, target.transport.base)

        self.assertFalse(shared_repo.has_revision('1'))

    def test_sprout_bzrdir_repository_revision(self):

        # test for revision limiting, [smoke test, not corner case checks].

        # make a repository with some revisions,

        # and sprout it with a revision limit.

        #

        tree = self.make_branch_and_tree('commit_tree')

        self.build_tree(['commit_tree/foo'])

        tree.add('foo')

        tree.commit('revision 1', rev_id='1')

        tree.bzrdir.open_branch().set_revision_history([])

        tree.set_parent_trees([])

        tree.commit('revision 2', rev_id='2')

        source = self.make_repository('source')

        tree.branch.repository.copy_content_into(source)

        dir = source.bzrdir

        target = self.sproutOrSkip(dir, self.get_url('target'), revision_id='2')

        raise TestSkipped('revision limiting not strict yet')

    def test_sprout_bzrdir_branch_and_repo_shared(self):

        # sprouting a branch with a repo into a shared repo uses the shared

        # repo

        tree = self.make_branch_and_tree('commit_tree')

        self.build_tree(['commit_tree/foo'])

        tree.add('foo')

        tree.commit('revision 1', rev_id='1')

        source = self.make_branch('source')

        tree.branch.repository.copy_content_into(source.repository)

        tree.bzrdir.open_branch().copy_content_into(source)

        dir = source.bzrdir

        try:

            shared_repo = self.make_repository('target', shared=True)

        except errors.IncompatibleFormat:

            return

        target = dir.sprout(self.get_url('target/child'))

        self.assertTrue(shared_repo.has_revision('1'))

    def test_sprout_bzrdir_branch_and_repo_shared_force_new_repo(self):

        # sprouting a branch with a repo into a shared repo uses the shared

        # repo

        tree = self.make_branch_and_tree('commit_tree')

        self.build_tree(['commit_tree/foo'])

        tree.add('foo')

        tree.commit('revision 1', rev_id='1')

        source = self.make_branch('source')