1
# Copyright (C) 2005, 2007, 2008 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
18
"""Black-box tests for bzr push."""
28
from bzrlib.branch import Branch
29
from bzrlib.bzrdir import BzrDirMetaFormat1
30
from bzrlib.osutils import abspath
31
from bzrlib.repofmt.knitrepo import RepositoryFormatKnit1
32
from bzrlib.tests.blackbox import ExternalBase
33
from bzrlib.tests.http_server import HttpServer
34
from bzrlib.transport.memory import MemoryServer, MemoryTransport
35
from bzrlib.uncommit import uncommit
36
from bzrlib.urlutils import local_path_from_url
37
from bzrlib.workingtree import WorkingTree
40
class TestPush(ExternalBase):
42
def test_push_error_on_vfs_http(self):
43
""" pushing a branch to a HTTP server fails cleanly. """
44
# the trunk is published on a web server
45
self.transport_readonly_server = HttpServer
46
self.make_branch('source')
47
public_url = self.get_readonly_url('target')
48
self.run_bzr_error(['http does not support mkdir'],
52
def test_push_remember(self):
53
"""Push changes from one branch to another and test push location."""
54
transport = self.get_transport()
55
tree_a = self.make_branch_and_tree('branch_a')
56
branch_a = tree_a.branch
57
self.build_tree(['branch_a/a'])
59
tree_a.commit('commit a')
60
tree_b = branch_a.bzrdir.sprout('branch_b').open_workingtree()
61
branch_b = tree_b.branch
62
tree_c = branch_a.bzrdir.sprout('branch_c').open_workingtree()
63
branch_c = tree_c.branch
64
self.build_tree(['branch_a/b'])
66
tree_a.commit('commit b')
67
self.build_tree(['branch_b/c'])
69
tree_b.commit('commit c')
70
# initial push location must be empty
71
self.assertEqual(None, branch_b.get_push_location())
73
# test push for failure without push location set
75
out = self.run_bzr('push', retcode=3)
76
self.assertEquals(out,
77
('','bzr: ERROR: No push location known or specified.\n'))
79
# test not remembered if cannot actually push
80
self.run_bzr('push ../path/which/doesnt/exist', retcode=3)
81
out = self.run_bzr('push', retcode=3)
83
('', 'bzr: ERROR: No push location known or specified.\n'),
86
# test implicit --remember when no push location set, push fails
87
out = self.run_bzr('push ../branch_b', retcode=3)
88
self.assertEquals(out,
89
('','bzr: ERROR: These branches have diverged. '
90
'Try using "merge" and then "push".\n'))
91
self.assertEquals(abspath(branch_a.get_push_location()),
92
abspath(branch_b.bzrdir.root_transport.base))
94
# test implicit --remember after resolving previous failure
95
uncommit(branch=branch_b, tree=tree_b)
96
transport.delete('branch_b/c')
97
out, err = self.run_bzr('push')
98
path = branch_a.get_push_location()
99
self.assertEquals(out,
100
'Using saved push location: %s\n'
101
% local_path_from_url(path))
102
self.assertEqual(err,
103
'All changes applied successfully.\n'
104
'Pushed up to revision 2.\n')
105
self.assertEqual(path,
106
branch_b.bzrdir.root_transport.base)
107
# test explicit --remember
108
self.run_bzr('push ../branch_c --remember')
109
self.assertEquals(branch_a.get_push_location(),
110
branch_c.bzrdir.root_transport.base)
112
def test_push_without_tree(self):
113
# bzr push from a branch that does not have a checkout should work.
114
b = self.make_branch('.')
115
out, err = self.run_bzr('push pushed-location')
116
self.assertEqual('', out)
117
self.assertEqual('Created new branch.\n', err)
118
b2 = Branch.open('pushed-location')
119
self.assertEndsWith(b2.base, 'pushed-location/')
121
def test_push_new_branch_revision_count(self):
122
# bzr push of a branch with revisions to a new location
123
# should print the number of revisions equal to the length of the
125
t = self.make_branch_and_tree('tree')
126
self.build_tree(['tree/file'])
130
out, err = self.run_bzr('push pushed-to')
132
self.assertEqual('', out)
133
self.assertEqual('Created new branch.\n', err)
135
def test_push_only_pushes_history(self):
136
# Knit branches should only push the history for the current revision.
137
format = BzrDirMetaFormat1()
138
format.repository_format = RepositoryFormatKnit1()
139
shared_repo = self.make_repository('repo', format=format, shared=True)
140
shared_repo.set_make_working_trees(True)
142
def make_shared_tree(path):
143
shared_repo.bzrdir.root_transport.mkdir(path)
144
shared_repo.bzrdir.create_branch_convenience('repo/' + path)
145
return WorkingTree.open('repo/' + path)
146
tree_a = make_shared_tree('a')
147
self.build_tree(['repo/a/file'])
149
tree_a.commit('commit a-1', rev_id='a-1')
150
f = open('repo/a/file', 'ab')
151
f.write('more stuff\n')
153
tree_a.commit('commit a-2', rev_id='a-2')
155
tree_b = make_shared_tree('b')
156
self.build_tree(['repo/b/file'])
158
tree_b.commit('commit b-1', rev_id='b-1')
160
self.assertTrue(shared_repo.has_revision('a-1'))
161
self.assertTrue(shared_repo.has_revision('a-2'))
162
self.assertTrue(shared_repo.has_revision('b-1'))
164
# Now that we have a repository with shared files, make sure
165
# that things aren't copied out by a 'push'
167
self.run_bzr('push ../../push-b')
168
pushed_tree = WorkingTree.open('../../push-b')
169
pushed_repo = pushed_tree.branch.repository
170
self.assertFalse(pushed_repo.has_revision('a-1'))
171
self.assertFalse(pushed_repo.has_revision('a-2'))
172
self.assertTrue(pushed_repo.has_revision('b-1'))
174
def test_push_funky_id(self):
175
t = self.make_branch_and_tree('tree')
177
self.build_tree(['filename'])
178
t.add('filename', 'funky-chars<>%&;"\'')
179
t.commit('commit filename')
180
self.run_bzr('push ../new-tree')
182
def test_push_dash_d(self):
183
t = self.make_branch_and_tree('from')
184
t.commit(allow_pointless=True,
185
message='first commit')
186
self.run_bzr('push -d from to-one')
187
self.failUnlessExists('to-one')
188
self.run_bzr('push -d %s %s'
189
% tuple(map(urlutils.local_path_to_url, ['from', 'to-two'])))
190
self.failUnlessExists('to-two')
192
def test_push_smart_non_stacked_streaming_acceptance(self):
193
self.setup_smart_server_with_call_log()
194
t = self.make_branch_and_tree('from')
195
t.commit(allow_pointless=True, message='first commit')
196
self.reset_smart_call_log()
197
self.run_bzr(['push', self.get_url('to-one')], working_dir='from')
198
# This figure represent the amount of work to perform this use case. It
199
# is entirely ok to reduce this number if a test fails due to rpc_count
200
# being too low. If rpc_count increases, more network roundtrips have
201
# become necessary for this use case. Please do not adjust this number
202
# upwards without agreement from bzr's network support maintainers.
203
self.assertLength(9, self.hpss_calls)
205
def test_push_smart_stacked_streaming_acceptance(self):
206
self.setup_smart_server_with_call_log()
207
parent = self.make_branch_and_tree('parent', format='1.9')
208
parent.commit(message='first commit')
209
local = parent.bzrdir.sprout('local').open_workingtree()
210
local.commit(message='local commit')
211
self.reset_smart_call_log()
212
self.run_bzr(['push', '--stacked', '--stacked-on', '../parent',
213
self.get_url('public')], working_dir='local')
214
# This figure represent the amount of work to perform this use case. It
215
# is entirely ok to reduce this number if a test fails due to rpc_count
216
# being too low. If rpc_count increases, more network roundtrips have
217
# become necessary for this use case. Please do not adjust this number
218
# upwards without agreement from bzr's network support maintainers.
219
self.assertLength(14, self.hpss_calls)
220
remote = Branch.open('public')
221
self.assertEndsWith(remote.get_stacked_on_url(), '/parent')
223
def test_push_smart_with_default_stacking_url_path_segment(self):
224
# If the default stacked-on location is a path element then branches
225
# we push there over the smart server are stacked and their
226
# stacked_on_url is that exact path segment. Added to nail bug 385132.
227
self.setup_smart_server_with_call_log()
228
self.make_branch('stack-on', format='1.9')
229
self.make_bzrdir('.').get_config().set_default_stack_on('/stack-on')
230
self.make_branch('from', format='1.9')
231
self.reset_smart_call_log()
232
out, err = self.run_bzr(['push', '-d', 'from', self.get_url('to')])
233
branch = Branch.open(self.get_url('to'))
234
self.assertEqual('/stack-on', branch.get_stacked_on_url())
236
def test_push_smart_with_default_stacking_relative_path(self):
237
# If the default stacked-on location is a relative path then branches
238
# we push there over the smart server are stacked and their
239
# stacked_on_url is a relative path. Added to nail bug 385132.
240
self.setup_smart_server_with_call_log()
241
self.make_branch('stack-on', format='1.9')
242
self.make_bzrdir('.').get_config().set_default_stack_on('stack-on')
243
self.make_branch('from', format='1.9')
244
self.reset_smart_call_log()
245
out, err = self.run_bzr(['push', '-d', 'from', self.get_url('to')])
246
branch = Branch.open(self.get_url('to'))
247
self.assertEqual('../stack-on', branch.get_stacked_on_url())
249
def create_simple_tree(self):
250
tree = self.make_branch_and_tree('tree')
251
self.build_tree(['tree/a'])
252
tree.add(['a'], ['a-id'])
253
tree.commit('one', rev_id='r1')
256
def test_push_create_prefix(self):
257
"""'bzr push --create-prefix' will create leading directories."""
258
tree = self.create_simple_tree()
260
self.run_bzr_error(['Parent directory of ../new/tree does not exist'],
263
self.run_bzr('push ../new/tree --create-prefix',
265
new_tree = WorkingTree.open('new/tree')
266
self.assertEqual(tree.last_revision(), new_tree.last_revision())
267
self.failUnlessExists('new/tree/a')
269
def test_push_use_existing(self):
270
"""'bzr push --use-existing-dir' can push into an existing dir.
272
By default, 'bzr push' will not use an existing, non-versioned dir.
274
tree = self.create_simple_tree()
275
self.build_tree(['target/'])
277
self.run_bzr_error(['Target directory ../target already exists',
278
'Supply --use-existing-dir',
280
'push ../target', working_dir='tree')
282
self.run_bzr('push --use-existing-dir ../target',
285
new_tree = WorkingTree.open('target')
286
self.assertEqual(tree.last_revision(), new_tree.last_revision())
287
# The push should have created target/a
288
self.failUnlessExists('target/a')
290
def test_push_onto_repo(self):
291
"""We should be able to 'bzr push' into an existing bzrdir."""
292
tree = self.create_simple_tree()
293
repo = self.make_repository('repo', shared=True)
295
self.run_bzr('push ../repo',
298
# Pushing onto an existing bzrdir will create a repository and
299
# branch as needed, but will only create a working tree if there was
301
self.assertRaises(errors.NoWorkingTree, WorkingTree.open, 'repo')
302
new_branch = Branch.open('repo')
303
self.assertEqual(tree.last_revision(), new_branch.last_revision())
305
def test_push_onto_just_bzrdir(self):
306
"""We don't handle when the target is just a bzrdir.
308
Because you shouldn't be able to create *just* a bzrdir in the wild.
310
# TODO: jam 20070109 Maybe it would be better to create the repository
312
tree = self.create_simple_tree()
313
a_bzrdir = self.make_bzrdir('dir')
315
self.run_bzr_error(['At ../dir you have a valid .bzr control'],
319
def test_push_with_revisionspec(self):
320
"""We should be able to push a revision older than the tip."""
321
tree_from = self.make_branch_and_tree('from')
322
tree_from.commit("One.", rev_id="from-1")
323
tree_from.commit("Two.", rev_id="from-2")
325
self.run_bzr('push -r1 ../to', working_dir='from')
327
tree_to = WorkingTree.open('to')
328
repo_to = tree_to.branch.repository
329
self.assertTrue(repo_to.has_revision('from-1'))
330
self.assertFalse(repo_to.has_revision('from-2'))
331
self.assertEqual(tree_to.branch.last_revision_info()[1], 'from-1')
334
['bzr: ERROR: bzr push --revision '
335
'takes exactly one revision identifier\n'],
336
'push -r0..2 ../to', working_dir='from')
338
def create_trunk_and_feature_branch(self):
340
trunk_tree = self.make_branch_and_tree('target',
342
trunk_tree.commit('mainline')
343
# and a branch from it
344
branch_tree = self.make_branch_and_tree('branch',
346
branch_tree.pull(trunk_tree.branch)
347
branch_tree.branch.set_parent(trunk_tree.branch.base)
348
# with some work on it
349
branch_tree.commit('moar work plz')
350
return trunk_tree, branch_tree
352
def assertPublished(self, branch_revid, stacked_on):
353
"""Assert that the branch 'published' has been published correctly."""
354
published_branch = Branch.open('published')
355
# The published branch refers to the mainline
356
self.assertEqual(stacked_on, published_branch.get_stacked_on_url())
357
# and the branch's work was pushed
358
self.assertTrue(published_branch.repository.has_revision(branch_revid))
360
def test_push_new_branch_stacked_on(self):
361
"""Pushing a new branch with --stacked-on creates a stacked branch."""
362
trunk_tree, branch_tree = self.create_trunk_and_feature_branch()
363
# we publish branch_tree with a reference to the mainline.
364
out, err = self.run_bzr(['push', '--stacked-on', trunk_tree.branch.base,
365
self.get_url('published')], working_dir='branch')
366
self.assertEqual('', out)
367
self.assertEqual('Created new stacked branch referring to %s.\n' %
368
trunk_tree.branch.base, err)
369
self.assertPublished(branch_tree.last_revision(),
370
trunk_tree.branch.base)
372
def test_push_new_branch_stacked_uses_parent_when_no_public_url(self):
373
"""When the parent has no public url the parent is used as-is."""
374
trunk_tree, branch_tree = self.create_trunk_and_feature_branch()
375
# now we do a stacked push, which should determine the public location
377
out, err = self.run_bzr(['push', '--stacked',
378
self.get_url('published')], working_dir='branch')
379
self.assertEqual('', out)
380
self.assertEqual('Created new stacked branch referring to %s.\n' %
381
trunk_tree.branch.base, err)
382
self.assertPublished(branch_tree.last_revision(), trunk_tree.branch.base)
384
def test_push_new_branch_stacked_uses_parent_public(self):
385
"""Pushing a new branch with --stacked creates a stacked branch."""
386
trunk_tree, branch_tree = self.create_trunk_and_feature_branch()
387
# the trunk is published on a web server
388
self.transport_readonly_server = HttpServer
389
trunk_public = self.make_branch('public_trunk', format='1.9')
390
trunk_public.pull(trunk_tree.branch)
391
trunk_public_url = self.get_readonly_url('public_trunk')
392
trunk_tree.branch.set_public_branch(trunk_public_url)
393
# now we do a stacked push, which should determine the public location
395
out, err = self.run_bzr(['push', '--stacked',
396
self.get_url('published')], working_dir='branch')
397
self.assertEqual('', out)
398
self.assertEqual('Created new stacked branch referring to %s.\n' %
399
trunk_public_url, err)
400
self.assertPublished(branch_tree.last_revision(), trunk_public_url)
402
def test_push_new_branch_stacked_no_parent(self):
403
"""Pushing with --stacked and no parent branch errors."""
404
branch = self.make_branch_and_tree('branch', format='1.9')
405
# now we do a stacked push, which should fail as the place to refer too
406
# cannot be determined.
407
out, err = self.run_bzr_error(
408
['Could not determine branch to refer to\\.'], ['push', '--stacked',
409
self.get_url('published')], working_dir='branch')
410
self.assertEqual('', out)
411
self.assertFalse(self.get_transport('published').has('.'))
413
def test_push_notifies_default_stacking(self):
414
self.make_branch('stack_on', format='1.6')
415
self.make_bzrdir('.').get_config().set_default_stack_on('stack_on')
416
self.make_branch('from', format='1.6')
417
out, err = self.run_bzr('push -d from to')
418
self.assertContainsRe(err,
419
'Using default stacking branch stack_on at .*')
421
def test_push_stacks_with_default_stacking_if_target_is_stackable(self):
422
self.make_branch('stack_on', format='1.6')
423
self.make_bzrdir('.').get_config().set_default_stack_on('stack_on')
424
self.make_branch('from', format='pack-0.92')
425
out, err = self.run_bzr('push -d from to')
426
branch = Branch.open('to')
427
self.assertEqual('../stack_on', branch.get_stacked_on_url())
429
def test_push_does_not_change_format_with_default_if_target_cannot(self):
430
self.make_branch('stack_on', format='pack-0.92')
431
self.make_bzrdir('.').get_config().set_default_stack_on('stack_on')
432
self.make_branch('from', format='pack-0.92')
433
out, err = self.run_bzr('push -d from to')
434
branch = Branch.open('to')
435
self.assertRaises(errors.UnstackableBranchFormat,
436
branch.get_stacked_on_url)
438
def test_push_doesnt_create_broken_branch(self):
439
"""Pushing a new standalone branch works even when there's a default
440
stacking policy at the destination.
442
The new branch will preserve the repo format (even if it isn't the
443
default for the branch), and will be stacked when the repo format
444
allows (which means that the branch format isn't necessarly preserved).
446
self.make_repository('repo', shared=True, format='1.6')
447
builder = self.make_branch_builder('repo/local', format='pack-0.92')
448
builder.start_series()
449
builder.build_snapshot('rev-1', None, [
450
('add', ('', 'root-id', 'directory', '')),
451
('add', ('filename', 'f-id', 'file', 'content\n'))])
452
builder.build_snapshot('rev-2', ['rev-1'], [])
453
builder.build_snapshot('rev-3', ['rev-2'],
454
[('modify', ('f-id', 'new-content\n'))])
455
builder.finish_series()
456
branch = builder.get_branch()
457
# Push rev-1 to "trunk", so that we can stack on it.
458
self.run_bzr('push -d repo/local trunk -r 1')
459
# Set a default stacking policy so that new branches will automatically
461
self.make_bzrdir('.').get_config().set_default_stack_on('trunk')
462
# Push rev-2 to a new branch "remote". It will be stacked on "trunk".
463
out, err = self.run_bzr('push -d repo/local remote -r 2')
464
self.assertContainsRe(
465
err, 'Using default stacking branch trunk at .*')
466
# Push rev-3 onto "remote". If "remote" not stacked and is missing the
467
# fulltext record for f-id @ rev-1, then this will fail.
468
out, err = self.run_bzr('push -d repo/local remote -r 3')
470
def test_push_verbose_shows_log(self):
471
tree = self.make_branch_and_tree('source')
473
out, err = self.run_bzr('push -v -d source target')
474
# initial push contains log
475
self.assertContainsRe(out, 'rev1')
477
out, err = self.run_bzr('push -v -d source target')
478
# subsequent push contains log
479
self.assertContainsRe(out, 'rev2')
480
# subsequent log is accurate
481
self.assertNotContainsRe(out, 'rev1')
484
class RedirectingMemoryTransport(MemoryTransport):
486
def mkdir(self, relpath, mode=None):
487
from bzrlib.trace import mutter
488
mutter('cwd: %r, rel: %r, abs: %r' % (self._cwd, relpath, abspath))
489
if self._cwd == '/source/':
490
raise errors.RedirectRequested(self.abspath(relpath),
491
self.abspath('../target'),
493
elif self._cwd == '/infinite-loop/':
494
raise errors.RedirectRequested(self.abspath(relpath),
495
self.abspath('../infinite-loop'),
498
return super(RedirectingMemoryTransport, self).mkdir(
501
def _redirected_to(self, source, target):
502
# We do accept redirections
503
return transport.get_transport(target)
506
class RedirectingMemoryServer(MemoryServer):
509
self._dirs = {'/': None}
512
self._scheme = 'redirecting-memory+%s:///' % id(self)
513
transport.register_transport(self._scheme, self._memory_factory)
515
def _memory_factory(self, url):
516
result = RedirectingMemoryTransport(url)
517
result._dirs = self._dirs
518
result._files = self._files
519
result._locks = self._locks
523
transport.unregister_transport(self._scheme, self._memory_factory)
526
class TestPushRedirect(ExternalBase):
529
ExternalBase.setUp(self)
530
self.memory_server = RedirectingMemoryServer()
531
self.memory_server.setUp()
532
self.addCleanup(self.memory_server.tearDown)
534
# Make the branch and tree that we'll be pushing.
535
t = self.make_branch_and_tree('tree')
536
self.build_tree(['tree/file'])
540
def test_push_redirects_on_mkdir(self):
541
"""If the push requires a mkdir, push respects redirect requests.
543
This is added primarily to handle lp:/ URI support, so that users can
544
push to new branches by specifying lp:/ URIs.
546
destination_url = self.memory_server.get_url() + 'source'
547
self.run_bzr(['push', '-d', 'tree', destination_url])
549
local_revision = Branch.open('tree').last_revision()
550
remote_revision = Branch.open(
551
self.memory_server.get_url() + 'target').last_revision()
552
self.assertEqual(remote_revision, local_revision)
554
def test_push_gracefully_handles_too_many_redirects(self):
555
"""Push fails gracefully if the mkdir generates a large number of
558
destination_url = self.memory_server.get_url() + 'infinite-loop'
559
out, err = self.run_bzr_error(
560
['Too many redirections trying to make %s\\.\n'
561
% re.escape(destination_url)],
562
['push', '-d', 'tree', destination_url], retcode=3)
563
self.assertEqual('', out)