1
# Copyright (C) 2009, 2010 Canonical Ltd
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.
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.
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., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
17
"""Tests for branch.push behaviour."""
19
from cStringIO import StringIO
33
from bzrlib.branch import Branch
34
from bzrlib.bzrdir import BzrDir
35
from bzrlib.memorytree import MemoryTree
36
from bzrlib.revision import NULL_REVISION
37
from bzrlib.smart import client, server
38
from bzrlib.smart.repository import SmartServerRepositoryGetParentMap
39
from bzrlib.tests.per_interbranch import (
40
TestCaseWithInterBranch,
42
from bzrlib.transport import get_transport
43
from bzrlib.tests import test_server
46
# These tests are based on similar tests in
47
# bzrlib.tests.per_branch.test_push.
50
class TestPush(TestCaseWithInterBranch):
52
def test_push_convergence_simple(self):
53
# when revisions are pushed, the left-most accessible parents must
54
# become the revision-history.
55
mine = self.make_from_branch_and_tree('mine')
56
mine.commit('1st post', rev_id='P1', allow_pointless=True)
57
other = self.sprout_to(mine.bzrdir, 'other').open_workingtree()
58
other.commit('my change', rev_id='M1', allow_pointless=True)
59
mine.merge_from_branch(other.branch)
60
mine.commit('merge my change', rev_id='P2')
61
result = mine.branch.push(other.branch)
62
self.assertEqual(['P1', 'P2'], other.branch.revision_history())
63
# result object contains some structured data
64
self.assertEqual(result.old_revid, 'M1')
65
self.assertEqual(result.new_revid, 'P2')
66
# and it can be treated as an integer for compatibility
67
self.assertEqual(int(result), 0)
69
def test_push_merged_indirect(self):
70
# it should be possible to do a push from one branch into another
71
# when the tip of the target was merged into the source branch
72
# via a third branch - so its buried in the ancestry and is not
73
# directly accessible.
74
mine = self.make_from_branch_and_tree('mine')
75
mine.commit('1st post', rev_id='P1', allow_pointless=True)
76
target = self.sprout_to(mine.bzrdir, 'target').open_workingtree()
77
target.commit('my change', rev_id='M1', allow_pointless=True)
78
other = self.sprout_to(mine.bzrdir, 'other').open_workingtree()
79
other.merge_from_branch(target.branch)
80
other.commit('merge my change', rev_id='O2')
81
mine.merge_from_branch(other.branch)
82
mine.commit('merge other', rev_id='P2')
83
mine.branch.push(target.branch)
84
self.assertEqual(['P1', 'P2'], target.branch.revision_history())
86
def test_push_to_checkout_updates_master(self):
87
"""Pushing into a checkout updates the checkout and the master branch"""
88
master_tree = self.make_to_branch_and_tree('master')
89
checkout = self.make_to_branch_and_tree('checkout')
91
checkout.branch.bind(master_tree.branch)
92
except errors.UpgradeRequired:
93
# cant bind this format, the test is irrelevant.
95
rev1 = checkout.commit('master')
97
other_bzrdir = self.sprout_from(master_tree.branch.bzrdir, 'other')
98
other = other_bzrdir.open_workingtree()
99
rev2 = other.commit('other commit')
100
# now push, which should update both checkout and master.
101
other.branch.push(checkout.branch)
102
self.assertEqual([rev1, rev2], checkout.branch.revision_history())
103
self.assertEqual([rev1, rev2], master_tree.branch.revision_history())
105
def test_push_raises_specific_error_on_master_connection_error(self):
106
master_tree = self.make_to_branch_and_tree('master')
107
checkout = self.make_to_branch_and_tree('checkout')
109
checkout.branch.bind(master_tree.branch)
110
except errors.UpgradeRequired:
111
# cant bind this format, the test is irrelevant.
113
other_bzrdir = self.sprout_from(master_tree.branch.bzrdir, 'other')
114
other = other_bzrdir.open_workingtree()
115
# move the branch out of the way on disk to cause a connection
117
os.rename('master', 'master_gone')
118
# try to push, which should raise a BoundBranchConnectionFailure.
119
self.assertRaises(errors.BoundBranchConnectionFailure,
120
other.branch.push, checkout.branch)
122
def test_push_uses_read_lock(self):
123
"""Push should only need a read lock on the source side."""
124
source = self.make_from_branch_and_tree('source')
125
target = self.make_to_branch('target')
127
self.build_tree(['source/a'])
131
source.branch.lock_read()
135
source.branch.push(target, stop_revision=source.last_revision())
139
source.branch.unlock()
141
def test_push_within_repository(self):
142
"""Push from one branch to another inside the same repository."""
144
repo = self.make_repository('repo', shared=True)
145
except (errors.IncompatibleFormat, errors.UninitializableFormat):
146
# This Branch format cannot create shared repositories
148
# This is a little bit trickier because make_branch_and_tree will not
149
# re-use a shared repository.
151
a_branch = self.make_from_branch('repo/tree')
152
except (errors.UninitializableFormat):
153
# Cannot create these branches
156
tree = a_branch.bzrdir.create_workingtree()
157
except errors.NotLocalUrl:
158
if self.vfs_transport_factory is test_server.LocalURLServer:
159
# the branch is colocated on disk, we cannot create a checkout.
160
# hopefully callers will expect this.
161
local_controldir = bzrdir.BzrDir.open(self.get_vfs_only_url('repo/tree'))
162
tree = local_controldir.create_workingtree()
164
tree = a_branch.create_checkout('repo/tree', lightweight=True)
165
self.build_tree(['repo/tree/a'])
169
to_branch = self.make_to_branch('repo/branch')
170
tree.branch.push(to_branch)
172
self.assertEqual(tree.branch.last_revision(),
173
to_branch.last_revision())
175
def test_push_overwrite_of_non_tip_with_stop_revision(self):
176
"""Combining the stop_revision and overwrite options works.
178
This was <https://bugs.launchpad.net/bzr/+bug/234229>.
180
source = self.make_from_branch_and_tree('source')
181
target = self.make_to_branch('target')
183
source.commit('1st commit')
184
source.branch.push(target)
185
source.commit('2nd commit', rev_id='rev-2')
186
source.commit('3rd commit')
188
source.branch.push(target, stop_revision='rev-2', overwrite=True)
189
self.assertEqual('rev-2', target.last_revision())
191
def test_push_with_default_stacking_does_not_create_broken_branch(self):
192
"""Pushing a new standalone branch works even when there's a default
193
stacking policy at the destination.
195
The new branch will preserve the repo format (even if it isn't the
196
default for the branch), and will be stacked when the repo format
197
allows (which means that the branch format isn't necessarly preserved).
199
if isinstance(self.branch_format_from, branch.BzrBranchFormat4):
200
raise tests.TestNotApplicable('Not a metadir format.')
201
if isinstance(self.branch_format_from, branch.BranchReferenceFormat):
202
# This test could in principle apply to BranchReferenceFormat, but
203
# make_branch_builder doesn't support it.
204
raise tests.TestSkipped(
205
"BranchBuilder can't make reference branches.")
206
# Make a branch called "local" in a stackable repository
207
# The branch has 3 revisions:
208
# - rev-1, adds a file
209
# - rev-2, no changes
210
# - rev-3, modifies the file.
211
repo = self.make_repository('repo', shared=True, format='1.6')
212
builder = self.make_from_branch_builder('repo/local')
213
builder.start_series()
214
builder.build_snapshot('rev-1', None, [
215
('add', ('', 'root-id', 'directory', '')),
216
('add', ('filename', 'f-id', 'file', 'content\n'))])
217
builder.build_snapshot('rev-2', ['rev-1'], [])
218
builder.build_snapshot('rev-3', ['rev-2'],
219
[('modify', ('f-id', 'new-content\n'))])
220
builder.finish_series()
221
trunk = builder.get_branch()
222
# Sprout rev-1 to "trunk", so that we can stack on it.
223
trunk.bzrdir.sprout(self.get_url('trunk'), revision_id='rev-1')
224
# Set a default stacking policy so that new branches will automatically
226
self.make_bzrdir('.').get_config().set_default_stack_on('trunk')
227
# Push rev-2 to a new branch "remote". It will be stacked on "trunk".
229
push._show_push_branch(trunk, 'rev-2', self.get_url('remote'), output)
230
# Push rev-3 onto "remote". If "remote" not stacked and is missing the
231
# fulltext record for f-id @ rev-1, then this will fail.
232
remote_branch = Branch.open(self.get_url('remote'))
233
trunk.push(remote_branch)
234
check.check_dwim(remote_branch.base, False, True, True)
236
def test_no_get_parent_map_after_insert_stream(self):
237
# Effort test for bug 331823
238
self.setup_smart_server_with_call_log()
239
# Make a local branch with four revisions. Four revisions because:
240
# one to push, one there for _walk_to_common_revisions to find, one we
241
# don't want to access, one for luck :)
242
if isinstance(self.branch_format_from, branch.BranchReferenceFormat):
243
# This test could in principle apply to BranchReferenceFormat, but
244
# make_branch_builder doesn't support it.
245
raise tests.TestSkipped(
246
"BranchBuilder can't make reference branches.")
248
builder = self.make_from_branch_builder('local')
249
except (errors.TransportNotPossible, errors.UninitializableFormat):
250
raise tests.TestNotApplicable('format not directly constructable')
251
builder.start_series()
252
builder.build_snapshot('first', None, [
253
('add', ('', 'root-id', 'directory', ''))])
254
builder.build_snapshot('second', ['first'], [])
255
builder.build_snapshot('third', ['second'], [])
256
builder.build_snapshot('fourth', ['third'], [])
257
builder.finish_series()
258
local = branch.Branch.open(self.get_vfs_only_url('local'))
259
# Initial push of three revisions
260
remote_bzrdir = local.bzrdir.sprout(
261
self.get_url('remote'), revision_id='third')
262
remote = remote_bzrdir.open_branch()
263
# Push fourth revision
264
self.reset_smart_call_log()
265
self.disableOptimisticGetParentMap()
266
self.assertFalse(local.is_locked())
268
hpss_call_names = [item.call.method for item in self.hpss_calls]
269
self.assertTrue('Repository.insert_stream_1.19' in hpss_call_names)
270
insert_stream_idx = hpss_call_names.index(
271
'Repository.insert_stream_1.19')
272
calls_after_insert_stream = hpss_call_names[insert_stream_idx:]
273
# After inserting the stream the client has no reason to query the
274
# remote graph any further.
276
['Repository.insert_stream_1.19', 'Repository.insert_stream_1.19',
277
'get', 'Branch.set_last_revision_info', 'Branch.unlock'],
278
calls_after_insert_stream)
280
def disableOptimisticGetParentMap(self):
281
# Tweak some class variables to stop remote get_parent_map calls asking
282
# for or receiving more data than the caller asked for.
283
self.overrideAttr(repository.InterRepository,
284
'_walk_to_common_revisions_batch_size', 1)
285
self.overrideAttr(SmartServerRepositoryGetParentMap,
286
'no_extra_results', True)
289
class TestPushHook(TestCaseWithInterBranch):
293
TestCaseWithInterBranch.setUp(self)
295
def capture_post_push_hook(self, result):
296
"""Capture post push hook calls to self.hook_calls.
298
The call is logged, as is some state of the two branches.
300
if result.local_branch:
301
local_locked = result.local_branch.is_locked()
302
local_base = result.local_branch.base
306
self.hook_calls.append(
307
('post_push', result.source_branch, local_base,
308
result.master_branch.base,
309
result.old_revno, result.old_revid,
310
result.new_revno, result.new_revid,
311
result.source_branch.is_locked(), local_locked,
312
result.master_branch.is_locked()))
314
def test_post_push_empty_history(self):
315
target = self.make_to_branch('target')
316
source = self.make_from_branch('source')
317
Branch.hooks.install_named_hook('post_push',
318
self.capture_post_push_hook, None)
320
# with nothing there we should still get a notification, and
321
# have both branches locked at the notification time.
323
('post_push', source, None, target.base, 0, NULL_REVISION,
324
0, NULL_REVISION, True, None, True)
328
def test_post_push_bound_branch(self):
329
# pushing to a bound branch should pass in the master branch to the
330
# hook, allowing the correct number of emails to be sent, while still
331
# allowing hooks that want to modify the target to do so to both
333
target = self.make_to_branch('target')
334
local = self.make_from_branch('local')
337
except errors.UpgradeRequired:
338
# We can't bind this format to itself- typically it is the local
339
# branch that doesn't support binding. As of May 2007
340
# remotebranches can't be bound. Let's instead make a new local
341
# branch of the default type, which does allow binding.
342
# See https://bugs.launchpad.net/bzr/+bug/112020
343
local = BzrDir.create_branch_convenience('local2')
345
source = self.make_from_branch('source')
346
Branch.hooks.install_named_hook('post_push',
347
self.capture_post_push_hook, None)
349
# with nothing there we should still get a notification, and
350
# have both branches locked at the notification time.
352
('post_push', source, local.base, target.base, 0, NULL_REVISION,
353
0, NULL_REVISION, True, True, True)
357
def test_post_push_nonempty_history(self):
358
target = self.make_to_branch_and_tree('target')
361
rev1 = target.commit('rev 1')
363
sourcedir = target.bzrdir.clone(self.get_url('source'))
364
source = MemoryTree.create_on_branch(sourcedir.open_branch())
365
rev2 = source.commit('rev 2')
366
Branch.hooks.install_named_hook('post_push',
367
self.capture_post_push_hook, None)
368
source.branch.push(target.branch)
369
# with nothing there we should still get a notification, and
370
# have both branches locked at the notification time.
372
('post_push', source.branch, None, target.branch.base, 1, rev1,
373
2, rev2, True, None, True)
added
removed
bzrlib/tests/per_interbranch/test_push.py
