~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/branch_implementations/test_push.py

  • Committer: Robert Collins
  • Date: 2007-07-04 08:08:13 UTC
  • mfrom: (2572 +trunk)
  • mto: This revision was merged to the branch mainline in revision 2587.
  • Revision ID: robertc@robertcollins.net-20070704080813-wzebx0r88fvwj5rq
Merge bzr.dev.

Show diffs side-by-side

added added

removed removed

Lines of Context:
 
1
# Copyright (C) 2004, 2005, 2007 Canonical Ltd
 
2
#
 
3
# This program is free software; you can redistribute it and/or modify
 
4
# it under the terms of the GNU General Public License as published by
 
5
# the Free Software Foundation; either version 2 of the License, or
 
6
# (at your option) any later version.
 
7
#
 
8
# This program is distributed in the hope that it will be useful,
 
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
 
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 
11
# GNU General Public License for more details.
 
12
#
 
13
# You should have received a copy of the GNU General Public License
 
14
# along with this program; if not, write to the Free Software
 
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 
16
 
 
17
"""Tests for branch.push behaviour."""
 
18
 
 
19
import os
 
20
 
 
21
from bzrlib import bzrdir, errors
 
22
from bzrlib.branch import Branch
 
23
from bzrlib.bzrdir import BzrDir
 
24
from bzrlib.memorytree import MemoryTree
 
25
from bzrlib.remote import RemoteBranch
 
26
from bzrlib.revision import NULL_REVISION
 
27
from bzrlib.tests import TestSkipped
 
28
from bzrlib.tests.branch_implementations.test_branch import TestCaseWithBranch
 
29
from bzrlib.transport.local import LocalURLServer
 
30
 
 
31
 
 
32
class TestPush(TestCaseWithBranch):
 
33
 
 
34
    def test_push_convergence_simple(self):
 
35
        # when revisions are pushed, the left-most accessible parents must 
 
36
        # become the revision-history.
 
37
        mine = self.make_branch_and_tree('mine')
 
38
        mine.commit('1st post', rev_id='P1', allow_pointless=True)
 
39
        other = mine.bzrdir.sprout('other').open_workingtree()
 
40
        other.commit('my change', rev_id='M1', allow_pointless=True)
 
41
        mine.merge_from_branch(other.branch)
 
42
        mine.commit('merge my change', rev_id='P2')
 
43
        result = mine.branch.push(other.branch)
 
44
        self.assertEqual(['P1', 'P2'], other.branch.revision_history())
 
45
        # result object contains some structured data
 
46
        self.assertEqual(result.old_revid, 'M1')
 
47
        self.assertEqual(result.new_revid, 'P2')
 
48
        # and it can be treated as an integer for compatibility
 
49
        self.assertEqual(int(result), 0)
 
50
 
 
51
    def test_push_merged_indirect(self):
 
52
        # it should be possible to do a push from one branch into another
 
53
        # when the tip of the target was merged into the source branch
 
54
        # via a third branch - so its buried in the ancestry and is not
 
55
        # directly accessible.
 
56
        mine = self.make_branch_and_tree('mine')
 
57
        mine.commit('1st post', rev_id='P1', allow_pointless=True)
 
58
        target = mine.bzrdir.sprout('target').open_workingtree()
 
59
        target.commit('my change', rev_id='M1', allow_pointless=True)
 
60
        other = mine.bzrdir.sprout('other').open_workingtree()
 
61
        other.merge_from_branch(target.branch)
 
62
        other.commit('merge my change', rev_id='O2')
 
63
        mine.merge_from_branch(other.branch)
 
64
        mine.commit('merge other', rev_id='P2')
 
65
        mine.branch.push(target.branch)
 
66
        self.assertEqual(['P1', 'P2'], target.branch.revision_history())
 
67
 
 
68
    def test_push_to_checkout_updates_master(self):
 
69
        """Pushing into a checkout updates the checkout and the master branch"""
 
70
        master_tree = self.make_branch_and_tree('master')
 
