~bzr-pqm/bzr/bzr.dev

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

from bzrlib import (

    branch as _mod_branch,

    bzrdir,

    config,

    delta as _mod_delta,

    errors,

    gpg,

    merge,

    urlutils,

    transport,

    remote,

    repository,

    revision,

    symbol_versioning,

    tests,

    )

from bzrlib.tests import (

    per_branch,

    )

from bzrlib.tests.http_server import HttpServer

from bzrlib.transport import memory

class TestTestCaseWithBranch(per_branch.TestCaseWithBranch):

    def test_branch_format_matches_bzrdir_branch_format(self):

        bzrdir_branch_format = self.bzrdir_format.get_branch_format()

        self.assertIs(self.branch_format.__class__,

                      bzrdir_branch_format.__class__)

    def test_make_branch_gets_expected_format(self):

        branch = self.make_branch('.')

        self.assertIs(self.branch_format.__class__,

            branch._format.__class__)

class TestBranch(per_branch.TestCaseWithBranch):

    def test_create_tree_with_merge(self):

        tree = self.create_tree_with_merge()

        tree.lock_read()

        self.addCleanup(tree.unlock)

        graph = tree.branch.repository.get_graph()

        ancestry_graph = graph.get_parent_map(

            tree.branch.repository.all_revision_ids())

        self.assertEqual({'rev-1':('null:',),

                          'rev-2':('rev-1', ),

                          'rev-1.1.1':('rev-1', ),

                          'rev-3':('rev-2', 'rev-1.1.1', ),

                         }, ancestry_graph)

    def test_revision_ids_are_utf8(self):

        wt = self.make_branch_and_tree('tree')

        wt.commit('f', rev_id='rev1')

        wt.commit('f', rev_id='rev2')

        wt.commit('f', rev_id='rev3')

        br = self.get_branch()

        br.fetch(wt.branch)

        br.generate_revision_history('rev3')

        rh = br.revision_history()

        self.assertEqual(['rev1', 'rev2', 'rev3'], rh)

        for revision_id in rh:

            self.assertIsInstance(revision_id, str)

        last = br.last_revision()

        self.assertEqual('rev3', last)

        self.assertIsInstance(last, str)

        revno, last = br.last_revision_info()

        self.assertEqual(3, revno)

        self.assertEqual('rev3', last)

        self.assertIsInstance(last, str)

    def test_fetch_revisions(self):

        """Test fetch-revision operation."""

        wt = self.make_branch_and_tree('b1')

        b1 = wt.branch

        self.build_tree_contents([('b1/foo', 'hello')])

        wt.add(['foo'], ['foo-id'])

        wt.commit('lala!', rev_id='revision-1', allow_pointless=False)

        b2 = self.make_branch('b2')

        b2.fetch(b1)

        rev = b2.repository.get_revision('revision-1')

        tree = b2.repository.revision_tree('revision-1')

        tree.lock_read()

        self.addCleanup(tree.unlock)

        self.assertEqual(tree.get_file_text('foo-id'), 'hello')

    def test_get_revision_delta(self):

        tree_a = self.make_branch_and_tree('a')

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

        tree_a.add('foo', 'file1')

        tree_a.commit('rev1', rev_id='rev1')

        self.build_tree(['a/vla'])

        tree_a.add('vla', 'file2')

        tree_a.commit('rev2', rev_id='rev2')

        delta = tree_a.branch.get_revision_delta(1)

        self.assertIsInstance(delta, _mod_delta.TreeDelta)

        self.assertEqual([('foo', 'file1', 'file')], delta.added)

        delta = tree_a.branch.get_revision_delta(2)

        self.assertIsInstance(delta, _mod_delta.TreeDelta)

        self.assertEqual([('vla', 'file2', 'file')], delta.added)

    def get_unbalanced_tree_pair(self):

        """Return two branches, a and b, with one file in a."""

        tree_a = self.make_branch_and_tree('a')

        self.build_tree_contents([('a/b', 'b')])

        tree_a.add('b')

        tree_a.commit("silly commit", rev_id='A')

        tree_b = self.make_branch_and_tree('b')

        return tree_a, tree_b

    def get_balanced_branch_pair(self):

        """Returns br_a, br_b as with one commit in a, and b has a's stores."""

        tree_a, tree_b = self.get_unbalanced_tree_pair()

        tree_b.branch.repository.fetch(tree_a.branch.repository)

        return tree_a, tree_b

    def test_clone_partial(self):

        """Copy only part of the history of a branch."""

        # TODO: RBC 20060208 test with a revision not on revision-history.

        #       what should that behaviour be ? Emailed the list.

        # First, make a branch with two commits.

        wt_a = self.make_branch_and_tree('a')

        self.build_tree(['a/one'])

        wt_a.add(['one'])

        wt_a.commit('commit one', rev_id='1')

        self.build_tree(['a/two'])

        wt_a.add(['two'])

        wt_a.commit('commit two', rev_id='2')

        # Now make a copy of the repository.

        repo_b = self.make_repository('b')

        wt_a.branch.repository.copy_content_into(repo_b)

        # wt_a might be a lightweight checkout, so get a hold of the actual

        # branch (because you can't do a partial clone of a lightweight

        # checkout).

        branch = wt_a.branch.bzrdir.open_branch()

        # Then make a branch where the new repository is, but specify a revision

        # ID.  The new branch's history will stop at the specified revision.

        br_b = branch.clone(repo_b.bzrdir, revision_id='1')

        self.assertEqual('1', br_b.last_revision())

    def get_parented_branch(self):

        wt_a = self.make_branch_and_tree('a')

        self.build_tree(['a/one'])

        wt_a.add(['one'])

        wt_a.commit('commit one', rev_id='1')

        branch_b = wt_a.branch.bzrdir.sprout('b', revision_id='1').open_branch()

        self.assertEqual(wt_a.branch.base, branch_b.get_parent())

        return branch_b

    def test_clone_branch_nickname(self):

        # test the nick name is preserved always

        raise tests.TestSkipped('XXX branch cloning is not yet tested.')

    def test_clone_branch_parent(self):

        # test the parent is preserved always

        branch_b = self.get_parented_branch()

        repo_c = self.make_repository('c')

        branch_b.repository.copy_content_into(repo_c)

        branch_c = branch_b.clone(repo_c.bzrdir)

        self.assertNotEqual(None, branch_c.get_parent())

        self.assertEqual(branch_b.get_parent(), branch_c.get_parent())

        # We can also set a specific parent, and it should be honored

        random_parent = 'http://example.com/path/to/branch'

        branch_b.set_parent(random_parent)

        repo_d = self.make_repository('d')

        branch_b.repository.copy_content_into(repo_d)

        branch_d = branch_b.clone(repo_d.bzrdir)

        self.assertEqual(random_parent, branch_d.get_parent())

    def test_submit_branch(self):

        """Submit location can be queried and set"""

        branch = self.make_branch('branch')

        self.assertEqual(branch.get_submit_branch(), None)

        branch.set_submit_branch('sftp://example.com')

        self.assertEqual(branch.get_submit_branch(), 'sftp://example.com')

        branch.set_submit_branch('sftp://example.net')

        self.assertEqual(branch.get_submit_branch(), 'sftp://example.net')

    def test_public_branch(self):

        """public location can be queried and set"""

        branch = self.make_branch('branch')

        self.assertEqual(branch.get_public_branch(), None)

        branch.set_public_branch('sftp://example.com')

        self.assertEqual(branch.get_public_branch(), 'sftp://example.com')

        branch.set_public_branch('sftp://example.net')

        self.assertEqual(branch.get_public_branch(), 'sftp://example.net')

        branch.set_public_branch(None)

        self.assertEqual(branch.get_public_branch(), None)

    def test_record_initial_ghost(self):

        """Branches should support having ghosts."""

        wt = self.make_branch_and_tree('.')

        wt.set_parent_ids(['non:existent@rev--ision--0--2'],

            allow_leftmost_as_ghost=True)

        self.assertEqual(['non:existent@rev--ision--0--2'],

            wt.get_parent_ids())

        rev_id = wt.commit('commit against a ghost first parent.')

        rev = wt.branch.repository.get_revision(rev_id)

        self.assertEqual(rev.parent_ids, ['non:existent@rev--ision--0--2'])

        # parent_sha1s is not populated now, WTF. rbc 20051003

        self.assertEqual(len(rev.parent_sha1s), 0)

    def test_record_two_ghosts(self):

        """Recording with all ghosts works."""

        wt = self.make_branch_and_tree('.')

        wt.set_parent_ids([

                'foo@azkhazan-123123-abcabc',

                'wibble@fofof--20050401--1928390812',

            ],

            allow_leftmost_as_ghost=True)

        rev_id = wt.commit("commit from ghost base with one merge")

        # the revision should have been committed with two parents

        rev = wt.branch.repository.get_revision(rev_id)

        self.assertEqual(['foo@azkhazan-123123-abcabc',

            'wibble@fofof--20050401--1928390812'],

            rev.parent_ids)

    def test_bad_revision(self):

        self.assertRaises(errors.InvalidRevisionId,

                          self.get_branch().repository.get_revision,

                          None)

