/brz/remove-bazaar

# Copyright (C) 2006-2010 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.pull behaviour."""

from breezy import (

    branch,

    controldir,

    errors,

    revision,

    )

from breezy.tests import (

    fixtures,

    per_branch,

    TestNotApplicable,

    )

class TestPull(per_branch.TestCaseWithBranch):

    def test_pull_convergence_simple(self):

        # when revisions are pulled, the left-most accessible parents must

        # become the revision-history.

        parent = self.make_branch_and_tree('parent')

        p1 = parent.commit('1st post', allow_pointless=True)

        mine = parent.controldir.sprout('mine').open_workingtree()

        m1 = mine.commit('my change', allow_pointless=True)

        parent.merge_from_branch(mine.branch)

        p2 = parent.commit('merge my change')

        mine.pull(parent.branch)

        self.assertEqual(p2, mine.branch.last_revision())

    def test_pull_merged_indirect(self):

        # it should be possible to do a pull from one branch into another

        # when the tip of the target was merged into the source branch

        # via a third branch - so its buried in the ancestry and is not

        # directly accessible.

        parent = self.make_branch_and_tree('parent')

        p1 = parent.commit('1st post', allow_pointless=True)

        mine = parent.controldir.sprout('mine').open_workingtree()

        m1 = mine.commit('my change', allow_pointless=True)

        other = parent.controldir.sprout('other').open_workingtree()

        other.merge_from_branch(mine.branch)

        o2 = other.commit('merge my change')

        parent.merge_from_branch(other.branch)

        p2 = parent.commit('merge other')

        mine.pull(parent.branch)

        self.assertEqual(p2, mine.branch.last_revision())

    def test_pull_updates_checkout_and_master(self):

        """Pulling into a checkout updates the checkout and the master branch"""

        master_tree = self.make_branch_and_tree('master')

        rev1 = master_tree.commit('master')

        checkout = master_tree.branch.create_checkout('checkout')

        other = master_tree.branch.controldir.sprout(

            'other').open_workingtree()

        rev2 = other.commit('other commit')

        # now pull, which should update both checkout and master.

        checkout.branch.pull(other.branch)

        self.assertEqual(rev2, checkout.branch.last_revision())

        self.assertEqual(rev2, master_tree.branch.last_revision())

    def test_pull_local_updates_checkout_only(self):

        """Pulling --local into a checkout updates the checkout and not the

        master branch"""

        master_tree = self.make_branch_and_tree('master')

        rev1 = master_tree.commit('master')

        checkout = master_tree.branch.create_checkout('checkout')

        other = master_tree.branch.controldir.sprout(

            'other').open_workingtree()

        rev2 = other.commit('other commit')

        # now pull local, which should update checkout but not master.

        checkout.branch.pull(other.branch, local=True)

        self.assertEqual(rev2, checkout.branch.last_revision())

        self.assertEqual(rev1, master_tree.branch.last_revision())

    def test_pull_local_raises_LocalRequiresBoundBranch_on_unbound(self):

        """Pulling --local into a branch that is not bound should fail."""

        master_tree = self.make_branch_and_tree('branch')

        rev1 = master_tree.commit('master')

        other = master_tree.branch.controldir.sprout(

            'other').open_workingtree()

        rev2 = other.commit('other commit')

        # now pull --local, which should raise LocalRequiresBoundBranch error.

        self.assertRaises(errors.LocalRequiresBoundBranch,

                          master_tree.branch.pull, other.branch, local=True)

        self.assertEqual(rev1, master_tree.branch.last_revision())

    def test_pull_returns_result(self):

        parent = self.make_branch_and_tree('parent')

        p1 = parent.commit('1st post')

        mine = parent.controldir.sprout('mine').open_workingtree()

        m1 = mine.commit('my change')

        result = parent.branch.pull(mine.branch)

        self.assertIsNot(None, result)

        self.assertIs(mine.branch, result.source_branch)

        self.assertIs(parent.branch, result.target_branch)

        self.assertIs(parent.branch, result.master_branch)

        self.assertIs(None, result.local_branch)

        self.assertEqual(1, result.old_revno)

        self.assertEqual(p1, result.old_revid)

        self.assertEqual(2, result.new_revno)

        self.assertEqual(m1, result.new_revid)

        self.assertEqual([], list(result.tag_conflicts))

    def test_pull_overwrite(self):

        tree_a = self.make_branch_and_tree('tree_a')

        tree_a.commit('message 1')

        tree_b = tree_a.controldir.sprout('tree_b').open_workingtree()

        rev2a = tree_a.commit('message 2a')

        rev2b = tree_b.commit('message 2b')

        self.assertRaises(errors.DivergedBranches, tree_a.pull, tree_b.branch)

        self.assertRaises(errors.DivergedBranches,

                          tree_a.branch.pull, tree_b.branch,

                          overwrite=False, stop_revision=rev2b)

        # It should not have updated the branch tip, but it should have fetched

        # the revision if the repository supports "invisible" revisions

        self.assertEqual(rev2a, tree_a.branch.last_revision())

        if tree_a.branch.repository._format.supports_unreferenced_revisions:

            self.assertTrue(tree_a.branch.repository.has_revision(rev2b))

        tree_a.branch.pull(tree_b.branch, overwrite=True,

                           stop_revision=rev2b)

        self.assertEqual(rev2b, tree_a.branch.last_revision())

        self.assertEqual(tree_b.branch.last_revision(),

                         tree_a.branch.last_revision())

    def test_pull_overwrite_set(self):

        tree_a = self.make_branch_and_tree('tree_a')

        rev1 = tree_a.commit('message 1')

        tree_b = tree_a.controldir.sprout('tree_b').open_workingtree()

        rev2a = tree_a.commit('message 2a')

        rev2b = tree_b.commit('message 2b')

        self.assertRaises(errors.DivergedBranches, tree_a.pull, tree_b.branch)

        self.assertRaises(errors.DivergedBranches,

                          tree_a.branch.pull, tree_b.branch,

                          overwrite=set(), stop_revision=rev2b)

        # It should not have updated the branch tip, but it should have fetched

        # the revision if the repository supports "invisible" revisions

        self.assertEqual(rev2a, tree_a.branch.last_revision())

        if tree_a.branch.repository._format.supports_unreferenced_revisions:

            self.assertTrue(tree_a.branch.repository.has_revision(rev2b))

        tree_a.branch.pull(tree_b.branch, overwrite=set(['history']),

                           stop_revision=rev2b)

        self.assertEqual(rev2b, tree_a.branch.last_revision())

        self.assertEqual(tree_b.branch.last_revision(),

                         tree_a.branch.last_revision())

        tree_a.branch.pull(tree_b.branch, overwrite=set(['history', 'tags']),

                           stop_revision=rev2b)

    def test_pull_overwrite_set_tags(self):

        tree_a = self.make_branch_and_tree('tree_a')

        if not tree_a.branch.supports_tags():

            raise TestNotApplicable("branch does not support tags")

        rev1 = tree_a.commit('message 1')

        tree_a.branch.tags.set_tag('tag1', rev1)

        tree_b = tree_a.controldir.sprout('tree_b').open_workingtree()

        rev2b = tree_b.commit('message 2b')

        tree_b.branch.tags.set_tag('tag1', rev2b)

        rev1b = tree_a.commit('message 1b')

        tree_a.branch.get_config_stack().set('branch.fetch_tags', True)

        self.assertRaises(errors.DivergedBranches, tree_a.pull, tree_b.branch)

        self.assertRaises(errors.DivergedBranches,

                          tree_a.branch.pull, tree_b.branch,

                          overwrite=set(), stop_revision=rev2b)

        # It should not have updated the branch tip, but it should have fetched

        # the revision if the repository supports "invisible" revisions

        self.assertEqual(rev1b, tree_a.branch.last_revision())

        # It also should not have updated the tags

        self.assertEqual(tree_a.branch.tags.get_tag_dict(), {'tag1': rev1})

        if tree_a.branch.repository._format.supports_unreferenced_revisions:

            self.assertTrue(tree_a.branch.repository.has_revision(rev2b))

        tree_a.branch.pull(tree_b.branch, overwrite=set(['history']),

                           stop_revision=rev2b)

        self.assertEqual(rev2b, tree_a.branch.last_revision())

        self.assertEqual(tree_b.branch.last_revision(),

                         tree_a.branch.last_revision())

        self.assertEqual(rev1, tree_a.branch.tags.lookup_tag('tag1'))

        tree_a.branch.pull(tree_b.branch, overwrite=set(['history', 'tags']),

                           stop_revision=rev2b)

        self.assertEqual(rev2b, tree_a.branch.tags.lookup_tag('tag1'))

    def test_pull_merges_and_fetches_tags(self):

        """Tags are updated by br.pull(source), and revisions named in those

        tags are fetched.

        """

        # Make a source, sprout a target off it

        try:

            builder = self.make_branch_builder('source')

        except errors.UninitializableFormat:

            raise TestNotApplicable('uninitializeable format')

        source, rev1, rev2 = fixtures.build_branch_with_non_ancestral_rev(

            builder)

        target = source.controldir.sprout('target').open_branch()

        # Add a tag to the source, then pull from source

        try:

            source.tags.set_tag('tag-a', rev2)

        except errors.TagsNotSupported:

            raise TestNotApplicable('format does not support tags.')

        source.tags.set_tag('tag-a', rev2)

        source.get_config_stack().set('branch.fetch_tags', True)

        target.pull(source)

        # The tag is present, and so is its revision.

        self.assertEqual(rev2, target.tags.lookup_tag('tag-a'))

        target.repository.get_revision(rev2)

    def test_pull_stop_revision_merges_and_fetches_tags(self):

        """br.pull(source, stop_revision=REV) updates and fetches tags."""

        # Make a source, sprout a target off it

        try:

            builder = self.make_branch_builder('source')

        except errors.UninitializableFormat:

            raise TestNotApplicable('uninitializeable format')

        source, rev1, rev2 = fixtures.build_branch_with_non_ancestral_rev(

            builder)

        target = source.controldir.sprout('target').open_branch()

        # Add a new commit to the ancestry

        rev_2_again = builder.build_commit(message="Rev 2 again")

        # Add a tag to the source, then pull rev_2_again from source

        try:

            source.tags.set_tag('tag-a', rev2)

        except errors.TagsNotSupported:

            raise TestNotApplicable('format does not support tags.')

        source.get_config_stack().set('branch.fetch_tags', True)

        target.pull(source, stop_revision=rev_2_again)

        # The tag is present, and so is its revision.

        self.assertEqual(rev2, target.tags.lookup_tag('tag-a'))

        target.repository.get_revision(rev2)