71
        checkout = self.make_branch_and_tree('checkout')
 
72
        try:
 
73
            checkout.branch.bind(master_tree.branch)
 
74
        except errors.UpgradeRequired:
 
75
            # cant bind this format, the test is irrelevant.
 
76
            return
 
77
        rev1 = checkout.commit('master')
 
78
 
 
79
        other = master_tree.branch.bzrdir.sprout('other').open_workingtree()
 
80
        rev2 = other.commit('other commit')
 
81
        # now push, which should update both checkout and master.
 
82
        other.branch.push(checkout.branch)
 
83
        self.assertEqual([rev1, rev2], checkout.branch.revision_history())
 
84
        self.assertEqual([rev1, rev2], master_tree.branch.revision_history())
 
85
 
 
86
    def test_push_raises_specific_error_on_master_connection_error(self):
 
87
        master_tree = self.make_branch_and_tree('master')
 
88
        checkout = self.make_branch_and_tree('checkout')
 
89
        try:
 
90
            checkout.branch.bind(master_tree.branch)
 
91
        except errors.UpgradeRequired:
 
92
            # cant bind this format, the test is irrelevant.
 
93
            return
 
94
        other = master_tree.branch.bzrdir.sprout('other').open_workingtree()
 
95
        # move the branch out of the way on disk to cause a connection
 
96
        # error.
 
97
        os.rename('master', 'master_gone')
 
98
        # try to push, which should raise a BoundBranchConnectionFailure.
 
99
        self.assertRaises(errors.BoundBranchConnectionFailure,
 
100
                other.branch.push, checkout.branch)
 
101
 
 
102
    def test_push_uses_read_lock(self):
 
103
        """Push should only need a read lock on the source side."""
 
104
        source = self.make_branch_and_tree('source')
 
105
        target = self.make_branch('target')
 
106
 
 
107
        self.build_tree(['source/a'])
 
108
        source.add(['a'])
 
109
        source.commit('a')
 
110
 
 
111
        source.branch.lock_read()
 
112
        try:
 
113
            target.lock_write()
 
114
            try:
 
115
                source.branch.push(target, stop_revision=source.last_revision())
 
116
            finally:
 
117
                target.unlock()
 
118
        finally:
 
119
            source.branch.unlock()
 
120
 
 
121
    def test_push_within_repository(self):
 
122
        """Push from one branch to another inside the same repository."""
 
123
        try:
 
124
            repo = self.make_repository('repo', shared=True)
 
125
        except (errors.IncompatibleFormat, errors.UninitializableFormat):
 
126
            # This Branch format cannot create shared repositories
 
127
            return
 
128
        # This is a little bit trickier because make_branch_and_tree will not
 
129
        # re-use a shared repository.
 
130
        a_bzrdir = self.make_bzrdir('repo/tree')
 
131
        try:
 
132
            a_branch = self.branch_format.initialize(a_bzrdir)
 
133
        except (errors.UninitializableFormat):
 
134
            # Cannot create these branches
 
135
            return
 
136
        try:
 
137
            tree = a_branch.bzrdir.create_workingtree()
 
138
        except errors.NotLocalUrl:
 
139
            if self.vfs_transport_factory is LocalURLServer:
 
140
                # the branch is colocated on disk, we cannot create a checkout.
 
141
                # hopefully callers will expect this.
 
142
                local_controldir= bzrdir.BzrDir.open(self.get_vfs_only_url('repo/tree'))
 
143
                tree = local_controldir.create_workingtree()
 
144
            else:
 
145
                tree = a_branch.create_checkout('repo/tree', lightweight=True)
 
146
        self.build_tree(['repo/tree/a'])
 
147
        tree.add(['a'])
 
148
        tree.commit('a')
 
149
 
 
150
        to_bzrdir = self.make_bzrdir('repo/branch')
 
151
        to_branch = self.branch_format.initialize(to_bzrdir)
 
152
        tree.branch.push(to_branch)
 
153
 
 
154
        self.assertEqual(tree.branch.last_revision(),
 
155
                         to_branch.last_revision())
 
