/brz/remove-bazaar

# Copyright (C) 2006-2012, 2016 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."""

import breezy.branch

from breezy import (

    branch as _mod_branch,

    check,

    controldir,

    errors,

    gpg,

    osutils,

    repository as _mod_repository,

    revision as _mod_revision,

    transport,

    ui,

    urlutils,

    workingtree,

    )

from breezy.bzr import (

    bzrdir as _mod_bzrdir,

    )

from breezy.tests import (

    fixtures,

    ChrootedTestCase,

    TestNotApplicable,

    TestSkipped,

    )

from breezy.tests.per_controldir import TestCaseWithControlDir

from breezy.transport.local import LocalTransport

from breezy.ui import (

    CannedInputUIFactory,

    )

from breezy.bzr.remote import (

    RemoteBzrDir,

    RemoteBzrDirFormat,

    RemoteRepository,

    )

class TestControlDir(TestCaseWithControlDir):

    def skipIfNoWorkingTree(self, a_controldir):

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

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

        """

        try:

            a_controldir.open_workingtree()

        except (errors.NotLocalUrl, errors.NoWorkingTree):

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

                              % a_controldir.transport)

    def openWorkingTreeIfLocal(self, a_controldir):

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

        """

        if not isinstance(a_controldir.root_transport, LocalTransport):

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

            return

        a_controldir.open_workingtree()

    def createWorkingTreeOrSkip(self, a_controldir):

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

        A simple wrapper for create_workingtree that translates NotLocalUrl into

        TestSkipped.  Returns the newly created working tree.

        """

        try:

            return a_controldir.create_workingtree()

        except (errors.NotLocalUrl, errors.UnsupportedOperation):

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

                              % a_controldir.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_uninitializable(self):

        if self.bzrdir_format.is_initializable():

            raise TestNotApplicable("format is initializable")

        t = self.get_transport()

        self.assertRaises(errors.UninitializableFormat,

                          self.bzrdir_format.initialize, t.base)

    def test_multiple_initialization(self):

        # loopback test to check the current format initializes to itself.

        if not self.bzrdir_format.is_initializable():

            # unsupported formats are not loopback testable

            # because the default open will not open them and

            # they may not be initializable.

            raise TestNotApplicable("format is not initializable")

        self.bzrdir_format.initialize('.')

        self.assertRaises(errors.AlreadyControlDirError,

                          self.bzrdir_format.initialize, '.')

    def test_create_null_workingtree(self):

        dir = self.make_controldir('dir1')

        dir.create_repository()

        dir.create_branch()

        try:

            wt = dir.create_workingtree(

                revision_id=_mod_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.controldir

        try:

            bzrdir.destroy_workingtree()

        except errors.UnsupportedOperation:

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

        self.assertPathDoesNotExist('tree/file')

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

        bzrdir.create_workingtree()

        self.assertPathExists('tree/file')

        bzrdir.destroy_workingtree_metadata()

        self.assertPathExists('tree/file')

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

    def test_destroy_branch(self):

        branch = self.make_branch('branch')

        bzrdir = branch.controldir

        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_branch_no_branch(self):

        branch = self.make_repository('branch')

        bzrdir = branch.controldir

        try:

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

        except (errors.UnsupportedOperation, errors.TransportNotPossible):

            raise TestNotApplicable(

                'Format does not support destroying branch')

    def test_destroy_repository(self):

        repo = self.make_repository('repository')

        bzrdir = repo.controldir

        try:

            bzrdir.destroy_repository()

        except (errors.UnsupportedOperation, errors.TransportNotPossible):

            raise TestNotApplicable('Format does not support destroying'

                                    ' repository')

        self.assertRaises(errors.NoRepositoryPresent,

                          bzrdir.destroy_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_controldir('source')

        vfs_dir = controldir.ControlDir.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.

            raise TestNotApplicable("format does not support "

                                    "control directories without working tree")

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

    def test_clone_controldir_repository_under_shared(self):

        tree = self.make_branch_and_tree('commit_tree')

        self.build_tree(

            ['foo'], transport=tree.controldir.transport.clone('..'))

        tree.add('foo')

        rev1 = tree.commit('revision 1')

        dir = self.make_controldir('source')

        repo = dir.create_repository()

        if not repo._format.supports_nesting_repositories:

            raise TestNotApplicable("repository format does not support "

                                    "nesting")

        repo.fetch(tree.branch.repository)

        self.assertTrue(repo.has_revision(rev1))

        try:

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

        except errors.IncompatibleFormat:

            raise TestNotApplicable(

                "repository format does not support shared repositories")

        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_controldir_repository_branch_both_under_shared(self):

        # Create a shared repository

        try:

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

        except errors.IncompatibleFormat:

            raise TestNotApplicable(

                "repository format does not support shared repositories")

        if not shared_repo._format.supports_nesting_repositories:

            raise TestNotApplicable(

                "format does not support nesting repositories")

        # 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.controldir.root_transport)

        tree.add('foo')

        rev1 = tree.commit('revision 1')

        tree.controldir.open_branch().generate_revision_history(

            _mod_revision.NULL_REVISION)

        tree.set_parent_trees([])

        tree.commit('revision 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_controldir('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 controldirs.

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

        self.assertNotEqual(

            dir.transport.base, shared_repo.controldir.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(rev1))

    def test_clone_controldir_repository_branch_only_source_under_shared(self):

        try:

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

        except errors.IncompatibleFormat:

            raise TestNotApplicable(

                "repository format does not support shared repositories")

        if not shared_repo._format.supports_nesting_repositories:

            raise TestNotApplicable(

                "format does not support nesting repositories")

        tree = self.make_branch_and_tree('commit_tree')

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

        tree.add('foo')

        rev1 = tree.commit('revision 1')

        tree.branch.controldir.open_branch().generate_revision_history(

            _mod_revision.NULL_REVISION)

        tree.set_parent_trees([])

        tree.commit('revision 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(rev1))

        dir = self.make_controldir('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.controldir.transport.base)

        branch = target.open_branch()

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

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

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

    def test_clone_controldir_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')

        rev1 = tree.commit('revision 1')

        tree.branch.controldir.open_branch().generate_revision_history(

            _mod_revision.NULL_REVISION)

        tree.set_parent_trees([])

        rev2 = tree.commit('revision 2')

        source = self.make_repository('source')

        tree.branch.repository.copy_content_into(source)

        dir = source.controldir

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

        raise TestSkipped('revision limiting not strict yet')

    def test_clone_controldir_branch_and_repo_fixed_user_id(self):

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

        self.overrideEnv('BRZ_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

        if not tree_repo._format.supports_revision_signatures:

            self.skipTest('repository format does not support signing')

        with tree_repo.lock_write(), _mod_repository.WriteGroup(tree_repo):

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

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

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

        except errors.IncompatibleFormat:

            raise TestNotApplicable("repository format does not support "

                                    "shared repositories")

        if not shared_repo._format.supports_nesting_repositories:

            raise TestNotApplicable("format does not support nesting "

                                    "repositories")

        dir = source.controldir

        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.last_revision(),

                         target.open_branch().last_revision())

    def test_clone_controldir_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')

        rev1 = tree.commit('revision 1')

        tree.commit('revision 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.controldir

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

        self.assertEqual(rev1, target.open_branch().last_revision())

    def test_clone_controldir_with_colocated(self):

        if not self.bzrdir_format.colocated_branches:

            raise TestNotApplicable(

                'format does not supported colocated branches')

        tree = self.make_branch_and_tree('commit_tree')

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

        tree.add('foo')

        rev1 = tree.commit('revision 1')

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

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

        dir = tree.branch.controldir

        colo = dir.create_branch(name='colo')

        colo.pull(tree.branch, stop_revision=rev1)

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

        self.assertEqual(rev2, target.open_branch().last_revision())

        self.assertEqual(rev1, target.open_branch(name='colo').last_revision())

    def test_clone_on_transport_preserves_repo_format(self):

        if self.bzrdir_format == controldir.format_registry.make_controldir('default'):

            format = 'knit'

        else:

            format = None

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

        # Ensure no format data is cached

        a_dir = breezy.branch.Branch.open_from_transport(

            self.get_transport('source')).controldir

        target_transport = self.get_transport('target')

        target_controldir = a_dir.clone_on_transport(target_transport)

        target_repo = target_controldir.open_repository()

        source_branch = breezy.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_controldir_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')

        rev1 = tree.commit('revision 1')

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

        dir = tree.controldir

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

        self.skipIfNoWorkingTree(target)

        self.assertEqual([rev1], target.open_workingtree().get_parent_ids())

    def test_clone_controldir_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())

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

        a_branch = a_dir.open_branch()

        # If the new control dir actually uses the repository, it should

        # not have a working tree.

        if not a_branch.repository.has_same_location(repo):

            raise TestNotApplicable('new control dir does not use repository')

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

        try:

            child = branch.controldir.clone_on_transport(child_transport,

                                                         stacked_on=branch.base)

        except (_mod_branch.UnstackableBranchFormat,

                errors.UnstackableRepositoryFormat):

            raise TestNotApplicable("branch or repository format does "

                                    "not support stacking")

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

    def test_set_branch_reference(self):

        """set_branch_reference creates a branch reference"""

        referenced_branch = self.make_branch('referenced')

        dir = self.make_controldir('source')

        try:

            reference = dir.set_branch_reference(referenced_branch)

        except errors.IncompatibleFormat:

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

            raise TestNotApplicable("control directory does not "

                                    "support branch references")

        self.assertEqual(referenced_branch.user_url,

                         dir.get_branch_reference())

    def test_set_branch_reference_on_existing_reference(self):

        """set_branch_reference creates a branch reference"""

        referenced_branch1 = self.make_branch('old-referenced')

        referenced_branch2 = self.make_branch('new-referenced')

        dir = self.make_controldir('source')

        try:

            reference = dir.set_branch_reference(referenced_branch1)

        except errors.IncompatibleFormat:

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

            raise TestNotApplicable("control directory does not "

                                    "support branch references")

        reference = dir.set_branch_reference(referenced_branch2)

        self.assertEqual(

            referenced_branch2.user_url,

            dir.get_branch_reference())

    def test_set_branch_reference_on_existing_branch(self):

        """set_branch_reference creates a branch reference"""

        referenced_branch = self.make_branch('referenced')

        dir = self.make_branch('source').controldir

        try:

            reference = dir.set_branch_reference(referenced_branch)

        except errors.IncompatibleFormat:

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

            raise TestNotApplicable("control directory does not "

                                    "support branch references")

        self.assertEqual(

            referenced_branch.user_url,

            dir.get_branch_reference())

    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_controldir('source')

        try:

            dir.set_branch_reference(referenced_branch)

        except errors.IncompatibleFormat:

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

            raise TestNotApplicable("control directory does not "

                                    "support branch references")

        self.assertEqual(referenced_branch.user_url,

                         dir.get_branch_reference())

    def test_get_branch_reference_on_non_reference(self):

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

        dir = self.make_controldir('referenced')

        dir.create_repository()

        if dir._format.colocated_branches:

            # The default branch in a controldir might be a reference branch

            # (e.g. for Git), so let's create another one.

            name = 'foo'

        else:

            name = None

        branch = dir.create_branch(name)

        self.assertEqual(None, branch.controldir.get_branch_reference(name))

    def test_get_branch_reference_no_branch(self):

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

        dir = self.make_controldir('source')

        if dir.has_branch():

            # this format does not support branchless controldirs.

            raise TestNotApplicable("format does not support "

                                    "branchless control directories")

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

    def test_sprout_controldir_empty(self):

        dir = self.make_controldir('source')

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

        self.assertNotEqual(dir.control_transport.base,

                            target.control_transport.base)

        # creates a new repository branch and tree

        target.open_repository()

        target.open_branch()

        self.openWorkingTreeIfLocal(target)

    def test_sprout_controldir_empty_under_shared_repo(self):

        # sprouting an empty dir into a repo uses the repo

        dir = self.make_controldir('source')

        try:

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

        except errors.IncompatibleFormat:

            raise TestNotApplicable("format does not support shared "

                                    "repositories")

        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 controldirs can never have working trees.

            repo = target.find_repository()

            self.assertFalse(repo.controldir._format.supports_workingtrees)

    def test_sprout_controldir_empty_under_shared_repo_force_new(self):

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

        # controldir's sprout logic

        dir = self.make_controldir('source')

        try:

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

        except errors.IncompatibleFormat:

            raise TestNotApplicable("format does not support shared "

                                    "repositories")

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

        target.open_repository()

        target.open_branch()

        self.openWorkingTreeIfLocal(target)

    def test_sprout_controldir_with_repository_to_shared(self):

        tree = self.make_branch_and_tree('commit_tree')

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

        tree.add('foo')

        rev1 = tree.commit('revision 1')

        tree.controldir.open_branch().generate_revision_history(

            _mod_revision.NULL_REVISION)

        tree.set_parent_trees([])

        rev2 = tree.commit('revision 2')

        source = self.make_repository('source')

        tree.branch.repository.copy_content_into(source)

        dir = source.controldir