1
# Copyright (C) 2004, 2005, 2007 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., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
17
"""Tests for branch.push behaviour."""
29
from bzrlib.branch import Branch
30
from bzrlib.bzrdir import BzrDir
31
from bzrlib.memorytree import MemoryTree
32
from bzrlib.revision import NULL_REVISION
33
from bzrlib.smart import client, server
34
from bzrlib.tests.branch_implementations.test_branch import TestCaseWithBranch
35
from bzrlib.transport import get_transport
36
from bzrlib.transport.local import LocalURLServer
39
class TestPush(TestCaseWithBranch):
41
def test_push_convergence_simple(self):
42
# when revisions are pushed, the left-most accessible parents must
43
# become the revision-history.
44
mine = self.make_branch_and_tree('mine')
45
mine.commit('1st post', rev_id='P1', allow_pointless=True)
46
other = mine.bzrdir.sprout('other').open_workingtree()
47
other.commit('my change', rev_id='M1', allow_pointless=True)
48
mine.merge_from_branch(other.branch)
49
mine.commit('merge my change', rev_id='P2')
50
result = mine.branch.push(other.branch)
51
self.assertEqual(['P1', 'P2'], other.branch.revision_history())
52
# result object contains some structured data
53
self.assertEqual(result.old_revid, 'M1')
54
self.assertEqual(result.new_revid, 'P2')
55
# and it can be treated as an integer for compatibility
56
self.assertEqual(int(result), 0)
58
def test_push_merged_indirect(self):
59
# it should be possible to do a push from one branch into another
60
# when the tip of the target was merged into the source branch
61
# via a third branch - so its buried in the ancestry and is not
62
# directly accessible.
63
mine = self.make_branch_and_tree('mine')
64
mine.commit('1st post', rev_id='P1', allow_pointless=True)
65
target = mine.bzrdir.sprout('target').open_workingtree()
66
target.commit('my change', rev_id='M1', allow_pointless=True)
67
other = mine.bzrdir.sprout('other').open_workingtree()
68
other.merge_from_branch(target.branch)
69
other.commit('merge my change', rev_id='O2')
70
mine.merge_from_branch(other.branch)
71
mine.commit('merge other', rev_id='P2')
72
mine.branch.push(target.branch)
73
self.assertEqual(['P1', 'P2'], target.branch.revision_history())
75
def test_push_to_checkout_updates_master(self):
76
"""Pushing into a checkout updates the checkout and the master branch"""
77
master_tree = self.make_branch_and_tree('master')
78
checkout = self.make_branch_and_tree('checkout')
80
checkout.branch.bind(master_tree.branch)
81
except errors.UpgradeRequired:
82
# cant bind this format, the test is irrelevant.
84
rev1 = checkout.commit('master')
86
other = master_tree.branch.bzrdir.sprout('other').open_workingtree()
87
rev2 = other.commit('other commit')
88
# now push, which should update both checkout and master.
89
other.branch.push(checkout.branch)
90
self.assertEqual([rev1, rev2], checkout.branch.revision_history())
91
self.assertEqual([rev1, rev2], master_tree.branch.revision_history())
93
def test_push_raises_specific_error_on_master_connection_error(self):
94
master_tree = self.make_branch_and_tree('master')
95
checkout = self.make_branch_and_tree('checkout')
97
checkout.branch.bind(master_tree.branch)
98
except errors.UpgradeRequired:
99
# cant bind this format, the test is irrelevant.
101
other = master_tree.branch.bzrdir.sprout('other').open_workingtree()
102
# move the branch out of the way on disk to cause a connection
104
os.rename('master', 'master_gone')
105
# try to push, which should raise a BoundBranchConnectionFailure.
106
self.assertRaises(errors.BoundBranchConnectionFailure,
107
other.branch.push, checkout.branch)
109
def test_push_uses_read_lock(self):
110
"""Push should only need a read lock on the source side."""
111
source = self.make_branch_and_tree('source')
112
target = self.make_branch('target')
114
self.build_tree(['source/a'])
118
source.branch.lock_read()
122
source.branch.push(target, stop_revision=source.last_revision())
126
source.branch.unlock()
128
def test_push_within_repository(self):
129
"""Push from one branch to another inside the same repository."""
131
repo = self.make_repository('repo', shared=True)
132
except (errors.IncompatibleFormat, errors.UninitializableFormat):
133
# This Branch format cannot create shared repositories
135
# This is a little bit trickier because make_branch_and_tree will not
136
# re-use a shared repository.
137
a_bzrdir = self.make_bzrdir('repo/tree')
139
a_branch = self.branch_format.initialize(a_bzrdir)
140
except (errors.UninitializableFormat):
141
# Cannot create these branches
144
tree = a_branch.bzrdir.create_workingtree()
145
except errors.NotLocalUrl:
146
if self.vfs_transport_factory is LocalURLServer:
147
# the branch is colocated on disk, we cannot create a checkout.
148
# hopefully callers will expect this.
149
local_controldir= bzrdir.BzrDir.open(self.get_vfs_only_url('repo/tree'))
150
tree = local_controldir.create_workingtree()
152
tree = a_branch.create_checkout('repo/tree', lightweight=True)
153
self.build_tree(['repo/tree/a'])
157
to_bzrdir = self.make_bzrdir('repo/branch')
158
to_branch = self.branch_format.initialize(to_bzrdir)
159
tree.branch.push(to_branch)
161
self.assertEqual(tree.branch.last_revision(),
162
to_branch.last_revision())
164
def test_push_overwrite_of_non_tip_with_stop_revision(self):
165
"""Combining the stop_revision and overwrite options works.
167
This was <https://bugs.launchpad.net/bzr/+bug/234229>.
169
source = self.make_branch_and_tree('source')
170
target = self.make_branch('target')
172
source.commit('1st commit')
173
source.branch.push(target)
174
source.commit('2nd commit', rev_id='rev-2')
175
source.commit('3rd commit')
177
source.branch.push(target, stop_revision='rev-2', overwrite=True)
178
self.assertEqual('rev-2', target.last_revision())
181
class TestPushHook(TestCaseWithBranch):
185
TestCaseWithBranch.setUp(self)
187
def capture_post_push_hook(self, result):
188
"""Capture post push hook calls to self.hook_calls.
190
The call is logged, as is some state of the two branches.
192
if result.local_branch:
193
local_locked = result.local_branch.is_locked()
194
local_base = result.local_branch.base
198
self.hook_calls.append(
199
('post_push', result.source_branch, local_base,
200
result.master_branch.base,
201
result.old_revno, result.old_revid,
202
result.new_revno, result.new_revid,
203
result.source_branch.is_locked(), local_locked,
204
result.master_branch.is_locked()))
206
def test_post_push_empty_history(self):
207
target = self.make_branch('target')
208
source = self.make_branch('source')
209
Branch.hooks.install_named_hook('post_push',
210
self.capture_post_push_hook, None)
212
# with nothing there we should still get a notification, and
213
# have both branches locked at the notification time.
215
('post_push', source, None, target.base, 0, NULL_REVISION,
216
0, NULL_REVISION, True, None, True)
220
def test_post_push_bound_branch(self):
221
# pushing to a bound branch should pass in the master branch to the
222
# hook, allowing the correct number of emails to be sent, while still
223
# allowing hooks that want to modify the target to do so to both
225
target = self.make_branch('target')
226
local = self.make_branch('local')
229
except errors.UpgradeRequired:
230
# We can't bind this format to itself- typically it is the local
231
# branch that doesn't support binding. As of May 2007
232
# remotebranches can't be bound. Let's instead make a new local
233
# branch of the default type, which does allow binding.
234
# See https://bugs.launchpad.net/bzr/+bug/112020
235
local = BzrDir.create_branch_convenience('local2')
237
source = self.make_branch('source')
238
Branch.hooks.install_named_hook('post_push',
239
self.capture_post_push_hook, None)
241
# with nothing there we should still get a notification, and
242
# have both branches locked at the notification time.
244
('post_push', source, local.base, target.base, 0, NULL_REVISION,
245
0, NULL_REVISION, True, True, True)
249
def test_post_push_nonempty_history(self):
250
target = self.make_branch_and_memory_tree('target')
253
rev1 = target.commit('rev 1')
255
sourcedir = target.bzrdir.clone(self.get_url('source'))
256
source = MemoryTree.create_on_branch(sourcedir.open_branch())
257
rev2 = source.commit('rev 2')
258
Branch.hooks.install_named_hook('post_push',
259
self.capture_post_push_hook, None)
260
source.branch.push(target.branch)
261
# with nothing there we should still get a notification, and
262
# have both branches locked at the notification time.
264
('post_push', source.branch, None, target.branch.base, 1, rev1,
265
2, rev2, True, None, True)
270
class EmptyPushSmartEffortTests(TestCaseWithBranch):
271
"""Tests that a push of 0 revisions should make a limited number of smart
276
# Skip some scenarios that don't apply to these tests.
277
if (self.transport_server is not None and
278
issubclass(self.transport_server, server.SmartTCPServer)):
279
raise tests.TestNotApplicable(
280
'Does not apply when remote backing branch is also '
282
if isinstance(self.branch_format, branch.BzrBranchFormat4):
283
raise tests.TestNotApplicable(
284
'Branch format 4 is not usable via HPSS.')
285
super(EmptyPushSmartEffortTests, self).setUp()
286
# Create a smart server that publishes whatever the backing VFS server
288
self.smart_server = server.SmartTCPServer_for_testing()
289
self.smart_server.setUp(self.get_server())
290
self.addCleanup(self.smart_server.tearDown)
291
# Make two empty branches, 'empty' and 'target'.
292
self.empty_branch = self.make_branch('empty')
293
self.make_branch('target')
294
# Log all HPSS calls into self.hpss_calls.
295
client._SmartClient.hooks.install_named_hook(
296
'call', self.capture_hpss_call, None)
299
def capture_hpss_call(self, params):
300
self.hpss_calls.append(params.method)
302
def test_empty_branch_api(self):
303
"""The branch_obj.push API should make a limited number of HPSS calls.
305
transport = get_transport(self.smart_server.get_url()).clone('target')
306
target = Branch.open_from_transport(transport)
307
self.empty_branch.push(target)
310
'BzrDir.open_branch',
311
'BzrDir.find_repositoryV2',
312
'Branch.get_stacked_on_url',
314
'Branch.last_revision_info',
318
def test_empty_branch_command(self):
319
"""The 'bzr push' command should make a limited number of HPSS calls.
321
cmd = builtins.cmd_push()
322
cmd.outf = tests.StringIOWrapper()
324
directory=self.get_url() + 'empty',
325
location=self.smart_server.get_url() + 'target')
326
# HPSS calls as of 2008/09/22:
327
# [BzrDir.open, BzrDir.open_branch, BzrDir.find_repositoryV2,
328
# Branch.get_stacked_on_url, get, get, Branch.lock_write,
329
# Branch.last_revision_info, Branch.unlock]
330
self.assertTrue(len(self.hpss_calls) <= 9, self.hpss_calls)
added
removed
bzrlib/tests/branch_implementations/test_push.py