156
 
 
157
 
 
158
class TestPushHook(TestCaseWithBranch):
 
159
 
 
160
    def setUp(self):
 
161
        self.hook_calls = []
 
162
        TestCaseWithBranch.setUp(self)
 
163
 
 
164
    def capture_post_push_hook(self, result):
 
165
        """Capture post push hook calls to self.hook_calls.
 
166
        
 
167
        The call is logged, as is some state of the two branches.
 
168
        """
 
169
        if result.local_branch:
 
170
            local_locked = result.local_branch.is_locked()
 
171
            local_base = result.local_branch.base
 
172
        else:
 
173
            local_locked = None
 
174
            local_base = None
 
175
        self.hook_calls.append(
 
176
            ('post_push', result.source_branch, local_base,
 
177
             result.master_branch.base,
 
178
             result.old_revno, result.old_revid,
 
179
             result.new_revno, result.new_revid,
 
180
             result.source_branch.is_locked(), local_locked,
 
181
             result.master_branch.is_locked()))
 
182
 
 
183
    def test_post_push_empty_history(self):
 
184
        target = self.make_branch('target')
 
185
        source = self.make_branch('source')
 
186
        Branch.hooks.install_hook('post_push', self.capture_post_push_hook)
 
187
        source.push(target)
 
188
        # with nothing there we should still get a notification, and
 
189
        # have both branches locked at the notification time.
 
190
        self.assertEqual([
 
191
            ('post_push', source, None, target.base, 0, NULL_REVISION,
 
192
             0, NULL_REVISION, True, None, True)
 
193
            ],
 
194
            self.hook_calls)
 
195
 
 
196
    def test_post_push_bound_branch(self):
 
197
        # pushing to a bound branch should pass in the master branch to the
 
198
        # hook, allowing the correct number of emails to be sent, while still
 
199
        # allowing hooks that want to modify the target to do so to both 
 
200
        # instances.
 
201
        target = self.make_branch('target')
 
202
        local = self.make_branch('local')
 
203
        try:
 
204
            local.bind(target)
 
205
        except errors.UpgradeRequired:
 
206
            # We can't bind this format to itself- typically it is the local
 
207
            # branch that doesn't support binding.  As of May 2007
 
208
            # remotebranches can't be bound.  Let's instead make a new local
 
209
            # branch of the default type, which does allow binding.
 
210
            # See https://bugs.launchpad.net/bzr/+bug/112020
 
211
            local = BzrDir.create_branch_convenience('local2')
 
212
            local.bind(target)
 
213
        source = self.make_branch('source')
 
214
        Branch.hooks.install_hook('post_push', self.capture_post_push_hook)
 
215
        source.push(local)
 
216
        # with nothing there we should still get a notification, and
 
217
        # have both branches locked at the notification time.
 
218
        self.assertEqual([
 
219
            ('post_push', source, local.base, target.base, 0, NULL_REVISION,
 
220
             0, NULL_REVISION, True, True, True)
 
221
            ],
 
222
            self.hook_calls)
 
223
 
 
224
    def test_post_push_nonempty_history(self):
 
225
        target = self.make_branch_and_memory_tree('target')
 
226
        target.lock_write()
 
227
        target.add('')
 
228
        rev1 = target.commit('rev 1')
 
229
        target.unlock()
 
230
        sourcedir = target.bzrdir.clone(self.get_url('source'))
 
231
        source = MemoryTree.create_on_branch(sourcedir.open_branch())
 
232
        rev2 = source.commit('rev 2')
 
233
        Branch.hooks.install_hook('post_push', self.capture_post_push_hook)
 
234
        source.branch.push(target.branch)
 
235
        # with nothing there we should still get a notification, and
 
236
        # have both branches locked at the notification time.
 
237
        self.assertEqual([
 
238
            ('post_push', source.branch, None, target.branch.base, 1, rev1,
 
239
             2, rev2, True, None, True)
 
240
            ],
 
241
            self.hook_calls)