/brz/remove-bazaar

# Copyright (C) 2005, 2006, 2007 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 bzrdir implementations - tests a bzrdir format."""

from cStringIO import StringIO

import errno

from itertools import izip

import os

from stat import S_ISDIR

import sys

import bzrlib.branch

from bzrlib import (

    bzrdir,

    check,

    errors,

    lockdir,

    osutils,

    repository,

    revision as _mod_revision,

    transactions,

    transport,

    ui,

    urlutils,

    workingtree,

    )

from bzrlib.branch import Branch, needs_read_lock, needs_write_lock

from bzrlib.errors import (FileExists,

                           NoSuchRevision,

                           NoSuchFile,

                           UninitializableFormat,

                           NotBranchError,

                           )

import bzrlib.revision

from bzrlib.tests import (

                          ChrootedTestCase,

                          TestCase,

                          TestCaseWithTransport,

                          TestNotApplicable,

                          TestSkipped,

                          )

from bzrlib.tests.per_bzrdir import TestCaseWithBzrDir

from bzrlib.trace import mutter

from bzrlib.transport import get_transport

from bzrlib.transport.local import LocalTransport

from bzrlib.ui import (

    CannedInputUIFactory,

    )

from bzrlib.upgrade import upgrade

from bzrlib.remote import RemoteBzrDir, RemoteRepository

from bzrlib.repofmt import weaverepo