class TestPullHook(per_branch.TestCaseWithBranch):

    def setUp(self):

        self.hook_calls = []

        super(TestPullHook, self).setUp()

    def capture_post_pull_hook(self, result):

        """Capture post pull hook calls to self.hook_calls.

        The call is logged, as is some state of the two branches.

        """

        if result.local_branch:

            local_locked = result.local_branch.is_locked()

            local_base = result.local_branch.base

        else:

            local_locked = None

            local_base = None

        self.hook_calls.append(

            ('post_pull', result.source_branch, local_base,

             result.master_branch.base, result.old_revno,

             result.old_revid,

             result.new_revno, result.new_revid,

             result.source_branch.is_locked(), local_locked,

             result.master_branch.is_locked()))

    def test_post_pull_empty_history(self):

        target = self.make_branch('target')

        source = self.make_branch('source')

        branch.Branch.hooks.install_named_hook(

            'post_pull', self.capture_post_pull_hook, None)

        target.pull(source)

        # with nothing there we should still get a notification, and

        # have both branches locked at the notification time.

        self.assertEqual([

            ('post_pull', source, None, target.base, 0, revision.NULL_REVISION,

             0, revision.NULL_REVISION, True, None, True)

            ],

            self.hook_calls)

    def test_post_pull_bound_branch(self):

        # pulling to a bound branch should pass in the master branch to the

        # hook, allowing the correct number of emails to be sent, while still

        # allowing hooks that want to modify the target to do so to both

        # instances.

        target = self.make_branch('target')

        local = self.make_branch('local')

        try:

            local.bind(target)

        except errors.UpgradeRequired:

            # We can't bind this format to itself- typically it is the local

            # branch that doesn't support binding.  As of May 2007

            # remotebranches can't be bound.  Let's instead make a new local

            # branch of the default type, which does allow binding.

            # See https://bugs.launchpad.net/bzr/+bug/112020

            local = controldir.ControlDir.create_branch_convenience('local2')

            local.bind(target)

        source = self.make_branch('source')

        branch.Branch.hooks.install_named_hook(

            'post_pull', self.capture_post_pull_hook, None)

        local.pull(source)

        # with nothing there we should still get a notification, and

        # have both branches locked at the notification time.

        self.assertEqual([

            ('post_pull', source, local.base, target.base, 0,

             revision.NULL_REVISION, 0, revision.NULL_REVISION,

             True, True, True)

            ],

            self.hook_calls)

    def test_post_pull_nonempty_history(self):

        target = self.make_branch_and_memory_tree('target')

        target.lock_write()

        target.add('')

        rev1 = target.commit('rev 1')

        target.unlock()

        sourcedir = target.controldir.clone(self.get_url('source'))

        source = sourcedir.open_branch().create_memorytree()

        rev2 = source.commit('rev 2')

        branch.Branch.hooks.install_named_hook(

            'post_pull', self.capture_post_pull_hook, None)

        target.branch.pull(source.branch)

        # with nothing there we should still get a notification, and

        # have both branches locked at the notification time.

        self.assertEqual([

            ('post_pull', source.branch, None, target.branch.base, 1, rev1,

             2, rev2, True, None, True)

            ],

            self.hook_calls)