# TODO 20051003 RBC:

# compare the gpg-to-sign info for a commit with a ghost and

#     an identical tree without a ghost

# fetch missing should rewrite the TOC of weaves to list newly available parents.

    def test_sign_existing_revision(self):

        wt = self.make_branch_and_tree('.')

        branch = wt.branch

        wt.commit("base", allow_pointless=True, rev_id='A')

        from bzrlib.testament import Testament

        strategy = gpg.LoopbackGPGStrategy(None)

        branch.repository.lock_write()

        branch.repository.start_write_group()

        branch.repository.sign_revision('A', strategy)

        branch.repository.commit_write_group()

        branch.repository.unlock()

        self.assertEqual('-----BEGIN PSEUDO-SIGNED CONTENT-----\n' +

                         Testament.from_revision(branch.repository,

                         'A').as_short_text() +

                         '-----END PSEUDO-SIGNED CONTENT-----\n',

                         branch.repository.get_signature_text('A'))

    def test_store_signature(self):

        wt = self.make_branch_and_tree('.')

        branch = wt.branch

        branch.lock_write()

        try:

            branch.repository.start_write_group()

            try:

                branch.repository.store_revision_signature(

                    gpg.LoopbackGPGStrategy(None), 'FOO', 'A')

            except:

                branch.repository.abort_write_group()

                raise

            else:

                branch.repository.commit_write_group()

        finally:

            branch.unlock()

        # A signature without a revision should not be accessible.

        self.assertRaises(errors.NoSuchRevision,

                          branch.repository.has_signature_for_revision_id,

                          'A')

        wt.commit("base", allow_pointless=True, rev_id='A')

        self.assertEqual('-----BEGIN PSEUDO-SIGNED CONTENT-----\n'

                         'FOO-----END PSEUDO-SIGNED CONTENT-----\n',

                         branch.repository.get_signature_text('A'))

    def test_branch_keeps_signatures(self):

        wt = self.make_branch_and_tree('source')

        wt.commit('A', allow_pointless=True, rev_id='A')

        repo = wt.branch.repository

        repo.lock_write()

        repo.start_write_group()

        repo.sign_revision('A', gpg.LoopbackGPGStrategy(None))

        repo.commit_write_group()

        repo.unlock()

        #FIXME: clone should work to urls,

        # wt.clone should work to disks.

        self.build_tree(['target/'])

        d2 = repo.bzrdir.clone(urlutils.local_path_to_url('target'))

        self.assertEqual(repo.get_signature_text('A'),

                         d2.open_repository().get_signature_text('A'))

    def test_nicks_bzr(self):

        """Test the behaviour of branch nicks specific to bzr branches.

        Nicknames are implicitly the name of the branch's directory, unless an

        explicit nickname is set.  That is, an explicit nickname always

        overrides the implicit one.

        """

        t = self.get_transport()

        branch = self.make_branch('bzr.dev')

        if not isinstance(branch, _mod_branch.BzrBranch):

            raise tests.TestNotApplicable("not a bzr branch format")

        # The nick will be 'bzr.dev', because there is no explicit nick set.

        self.assertEqual(branch.nick, 'bzr.dev')

        # Move the branch to a different directory, 'bzr.ab'.  Now that branch

        # will report its nick as 'bzr.ab'.

        t.move('bzr.dev', 'bzr.ab')

        branch = _mod_branch.Branch.open(self.get_url('bzr.ab'))

        self.assertEqual(branch.nick, 'bzr.ab')

        # Set the branch nick explicitly.  This will ensure there's a branch

        # config file in the branch.

        branch.nick = "Aaron's branch"

        if not isinstance(branch, remote.RemoteBranch):

            self.assertTrue(branch._transport.has("branch.conf"))

        # Because the nick has been set explicitly, the nick is now always

        # "Aaron's branch", regardless of directory name.

        self.assertEqual(branch.nick, "Aaron's branch")

        t.move('bzr.ab', 'integration')

        branch = _mod_branch.Branch.open(self.get_url('integration'))

        self.assertEqual(branch.nick, "Aaron's branch")

        branch.nick = u"\u1234"

        self.assertEqual(branch.nick, u"\u1234")

    def test_nicks(self):

        """Test explicit and implicit branch nicknames.

        A nickname is always available, whether set explicitly or not.

        """

        t = self.get_transport()

        branch = self.make_branch('bzr.dev')

        # An implicit nick name is set; what it is exactly depends on the

        # format.

        self.assertIsInstance(branch.nick, basestring)

        # Set the branch nick explicitly.

        branch.nick = "Aaron's branch"

        # Because the nick has been set explicitly, the nick is now always

        # "Aaron's branch".

        self.assertEqual(branch.nick, "Aaron's branch")

        branch.nick = u"\u1234"

        self.assertEqual(branch.nick, u"\u1234")

    def test_commit_nicks(self):

        """Nicknames are committed to the revision"""

        wt = self.make_branch_and_tree('bzr.dev')

        branch = wt.branch

        branch.nick = "My happy branch"

        wt.commit('My commit respect da nick.')

        committed = branch.repository.get_revision(branch.last_revision())

        self.assertEqual(committed.properties["branch-nick"],

                         "My happy branch")

    def test_create_colocated(self):

        try:

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

        except errors.IncompatibleFormat:

            return

        self.assertEquals(0, len(repo.bzrdir.list_branches()))

        try:

            child_branch1 = self.branch_format.initialize(repo.bzrdir, 

                name='branch1')

        except (errors.UninitializableFormat, errors.NoColocatedBranchSupport):

            # branch references are not default init'able and

            # not all bzrdirs support colocated branches.

            return

        self.assertEquals(1, len(repo.bzrdir.list_branches()))

        self.branch_format.initialize(repo.bzrdir, name='branch2')

        self.assertEquals(2, len(repo.bzrdir.list_branches()))

    def test_create_append_revisions_only(self):

        try:

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

        except errors.IncompatibleFormat:

            return

        for val in (True, False):

            try:

                branch = self.branch_format.initialize(repo.bzrdir,

                    append_revisions_only=True)

            except (errors.UninitializableFormat, errors.UpgradeRequired):

                # branch references are not default init'able and

                # not all branches support append_revisions_only

                return

            self.assertEquals(True, branch.get_append_revisions_only())

            repo.bzrdir.destroy_branch()

    def test_get_set_append_revisions_only(self):

        branch = self.make_branch('.')

        if branch._format.supports_set_append_revisions_only():

            branch.set_append_revisions_only(True)

            self.assertTrue(branch.get_append_revisions_only())

            branch.set_append_revisions_only(False)

            self.assertFalse(branch.get_append_revisions_only())

        else:

            self.assertRaises(errors.UpgradeRequired,

                branch.set_append_revisions_only, True)

            self.assertFalse(branch.get_append_revisions_only())

    def test_create_open_branch_uses_repository(self):

        try:

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

        except errors.IncompatibleFormat:

            return

        child_transport = repo.bzrdir.root_transport.clone('child')

        child_transport.mkdir('.')

        try:

            child_dir = self.bzrdir_format.initialize_on_transport(child_transport)

        except errors.UninitializableFormat:

            return

        try:

            child_branch = self.branch_format.initialize(child_dir)

        except errors.UninitializableFormat:

            # branch references are not default init'able.

            return

        self.assertEqual(repo.bzrdir.root_transport.base,

                         child_branch.repository.bzrdir.root_transport.base)

        child_branch = _mod_branch.Branch.open(self.get_url('child'))

        self.assertEqual(repo.bzrdir.root_transport.base,

                         child_branch.repository.bzrdir.root_transport.base)

    def test_format_description(self):

        tree = self.make_branch_and_tree('tree')

        text = tree.branch._format.get_format_description()

        self.assertTrue(len(text))

    def test_get_commit_builder(self):

        branch = self.make_branch(".")

        branch.lock_write()

        builder = branch.get_commit_builder([])

        self.assertIsInstance(builder, repository.CommitBuilder)

        branch.repository.commit_write_group()

        branch.unlock()

    def test_generate_revision_history(self):

        """Create a fake revision history easily."""

        tree = self.make_branch_and_tree('.')

        rev1 = tree.commit('foo')

        orig_history = tree.branch.revision_history()

        rev2 = tree.commit('bar', allow_pointless=True)

        tree.branch.generate_revision_history(rev1)

        self.assertEqual(orig_history, tree.branch.revision_history())

    def test_generate_revision_history_NULL_REVISION(self):

        tree = self.make_branch_and_tree('.')

        rev1 = tree.commit('foo')

        tree.branch.generate_revision_history(revision.NULL_REVISION)

        self.assertEqual([], tree.branch.revision_history())

    def test_create_checkout(self):

        tree_a = self.make_branch_and_tree('a')

        branch_a = tree_a.branch

        checkout_b = branch_a.create_checkout('b')

        self.assertEqual('null:', checkout_b.last_revision())

        checkout_b.commit('rev1', rev_id='rev1')

        self.assertEqual('rev1', branch_a.last_revision())

        self.assertNotEqual(checkout_b.branch.base, branch_a.base)

        checkout_c = branch_a.create_checkout('c', lightweight=True)

        self.assertEqual('rev1', checkout_c.last_revision())

        checkout_c.commit('rev2', rev_id='rev2')

        self.assertEqual('rev2', branch_a.last_revision())

        self.assertEqual(checkout_c.branch.base, branch_a.base)

        checkout_d = branch_a.create_checkout('d', lightweight=True)

        self.assertEqual('rev2', checkout_d.last_revision())

        checkout_e = branch_a.create_checkout('e')

        self.assertEqual('rev2', checkout_e.last_revision())

    def test_create_anonymous_lightweight_checkout(self):

        """A lightweight checkout from a readonly branch should succeed."""

        tree_a = self.make_branch_and_tree('a')

        rev_id = tree_a.commit('put some content in the branch')

        # open the branch via a readonly transport

        source_branch = _mod_branch.Branch.open(self.get_readonly_url('a'))

        # sanity check that the test will be valid

        self.assertRaises((errors.LockError, errors.TransportNotPossible),

            source_branch.lock_write)

        checkout = source_branch.create_checkout('c', lightweight=True)

        self.assertEqual(rev_id, checkout.last_revision())

    def test_create_anonymous_heavyweight_checkout(self):

        """A regular checkout from a readonly branch should succeed."""

        tree_a = self.make_branch_and_tree('a')

        rev_id = tree_a.commit('put some content in the branch')

        # open the branch via a readonly transport

        source_branch = _mod_branch.Branch.open(self.get_readonly_url('a'))

        # sanity check that the test will be valid

        self.assertRaises((errors.LockError, errors.TransportNotPossible),

            source_branch.lock_write)

        checkout = source_branch.create_checkout('c')

        self.assertEqual(rev_id, checkout.last_revision())

    def test_set_revision_history(self):

        tree = self.make_branch_and_tree('a')

        tree.commit('a commit', rev_id='rev1')

        br = tree.branch

        self.applyDeprecated(symbol_versioning.deprecated_in((2, 4, 0)),

            br.set_revision_history, ["rev1"])

        self.assertEquals(br.revision_history(), ["rev1"])

        self.applyDeprecated(symbol_versioning.deprecated_in((2, 4, 0)),

            br.set_revision_history, [])

        self.assertEquals(br.revision_history(), [])

    def test_heads_to_fetch(self):

        # heads_to_fetch is a method that returns a collection of revids that

        # need to be fetched to copy this branch into another repo.  At a

        # minimum this will include the tip.

        # (In native formats, this is the tip + tags, but other formats may

        # have other revs needed)

        tree = self.make_branch_and_tree('a')

        tree.commit('first commit', rev_id='rev1')

        tree.commit('second commit', rev_id='rev2')

        must_fetch, should_fetch = tree.branch.heads_to_fetch()

        self.assertTrue('rev2' in must_fetch)

    def test_heads_to_fetch_not_null_revision(self):

        # NULL_REVISION does not appear in the result of heads_to_fetch, even

        # for an empty branch.

        tree = self.make_branch_and_tree('a')

        must_fetch, should_fetch = tree.branch.heads_to_fetch()

        self.assertFalse(revision.NULL_REVISION in must_fetch)

        self.assertFalse(revision.NULL_REVISION in should_fetch)

