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.smart import client, server
33
from bzrlib.tests.blackbox import ExternalBase
34
from bzrlib.tests.http_server import HttpServer
35
from bzrlib.transport.memory import MemoryServer, MemoryTransport
36
from bzrlib.uncommit import uncommit
37
from bzrlib.urlutils import local_path_from_url
38
from bzrlib.workingtree import WorkingTree
41
class TestPush(ExternalBase):
43
def test_push_error_on_vfs_http(self):
44
""" pushing a branch to a HTTP server fails cleanly. """
45
# the trunk is published on a web server
46
self.transport_readonly_server = HttpServer
47
self.make_branch('source')
48
public_url = self.get_readonly_url('target')
49
self.run_bzr_error(['http does not support mkdir'],
53
def test_push_remember(self):
54
"""Push changes from one branch to another and test push location."""
55
transport = self.get_transport()
56
tree_a = self.make_branch_and_tree('branch_a')
57
branch_a = tree_a.branch
58
self.build_tree(['branch_a/a'])
60
tree_a.commit('commit a')
61
tree_b = branch_a.bzrdir.sprout('branch_b').open_workingtree()
62
branch_b = tree_b.branch
63
tree_c = branch_a.bzrdir.sprout('branch_c').open_workingtree()
64
branch_c = tree_c.branch
65
self.build_tree(['branch_a/b'])
67
tree_a.commit('commit b')
68
self.build_tree(['branch_b/c'])
70
tree_b.commit('commit c')
71
# initial push location must be empty
72
self.assertEqual(None, branch_b.get_push_location())
74
# test push for failure without push location set
76
out = self.run_bzr('push', retcode=3)
77
self.assertEquals(out,
78
('','bzr: ERROR: No push location known or specified.\n'))
80
# test not remembered if cannot actually push
81
self.run_bzr('push ../path/which/doesnt/exist', retcode=3)
82
out = self.run_bzr('push', retcode=3)
84
('', 'bzr: ERROR: No push location known or specified.\n'),
87
# test implicit --remember when no push location set, push fails
88
out = self.run_bzr('push ../branch_b', retcode=3)
89
self.assertEquals(out,
90
('','bzr: ERROR: These branches have diverged. '
91
'Try using "merge" and then "push".\n'))
92
self.assertEquals(abspath(branch_a.get_push_location()),
93
abspath(branch_b.bzrdir.root_transport.base))
95
# test implicit --remember after resolving previous failure
96
uncommit(branch=branch_b, tree=tree_b)
97
transport.delete('branch_b/c')
98
out, err = self.run_bzr('push')
99
path = branch_a.get_push_location()
100
self.assertEquals(out,
101
'Using saved push location: %s\n'
102
% local_path_from_url(path))
103
self.assertEqual(err,
104
'All changes applied successfully.\n'
105
'Pushed up to revision 2.\n')
106
self.assertEqual(path,
107
branch_b.bzrdir.root_transport.base)
108
# test explicit --remember
109
self.run_bzr('push ../branch_c --remember')
110
self.assertEquals(branch_a.get_push_location(),
111
branch_c.bzrdir.root_transport.base)
113
def test_push_without_tree(self):
114
# bzr push from a branch that does not have a checkout should work.
115
b = self.make_branch('.')
116
out, err = self.run_bzr('push pushed-location')
117
self.assertEqual('', out)
118
self.assertEqual('Created new branch.\n', err)
119
b2 = Branch.open('pushed-location')
120
self.assertEndsWith(b2.base, 'pushed-location/')
122
def test_push_new_branch_revision_count(self):
123
# bzr push of a branch with revisions to a new location
124
# should print the number of revisions equal to the length of the
126
t = self.make_branch_and_tree('tree')
127
self.build_tree(['tree/file'])
131
out, err = self.run_bzr('push pushed-to')
133
self.assertEqual('', out)
134
self.assertEqual('Created new branch.\n', err)
136
def test_push_only_pushes_history(self):
137
# Knit branches should only push the history for the current revision.
138
format = BzrDirMetaFormat1()
139
format.repository_format = RepositoryFormatKnit1()
140
shared_repo = self.make_repository('repo', format=format, shared=True)
141
shared_repo.set_make_working_trees(True)
143
def make_shared_tree(path):
144
shared_repo.bzrdir.root_transport.mkdir(path)
145
shared_repo.bzrdir.create_branch_convenience('repo/' + path)
146
return WorkingTree.open('repo/' + path)
147
tree_a = make_shared_tree('a')
148
self.build_tree(['repo/a/file'])
150
tree_a.commit('commit a-1', rev_id='a-1')
151
f = open('repo/a/file', 'ab')
152
f.write('more stuff\n')
154
tree_a.commit('commit a-2', rev_id='a-2')
156
tree_b = make_shared_tree('b')
157
self.build_tree(['repo/b/file'])
159
tree_b.commit('commit b-1', rev_id='b-1')
161
self.assertTrue(shared_repo.has_revision('a-1'))
162
self.assertTrue(shared_repo.has_revision('a-2'))
163
self.assertTrue(shared_repo.has_revision('b-1'))
165
# Now that we have a repository with shared files, make sure
166
# that things aren't copied out by a 'push'
168
self.run_bzr('push ../../push-b')
169
pushed_tree = WorkingTree.open('../../push-b')
170
pushed_repo = pushed_tree.branch.repository
171
self.assertFalse(pushed_repo.has_revision('a-1'))
172
self.assertFalse(pushed_repo.has_revision('a-2'))
173
self.assertTrue(pushed_repo.has_revision('b-1'))
175
def test_push_funky_id(self):
176
t = self.make_branch_and_tree('tree')
178
self.build_tree(['filename'])
179
t.add('filename', 'funky-chars<>%&;"\'')
180
t.commit('commit filename')
181
self.run_bzr('push ../new-tree')
183
def test_push_dash_d(self):
184
t = self.make_branch_and_tree('from')
185
t.commit(allow_pointless=True,
186
message='first commit')
187
self.run_bzr('push -d from to-one')
188
self.failUnlessExists('to-one')
189
self.run_bzr('push -d %s %s'
190
% tuple(map(urlutils.local_path_to_url, ['from', 'to-two'])))
191
self.failUnlessExists('to-two')
193
def test_push_smart_non_stacked_streaming_acceptance(self):
194
self.setup_smart_server_with_call_log()
195
t = self.make_branch_and_tree('from')
196
t.commit(allow_pointless=True, message='first commit')
197
self.reset_smart_call_log()
198
self.run_bzr(['push', self.get_url('to-one')], working_dir='from')
199
# This figure represent the amount of work to perform this use case. It
200
# is entirely ok to reduce this number if a test fails due to rpc_count
201
# being too low. If rpc_count increases, more network roundtrips have
202
# become necessary for this use case. Please do not adjust this number
203
# upwards without agreement from bzr's network support maintainers.
204
self.assertLength(9, self.hpss_calls)
206
def test_push_smart_stacked_streaming_acceptance(self):
207
self.setup_smart_server_with_call_log()
208
parent = self.make_branch_and_tree('parent', format='1.9')
209
parent.commit(message='first commit')
210
local = parent.bzrdir.sprout('local').open_workingtree()
211
local.commit(message='local commit')
212
self.reset_smart_call_log()
213
self.run_bzr(['push', '--stacked', '--stacked-on', '../parent',
214
self.get_url('public')], working_dir='local')
215
# This figure represent the amount of work to perform this use case. It
216
# is entirely ok to reduce this number if a test fails due to rpc_count
217
# being too low. If rpc_count increases, more network roundtrips have
218
# become necessary for this use case. Please do not adjust this number
219
# upwards without agreement from bzr's network support maintainers.
220
self.assertLength(14, self.hpss_calls)
221
remote = Branch.open('public')
222
self.assertEndsWith(remote.get_stacked_on_url(), '/parent')
224
def create_simple_tree(self):
225
tree = self.make_branch_and_tree('tree')
226
self.build_tree(['tree/a'])
227
tree.add(['a'], ['a-id'])
228
tree.commit('one', rev_id='r1')
231
def test_push_create_prefix(self):
232
"""'bzr push --create-prefix' will create leading directories."""
233
tree = self.create_simple_tree()
235
self.run_bzr_error(['Parent directory of ../new/tree does not exist'],
238
self.run_bzr('push ../new/tree --create-prefix',
240
new_tree = WorkingTree.open('new/tree')
241
self.assertEqual(tree.last_revision(), new_tree.last_revision())
242
self.failUnlessExists('new/tree/a')
244
def test_push_use_existing(self):
245
"""'bzr push --use-existing-dir' can push into an existing dir.
247
By default, 'bzr push' will not use an existing, non-versioned dir.
249
tree = self.create_simple_tree()
250
self.build_tree(['target/'])
252
self.run_bzr_error(['Target directory ../target already exists',
253
'Supply --use-existing-dir',
255
'push ../target', working_dir='tree')
257
self.run_bzr('push --use-existing-dir ../target',
260
new_tree = WorkingTree.open('target')
261
self.assertEqual(tree.last_revision(), new_tree.last_revision())
262
# The push should have created target/a
263
self.failUnlessExists('target/a')
265
def test_push_onto_repo(self):
266
"""We should be able to 'bzr push' into an existing bzrdir."""
267
tree = self.create_simple_tree()
268
repo = self.make_repository('repo', shared=True)
270
self.run_bzr('push ../repo',
273
# Pushing onto an existing bzrdir will create a repository and
274
# branch as needed, but will only create a working tree if there was
276
self.assertRaises(errors.NoWorkingTree, WorkingTree.open, 'repo')
277
new_branch = Branch.open('repo')
278
self.assertEqual(tree.last_revision(), new_branch.last_revision())
280
def test_push_onto_just_bzrdir(self):
281
"""We don't handle when the target is just a bzrdir.
283
Because you shouldn't be able to create *just* a bzrdir in the wild.
285
# TODO: jam 20070109 Maybe it would be better to create the repository
287
tree = self.create_simple_tree()
288
a_bzrdir = self.make_bzrdir('dir')
290
self.run_bzr_error(['At ../dir you have a valid .bzr control'],
294
def test_push_with_revisionspec(self):
295
"""We should be able to push a revision older than the tip."""
296
tree_from = self.make_branch_and_tree('from')
297
tree_from.commit("One.", rev_id="from-1")
298
tree_from.commit("Two.", rev_id="from-2")
300
self.run_bzr('push -r1 ../to', working_dir='from')
302
tree_to = WorkingTree.open('to')
303
repo_to = tree_to.branch.repository
304
self.assertTrue(repo_to.has_revision('from-1'))
305
self.assertFalse(repo_to.has_revision('from-2'))
306
self.assertEqual(tree_to.branch.last_revision_info()[1], 'from-1')
309
['bzr: ERROR: bzr push --revision '
310
'takes exactly one revision identifier\n'],
311
'push -r0..2 ../to', working_dir='from')
313
def create_trunk_and_feature_branch(self):
315
trunk_tree = self.make_branch_and_tree('target',
317
trunk_tree.commit('mainline')
318
# and a branch from it
319
branch_tree = self.make_branch_and_tree('branch',
321
branch_tree.pull(trunk_tree.branch)
322
branch_tree.branch.set_parent(trunk_tree.branch.base)
323
# with some work on it
324
branch_tree.commit('moar work plz')
325
return trunk_tree, branch_tree
327
def assertPublished(self, branch_revid, stacked_on):
328
"""Assert that the branch 'published' has been published correctly."""
329
published_branch = Branch.open('published')
330
# The published branch refers to the mainline
331
self.assertEqual(stacked_on, published_branch.get_stacked_on_url())
332
# and the branch's work was pushed
333
self.assertTrue(published_branch.repository.has_revision(branch_revid))
335
def test_push_new_branch_stacked_on(self):
336
"""Pushing a new branch with --stacked-on creates a stacked branch."""
337
trunk_tree, branch_tree = self.create_trunk_and_feature_branch()
338
# we publish branch_tree with a reference to the mainline.
339
out, err = self.run_bzr(['push', '--stacked-on', trunk_tree.branch.base,
340
self.get_url('published')], working_dir='branch')
341
self.assertEqual('', out)
342
self.assertEqual('Created new stacked branch referring to %s.\n' %
343
trunk_tree.branch.base, err)
344
self.assertPublished(branch_tree.last_revision(),
345
trunk_tree.branch.base)
347
def test_push_new_branch_stacked_uses_parent_when_no_public_url(self):
348
"""When the parent has no public url the parent is used as-is."""
349
trunk_tree, branch_tree = self.create_trunk_and_feature_branch()
350
# now we do a stacked push, which should determine the public location
352
out, err = self.run_bzr(['push', '--stacked',
353
self.get_url('published')], working_dir='branch')
354
self.assertEqual('', out)
355
self.assertEqual('Created new stacked branch referring to %s.\n' %
356
trunk_tree.branch.base, err)
357
self.assertPublished(branch_tree.last_revision(), trunk_tree.branch.base)
359
def test_push_new_branch_stacked_uses_parent_public(self):
360
"""Pushing a new branch with --stacked creates a stacked branch."""
361
trunk_tree, branch_tree = self.create_trunk_and_feature_branch()
362
# the trunk is published on a web server
363
self.transport_readonly_server = HttpServer
364
trunk_public = self.make_branch('public_trunk', format='1.9')
365
trunk_public.pull(trunk_tree.branch)
366
trunk_public_url = self.get_readonly_url('public_trunk')
367
trunk_tree.branch.set_public_branch(trunk_public_url)
368
# now we do a stacked push, which should determine the public location
370
out, err = self.run_bzr(['push', '--stacked',
371
self.get_url('published')], working_dir='branch')
372
self.assertEqual('', out)
373
self.assertEqual('Created new stacked branch referring to %s.\n' %
374
trunk_public_url, err)
375
self.assertPublished(branch_tree.last_revision(), trunk_public_url)
377
def test_push_new_branch_stacked_no_parent(self):
378
"""Pushing with --stacked and no parent branch errors."""
379
branch = self.make_branch_and_tree('branch', format='1.9')
380
# now we do a stacked push, which should fail as the place to refer too
381
# cannot be determined.
382
out, err = self.run_bzr_error(
383
['Could not determine branch to refer to\\.'], ['push', '--stacked',
384
self.get_url('published')], working_dir='branch')
385
self.assertEqual('', out)
386
self.assertFalse(self.get_transport('published').has('.'))
388
def test_push_notifies_default_stacking(self):
389
self.make_branch('stack_on', format='1.6')
390
self.make_bzrdir('.').get_config().set_default_stack_on('stack_on')
391
self.make_branch('from', format='1.6')
392
out, err = self.run_bzr('push -d from to')
393
self.assertContainsRe(err,
394
'Using default stacking branch stack_on at .*')
396
def test_push_stacks_with_default_stacking_if_target_is_stackable(self):
397
self.make_branch('stack_on', format='1.6')
398
self.make_bzrdir('.').get_config().set_default_stack_on('stack_on')
399
self.make_branch('from', format='pack-0.92')
400
out, err = self.run_bzr('push -d from to')
401
branch = Branch.open('to')
402
self.assertEqual('../stack_on', branch.get_stacked_on_url())
404
def test_push_does_not_change_format_with_default_if_target_cannot(self):
405
self.make_branch('stack_on', format='pack-0.92')
406
self.make_bzrdir('.').get_config().set_default_stack_on('stack_on')
407
self.make_branch('from', format='pack-0.92')
408
out, err = self.run_bzr('push -d from to')
409
branch = Branch.open('to')
410
self.assertRaises(errors.UnstackableBranchFormat,
411
branch.get_stacked_on_url)
413
def test_push_doesnt_create_broken_branch(self):
414
"""Pushing a new standalone branch works even when there's a default
415
stacking policy at the destination.
417
The new branch will preserve the repo format (even if it isn't the
418
default for the branch), and will be stacked when the repo format
419
allows (which means that the branch format isn't necessarly preserved).
421
self.make_repository('repo', shared=True, format='1.6')
422
builder = self.make_branch_builder('repo/local', format='pack-0.92')
423
builder.start_series()
424
builder.build_snapshot('rev-1', None, [
425
('add', ('', 'root-id', 'directory', '')),
426
('add', ('filename', 'f-id', 'file', 'content\n'))])
427
builder.build_snapshot('rev-2', ['rev-1'], [])
428
builder.build_snapshot('rev-3', ['rev-2'],
429
[('modify', ('f-id', 'new-content\n'))])
430
builder.finish_series()
431
branch = builder.get_branch()
432
# Push rev-1 to "trunk", so that we can stack on it.
433
self.run_bzr('push -d repo/local trunk -r 1')
434
# Set a default stacking policy so that new branches will automatically
436
self.make_bzrdir('.').get_config().set_default_stack_on('trunk')
437
# Push rev-2 to a new branch "remote". It will be stacked on "trunk".
438
out, err = self.run_bzr('push -d repo/local remote -r 2')
439
self.assertContainsRe(
440
err, 'Using default stacking branch trunk at .*')
441
# Push rev-3 onto "remote". If "remote" not stacked and is missing the
442
# fulltext record for f-id @ rev-1, then this will fail.
443
out, err = self.run_bzr('push -d repo/local remote -r 3')
445
def test_push_verbose_shows_log(self):
446
tree = self.make_branch_and_tree('source')
448
out, err = self.run_bzr('push -v -d source target')
449
# initial push contains log
450
self.assertContainsRe(out, 'rev1')
452
out, err = self.run_bzr('push -v -d source target')
453
# subsequent push contains log
454
self.assertContainsRe(out, 'rev2')
455
# subsequent log is accurate
456
self.assertNotContainsRe(out, 'rev1')
459
class RedirectingMemoryTransport(MemoryTransport):
461
def mkdir(self, relpath, mode=None):
462
from bzrlib.trace import mutter
463
mutter('cwd: %r, rel: %r, abs: %r' % (self._cwd, relpath, abspath))
464
if self._cwd == '/source/':
465
raise errors.RedirectRequested(self.abspath(relpath),
466
self.abspath('../target'),
468
elif self._cwd == '/infinite-loop/':
469
raise errors.RedirectRequested(self.abspath(relpath),
470
self.abspath('../infinite-loop'),
473
return super(RedirectingMemoryTransport, self).mkdir(
476
def _redirected_to(self, source, target):
477
# We do accept redirections
478
return transport.get_transport(target)
481
class RedirectingMemoryServer(MemoryServer):
484
self._dirs = {'/': None}
487
self._scheme = 'redirecting-memory+%s:///' % id(self)
488
transport.register_transport(self._scheme, self._memory_factory)
490
def _memory_factory(self, url):
491
result = RedirectingMemoryTransport(url)
492
result._dirs = self._dirs
493
result._files = self._files
494
result._locks = self._locks
498
transport.unregister_transport(self._scheme, self._memory_factory)
501
class TestPushRedirect(ExternalBase):
504
ExternalBase.setUp(self)
505
self.memory_server = RedirectingMemoryServer()
506
self.memory_server.setUp()
507
self.addCleanup(self.memory_server.tearDown)
509
# Make the branch and tree that we'll be pushing.
510
t = self.make_branch_and_tree('tree')
511
self.build_tree(['tree/file'])
515
def test_push_redirects_on_mkdir(self):
516
"""If the push requires a mkdir, push respects redirect requests.
518
This is added primarily to handle lp:/ URI support, so that users can
519
push to new branches by specifying lp:/ URIs.
521
destination_url = self.memory_server.get_url() + 'source'
522
self.run_bzr(['push', '-d', 'tree', destination_url])
524
local_revision = Branch.open('tree').last_revision()
525
remote_revision = Branch.open(
526
self.memory_server.get_url() + 'target').last_revision()
527
self.assertEqual(remote_revision, local_revision)
529
def test_push_gracefully_handles_too_many_redirects(self):
530
"""Push fails gracefully if the mkdir generates a large number of
533
destination_url = self.memory_server.get_url() + 'infinite-loop'
534
out, err = self.run_bzr_error(
535
['Too many redirections trying to make %s\\.\n'
536
% re.escape(destination_url)],
537
['push', '-d', 'tree', destination_url], retcode=3)
538
self.assertEqual('', out)
added
removed
bzrlib/tests/blackbox/test_push.py
