/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

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

First attempt to merge .dev and resolve the conflicts (but tests are 
failing)

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