class TestBzrDir(TestCaseWithBzrDir):

    # Many of these tests test for disk equality rather than checking

    # for semantic equivalence. This works well for some tests but

    # is not good at handling changes in representation or the addition

    # or removal of control data. It would be nice to for instance:

    # sprout a new branch, check that the nickname has been reset by hand

    # and then set the nickname to match the source branch, at which point

    # a semantic equivalence should pass

    def assertDirectoriesEqual(self, source, target, ignore_list=[]):

        """Assert that the content of source and target are identical.

        paths in ignore list will be completely ignored.

        We ignore paths that represent data which is allowed to change during

        a clone or sprout: for instance, inventory.knit contains gzip fragements

        which have timestamps in them, and as we have read the inventory from

        the source knit, the already-read data is recompressed rather than

        reading it again, which leads to changed timestamps. This is ok though,

        because the inventory.kndx file is not ignored, and the integrity of

        knit joins is tested by test_knit and test_versionedfile.

        :seealso: Additionally, assertRepositoryHasSameItems provides value

            rather than representation checking of repositories for

            equivalence.

        """

        files = []

        directories = ['.']

        while directories:

            dir = directories.pop()

            for path in set(source.list_dir(dir) + target.list_dir(dir)):

                path = dir + '/' + path

                if path in ignore_list:

                    continue

                try:

                    stat = source.stat(path)

                except errors.NoSuchFile:

                    self.fail('%s not in source' % path)

                if S_ISDIR(stat.st_mode):

                    self.assertTrue(S_ISDIR(target.stat(path).st_mode))

                    directories.append(path)

                else:

                    self.assertEqualDiff(source.get(path).read(),

                                         target.get(path).read(),

                                         "text for file %r differs:\n" % path)

    def assertRepositoryHasSameItems(self, left_repo, right_repo):

        """require left_repo and right_repo to contain the same data."""

        # XXX: TODO: Doesn't work yet, because we need to be able to compare

        # local repositories to remote ones...  but this is an as-yet unsolved

        # aspect of format management and the Remote protocols...

        # self.assertEqual(left_repo._format.__class__,

        #     right_repo._format.__class__)

        left_repo.lock_read()

        try:

            right_repo.lock_read()

            try:

                # revs

                all_revs = left_repo.all_revision_ids()

                self.assertEqual(left_repo.all_revision_ids(),

                    right_repo.all_revision_ids())

                for rev_id in left_repo.all_revision_ids():

                    self.assertEqual(left_repo.get_revision(rev_id),

                        right_repo.get_revision(rev_id))

                # Assert the revision trees (and thus the inventories) are equal

                sort_key = lambda rev_tree: rev_tree.get_revision_id()

                rev_trees_a = sorted(

                    left_repo.revision_trees(all_revs), key=sort_key)

                rev_trees_b = sorted(

                    right_repo.revision_trees(all_revs), key=sort_key)

                for tree_a, tree_b in zip(rev_trees_a, rev_trees_b):

                    self.assertEqual([], list(tree_a.iter_changes(tree_b)))

                # texts

                text_index = left_repo._generate_text_key_index()

                self.assertEqual(text_index,

                    right_repo._generate_text_key_index())

                desired_files = []

                for file_id, revision_id in text_index.iterkeys():

                    desired_files.append(

                        (file_id, revision_id, (file_id, revision_id)))

                left_texts = list(left_repo.iter_files_bytes(desired_files))

                right_texts = list(right_repo.iter_files_bytes(desired_files))

                left_texts.sort()

                right_texts.sort()

                self.assertEqual(left_texts, right_texts)

                # signatures

                for rev_id in all_revs:

                    try:

                        left_text = left_repo.get_signature_text(rev_id)

                    except NoSuchRevision:

                        continue

                    right_text = right_repo.get_signature_text(rev_id)

                    self.assertEqual(left_text, right_text)

            finally:

                right_repo.unlock()

        finally:

            left_repo.unlock()

    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:

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

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

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

        """BzrDir.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 BzrDir format doesn't support BzrDirs without working trees,

            # so this test is irrelevant.

            return

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

    def test_clone_on_transport(self):

        a_dir = self.make_bzrdir('source')

        target_transport = a_dir.root_transport.clone('..').clone('target')

        target = a_dir.clone_on_transport(target_transport)

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

        self.assertDirectoriesEqual(a_dir.root_transport, target.root_transport,

                                    ['./.bzr/merge-hashes'])

    def test_clone_bzrdir_empty(self):

        dir = self.make_bzrdir('source')

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

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

        self.assertDirectoriesEqual(dir.root_transport, target.root_transport,

                                    ['./.bzr/merge-hashes'])

    def test_clone_bzrdir_empty_force_new_ignored(self):

        # the force_new_repo parameter should have no effect on an empty

        # bzrdir's clone logic

        dir = self.make_bzrdir('source')

        target = dir.clone(self.get_url('target'), force_new_repo=True)

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

        self.assertDirectoriesEqual(dir.root_transport, target.root_transport,

                                    ['./.bzr/merge-hashes'])

    def test_clone_bzrdir_repository(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'))

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

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

        self.assertDirectoriesEqual(dir.root_transport, target.root_transport,

                                    [

                                     './.bzr/merge-hashes',

                                     './.bzr/repository',

                                     ])

        self.assertRepositoryHasSameItems(tree.branch.repository,

            target.open_repository())

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

        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'), force_new_repo=True)

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

        self.assertDirectoriesEqual(dir.root_transport, target.root_transport,

                                    ['./.bzr/repository',

                                     ])

        self.assertRepositoryHasSameItems(tree.branch.repository, repo)

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

        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)

        dir = source.bzrdir

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

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

        self.assertDirectoriesEqual(dir.root_transport, target.root_transport,

                                    [

                                     './.bzr/basis-inventory-cache',

                                     './.bzr/checkout/stat-cache',

                                     './.bzr/merge-hashes',

                                     './.bzr/repository',

                                     './.bzr/stat-cache',

                                    ])

        self.assertRepositoryHasSameItems(

            tree.branch.repository, target.open_repository())

    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_and_repo_into_shared_repo_force_new_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'), force_new_repo=True)

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

        repo = target.open_repository()

        self.assertDirectoriesEqual(dir.root_transport, target.root_transport,

                                    ['./.bzr/repository',

                                     ])

        self.assertRepositoryHasSameItems(tree.branch.repository, repo)

    def test_clone_bzrdir_branch_reference(self):

        # cloning should preserve the reference status of the branch in a bzrdir

        referenced_branch = self.make_branch('referencced')

        dir = self.make_bzrdir('source')

        try:

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

                referenced_branch)

        except errors.IncompatibleFormat:

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

            return

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

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

        self.assertDirectoriesEqual(dir.root_transport, target.root_transport)

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

        tree = self.make_branch_and_tree('source')

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

        tree.add('foo')

        tree.commit('revision 1')

        dir = tree.bzrdir

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

        self.skipIfNoWorkingTree(target)

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

        self.assertDirectoriesEqual(dir.root_transport, target.root_transport,

                                    ['./.bzr/stat-cache',

                                     './.bzr/checkout/dirstate',

                                     './.bzr/checkout/stat-cache',

                                     './.bzr/checkout/merge-hashes',

                                     './.bzr/merge-hashes',

                                     './.bzr/repository',

                                     ])

        self.assertRepositoryHasSameItems(tree.branch.repository,

            target.open_repository())

        target.open_workingtree().revert()

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

        tree = self.make_branch_and_tree('source')

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

        tree.add('foo')

        tree.commit('revision 1')

        dir = tree.bzrdir

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

        self.skipIfNoWorkingTree(target)

        self.assertDirectoriesEqual(dir.root_transport, target.root_transport,

                                    ['./.bzr/stat-cache',

                                     './.bzr/checkout/dirstate',

                                     './.bzr/checkout/stat-cache',

                                     './.bzr/checkout/merge-hashes',

                                     './.bzr/merge-hashes',

                                     './.bzr/repository',

                                     ])

        self.assertRepositoryHasSameItems(tree.branch.repository,

            target.open_repository())

        target.open_workingtree().revert()

        self.assertDirectoriesEqual(dir.root_transport, target.root_transport,

                                    ['./.bzr/stat-cache',

                                     './.bzr/checkout/dirstate',