class TestBranchFormat(per_branch.TestCaseWithBranch):

    def test_branch_format_network_name(self):

        br = self.make_branch('.')

        format = br._format

        network_name = format.network_name()

        self.assertIsInstance(network_name, str)

        # We want to test that the network_name matches the actual format on

        # disk. For local branches that means that using network_name as a key

        # in the registry gives back the same format. For remote branches we

        # check that the network_name of the RemoteBranchFormat we have locally

        # matches the actual format present on disk.

        if isinstance(format, remote.RemoteBranchFormat):

            br._ensure_real()

            real_branch = br._real_branch

            self.assertEqual(real_branch._format.network_name(), network_name)

        else:

            registry = _mod_branch.network_format_registry

            looked_up_format = registry.get(network_name)

            self.assertEqual(format.__class__, looked_up_format.__class__)

    def get_get_config_calls(self):

        # Smoke test that all branch succeed getting a config

        br = self.make_branch('.')

        br.get_config()

        br.get_config_stack()

class ChrootedTests(per_branch.TestCaseWithBranch):

    """A support class that provides readonly urls outside the local namespace.

    This is done by checking if self.transport_server is a MemoryServer. if it

    is then we are chrooted already, if it is not then an HttpServer is used

    for readonly urls.

    """

    def setUp(self):

        super(ChrootedTests, self).setUp()

        if not self.vfs_transport_factory == memory.MemoryServer:

            self.transport_readonly_server = HttpServer

    def test_open_containing(self):

        self.assertRaises(errors.NotBranchError,

                          _mod_branch.Branch.open_containing,

                          self.get_readonly_url(''))