~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

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

  • Committer: Martin Pool
  • Date: 2005-05-05 07:00:17 UTC
  • Revision ID: mbp@sourcefrog.net-20050505070017-6af6a766fc558dc2
todo

Show diffs side-by-side

added added

removed removed

Lines of Context:
1
 
# Copyright (C) 2005, 2007, 2008 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., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
16
 
 
17
 
 
18
 
"""Black-box tests for bzr push."""
19
 
 
20
 
import re
21
 
 
22
 
from bzrlib import (
23
 
    branch,
24
 
    bzrdir,
25
 
    errors,
26
 
    osutils,
27
 
    tests,
28
 
    transport,
29
 
    uncommit,
30
 
    urlutils,
31
 
    workingtree
32
 
    )
33
 
from bzrlib.repofmt import knitrepo
34
 
from bzrlib.tests import (
35
 
    blackbox,
36
 
    http_server,
37
 
    test_foreign,
38
 
    )
39
 
from bzrlib.transport import memory
40
 
 
41
 
 
42
 
def load_tests(standard_tests, module, loader):
43
 
    """Multiply tests for the push command."""
44
 
    result = loader.suiteClass()
45
 
 
46
 
    # one for each king of change
47
 
    changes_tests, remaining_tests = tests.split_suite_by_condition(
48
 
        standard_tests, tests.condition_isinstance((
49
 
                TestPushStrictWithChanges,
50
 
                )))
51
 
    changes_scenarios = [
52
 
        ('uncommitted',
53
 
         dict(_changes_type= '_uncommitted_changes')),
54
 
        ('pending-merges',
55
 
         dict(_changes_type= '_pending_merges')),
56
 
        ('out-of-sync-trees',
57
 
         dict(_changes_type= '_out_of_sync_trees')),
58
 
        ]
59
 
    tests.multiply_tests(changes_tests, changes_scenarios, result)
60
 
    # No parametrization for the remaining tests
61
 
    result.addTests(remaining_tests)
62
 
 
63
 
    return result
64
 
 
65
 
 
66
 
class TestPush(tests.TestCaseWithTransport):
67
 
 
68
 
    def test_push_error_on_vfs_http(self):
69
 
        """ pushing a branch to a HTTP server fails cleanly. """
70
 
        # the trunk is published on a web server
71
 
        self.transport_readonly_server = http_server.HttpServer
72
 
        self.make_branch('source')
73
 
        public_url = self.get_readonly_url('target')
74
 
        self.run_bzr_error(['http does not support mkdir'],
75
 
                           ['push', public_url],
76
 
                           working_dir='source')
77
 
 
78
 
    def test_push_remember(self):
79
 
        """Push changes from one branch to another and test push location."""
80
 
        transport = self.get_transport()
81
 
        tree_a = self.make_branch_and_tree('branch_a')
82
 
        branch_a = tree_a.branch
83
 
        self.build_tree(['branch_a/a'])
84
 
        tree_a.add('a')
85
 
        tree_a.commit('commit a')
86
 
        tree_b = branch_a.bzrdir.sprout('branch_b').open_workingtree()
87
 
        branch_b = tree_b.branch
88
 
        tree_c = branch_a.bzrdir.sprout('branch_c').open_workingtree()
89
 
        branch_c = tree_c.branch
90
 
        self.build_tree(['branch_a/b'])
91
 
        tree_a.add('b')
92
 
        tree_a.commit('commit b')
93
 
        self.build_tree(['branch_b/c'])
94
 
        tree_b.add('c')
95
 
        tree_b.commit('commit c')
96
 
        # initial push location must be empty
97
 
        self.assertEqual(None, branch_b.get_push_location())
98
 
 
99
 
        # test push for failure without push location set
100
 
        out = self.run_bzr('push', working_dir='branch_a', retcode=3)
101
 
        self.assertEquals(out,
102
 
                ('','bzr: ERROR: No push location known or specified.\n'))
103
 
 
104
 
        # test not remembered if cannot actually push
105
 
        self.run_bzr('push path/which/doesnt/exist',
106
 
                     working_dir='branch_a', retcode=3)
107
 
        out = self.run_bzr('push', working_dir='branch_a', retcode=3)
108
 
        self.assertEquals(
109
 
                ('', 'bzr: ERROR: No push location known or specified.\n'),
110
 
                out)
111
 
 
112
 
        # test implicit --remember when no push location set, push fails
113
 
        out = self.run_bzr('push ../branch_b',
114
 
                           working_dir='branch_a', retcode=3)
115
 
        self.assertEquals(out,
116
 
                ('','bzr: ERROR: These branches have diverged.  '
117
 
                 'See "bzr help diverged-branches" for more information.\n'))
118
 
        self.assertEquals(osutils.abspath(branch_a.get_push_location()),
119
 
                          osutils.abspath(branch_b.bzrdir.root_transport.base))
120
 
 
121
 
        # test implicit --remember after resolving previous failure
122
 
        uncommit.uncommit(branch=branch_b, tree=tree_b)
123
 
        transport.delete('branch_b/c')
124
 
        out, err = self.run_bzr('push', working_dir='branch_a')
125
 
        path = branch_a.get_push_location()
126
 
        self.assertEquals(out,
127
 
                          'Using saved push location: %s\n'
128
 
                          % urlutils.local_path_from_url(path))
129
 
        self.assertEqual(err,
130
 
                         'All changes applied successfully.\n'
131
 
                         'Pushed up to revision 2.\n')
132
 
        self.assertEqual(path,
133
 
                         branch_b.bzrdir.root_transport.base)
134
 
        # test explicit --remember
135
 
        self.run_bzr('push ../branch_c --remember', working_dir='branch_a')
136
 
        self.assertEquals(branch_a.get_push_location(),
137
 
                          branch_c.bzrdir.root_transport.base)
138
 
 
139
 
    def test_push_without_tree(self):
140
 
        # bzr push from a branch that does not have a checkout should work.
141
 
        b = self.make_branch('.')
142
 
        out, err = self.run_bzr('push pushed-location')
143
 
        self.assertEqual('', out)
144
 
        self.assertEqual('Created new branch.\n', err)
145
 
        b2 = branch.Branch.open('pushed-location')
146
 
        self.assertEndsWith(b2.base, 'pushed-location/')
147
 
 
148
 
    def test_push_new_branch_revision_count(self):
149
 
        # bzr push of a branch with revisions to a new location
150
 
        # should print the number of revisions equal to the length of the
151
 
        # local branch.
152
 
        t = self.make_branch_and_tree('tree')
153
 
        self.build_tree(['tree/file'])
154
 
        t.add('file')
155
 
        t.commit('commit 1')
156
 
        out, err = self.run_bzr('push -d tree pushed-to')
157
 
        self.assertEqual('', out)
158
 
        self.assertEqual('Created new branch.\n', err)
159
 
 
160
 
    def test_push_only_pushes_history(self):
161
 
        # Knit branches should only push the history for the current revision.
162
 
        format = bzrdir.BzrDirMetaFormat1()
163
 
        format.repository_format = knitrepo.RepositoryFormatKnit1()
164
 
        shared_repo = self.make_repository('repo', format=format, shared=True)
165
 
        shared_repo.set_make_working_trees(True)
166
 
 
167
 
        def make_shared_tree(path):
168
 
            shared_repo.bzrdir.root_transport.mkdir(path)
169
 
            shared_repo.bzrdir.create_branch_convenience('repo/' + path)
170
 
            return workingtree.WorkingTree.open('repo/' + path)
171
 
        tree_a = make_shared_tree('a')
172
 
        self.build_tree(['repo/a/file'])
173
 
        tree_a.add('file')
174
 
        tree_a.commit('commit a-1', rev_id='a-1')
175
 
        f = open('repo/a/file', 'ab')
176
 
        f.write('more stuff\n')
177
 
        f.close()
178
 
        tree_a.commit('commit a-2', rev_id='a-2')
179
 
 
180
 
        tree_b = make_shared_tree('b')
181
 
        self.build_tree(['repo/b/file'])
182
 
        tree_b.add('file')
183
 
        tree_b.commit('commit b-1', rev_id='b-1')
184
 
 
185
 
        self.assertTrue(shared_repo.has_revision('a-1'))
186
 
        self.assertTrue(shared_repo.has_revision('a-2'))
187
 
        self.assertTrue(shared_repo.has_revision('b-1'))
188
 
 
189
 
        # Now that we have a repository with shared files, make sure
190
 
        # that things aren't copied out by a 'push'
191
 
        self.run_bzr('push ../../push-b', working_dir='repo/b')
192
 
        pushed_tree = workingtree.WorkingTree.open('push-b')
193
 
        pushed_repo = pushed_tree.branch.repository
194
 
        self.assertFalse(pushed_repo.has_revision('a-1'))
195
 
        self.assertFalse(pushed_repo.has_revision('a-2'))
196
 
        self.assertTrue(pushed_repo.has_revision('b-1'))
197
 
 
198
 
    def test_push_funky_id(self):
199
 
        t = self.make_branch_and_tree('tree')
200
 
        self.build_tree(['tree/filename'])
201
 
        t.add('filename', 'funky-chars<>%&;"\'')
202
 
        t.commit('commit filename')
203
 
        self.run_bzr('push -d tree new-tree')
204
 
 
205
 
    def test_push_dash_d(self):
206
 
        t = self.make_branch_and_tree('from')
207
 
        t.commit(allow_pointless=True,
208
 
                message='first commit')
209
 
        self.run_bzr('push -d from to-one')
210
 
        self.failUnlessExists('to-one')
211
 
        self.run_bzr('push -d %s %s'
212
 
            % tuple(map(urlutils.local_path_to_url, ['from', 'to-two'])))
213
 
        self.failUnlessExists('to-two')
214
 
 
215
 
    def test_push_smart_non_stacked_streaming_acceptance(self):
216
 
        self.setup_smart_server_with_call_log()
217
 
        t = self.make_branch_and_tree('from')
218
 
        t.commit(allow_pointless=True, message='first commit')
219
 
        self.reset_smart_call_log()
220
 
        self.run_bzr(['push', self.get_url('to-one')], working_dir='from')
221
 
        # This figure represent the amount of work to perform this use case. It
222
 
        # is entirely ok to reduce this number if a test fails due to rpc_count
223
 
        # being too low. If rpc_count increases, more network roundtrips have
224
 
        # become necessary for this use case. Please do not adjust this number
225
 
        # upwards without agreement from bzr's network support maintainers.
226
 
        self.assertLength(9, self.hpss_calls)
227
 
 
228
 
    def test_push_smart_stacked_streaming_acceptance(self):
229
 
        self.setup_smart_server_with_call_log()
230
 
        parent = self.make_branch_and_tree('parent', format='1.9')
231
 
        parent.commit(message='first commit')
232
 
        local = parent.bzrdir.sprout('local').open_workingtree()
233
 
        local.commit(message='local commit')
234
 
        self.reset_smart_call_log()
235
 
        self.run_bzr(['push', '--stacked', '--stacked-on', '../parent',
236
 
            self.get_url('public')], working_dir='local')
237
 
        # This figure represent the amount of work to perform this use case. It
238
 
        # is entirely ok to reduce this number if a test fails due to rpc_count
239
 
        # being too low. If rpc_count increases, more network roundtrips have
240
 
        # become necessary for this use case. Please do not adjust this number
241
 
        # upwards without agreement from bzr's network support maintainers.
242
 
        self.assertLength(14, self.hpss_calls)
243
 
        remote = branch.Branch.open('public')
244
 
        self.assertEndsWith(remote.get_stacked_on_url(), '/parent')
245
 
 
246
 
    def test_push_smart_tags_streaming_acceptance(self):
247
 
        self.setup_smart_server_with_call_log()
248
 
        t = self.make_branch_and_tree('from')
249
 
        rev_id = t.commit(allow_pointless=True, message='first commit')
250
 
        t.branch.tags.set_tag('new-tag', rev_id)
251
 
        self.reset_smart_call_log()
252
 
        self.run_bzr(['push', self.get_url('to-one')], working_dir='from')
253
 
        # This figure represent the amount of work to perform this use case. It
254
 
        # is entirely ok to reduce this number if a test fails due to rpc_count
255
 
        # being too low. If rpc_count increases, more network roundtrips have
256
 
        # become necessary for this use case. Please do not adjust this number
257
 
        # upwards without agreement from bzr's network support maintainers.
258
 
        self.assertLength(11, self.hpss_calls)
259
 
 
260
 
    def test_push_smart_incremental_acceptance(self):
261
 
        self.setup_smart_server_with_call_log()
262
 
        t = self.make_branch_and_tree('from')
263
 
        rev_id1 = t.commit(allow_pointless=True, message='first commit')
264
 
        rev_id2 = t.commit(allow_pointless=True, message='second commit')
265
 
        self.run_bzr(
266
 
            ['push', self.get_url('to-one'), '-r1'], working_dir='from')
267
 
        self.reset_smart_call_log()
268
 
        self.run_bzr(['push', self.get_url('to-one')], working_dir='from')
269
 
        # This figure represent the amount of work to perform this use case. It
270
 
        # is entirely ok to reduce this number if a test fails due to rpc_count
271
 
        # being too low. If rpc_count increases, more network roundtrips have
272
 
        # become necessary for this use case. Please do not adjust this number
273
 
        # upwards without agreement from bzr's network support maintainers.
274
 
        self.assertLength(11, self.hpss_calls)
275
 
 
276
 
    def test_push_smart_with_default_stacking_url_path_segment(self):
277
 
        # If the default stacked-on location is a path element then branches
278
 
        # we push there over the smart server are stacked and their
279
 
        # stacked_on_url is that exact path segment. Added to nail bug 385132.
280
 
        self.setup_smart_server_with_call_log()
281
 
        self.make_branch('stack-on', format='1.9')
282
 
        self.make_bzrdir('.').get_config().set_default_stack_on(
283
 
            '/stack-on')
284
 
        self.make_branch('from', format='1.9')
285
 
        out, err = self.run_bzr(['push', '-d', 'from', self.get_url('to')])
286
 
        b = branch.Branch.open(self.get_url('to'))
287
 
        self.assertEqual('/extra/stack-on', b.get_stacked_on_url())
288
 
 
289
 
    def test_push_smart_with_default_stacking_relative_path(self):
290
 
        # If the default stacked-on location is a relative path then branches
291
 
        # we push there over the smart server are stacked and their
292
 
        # stacked_on_url is a relative path. Added to nail bug 385132.
293
 
        self.setup_smart_server_with_call_log()
294
 
        self.make_branch('stack-on', format='1.9')
295
 
        self.make_bzrdir('.').get_config().set_default_stack_on('stack-on')
296
 
        self.make_branch('from', format='1.9')
297
 
        out, err = self.run_bzr(['push', '-d', 'from', self.get_url('to')])
298
 
        b = branch.Branch.open(self.get_url('to'))
299
 
        self.assertEqual('../stack-on', b.get_stacked_on_url())
300
 
 
301
 
    def create_simple_tree(self):
302
 
        tree = self.make_branch_and_tree('tree')
303
 
        self.build_tree(['tree/a'])
304
 
        tree.add(['a'], ['a-id'])
305
 
        tree.commit('one', rev_id='r1')
306
 
        return tree
307
 
 
308
 
    def test_push_create_prefix(self):
309
 
        """'bzr push --create-prefix' will create leading directories."""
310
 
        tree = self.create_simple_tree()
311
 
 
312
 
        self.run_bzr_error(['Parent directory of ../new/tree does not exist'],
313
 
                           'push ../new/tree',
314
 
                           working_dir='tree')
315
 
        self.run_bzr('push ../new/tree --create-prefix',
316
 
                     working_dir='tree')
317
 
        new_tree = workingtree.WorkingTree.open('new/tree')
318
 
        self.assertEqual(tree.last_revision(), new_tree.last_revision())
319
 
        self.failUnlessExists('new/tree/a')
320
 
 
321
 
    def test_push_use_existing(self):
322
 
        """'bzr push --use-existing-dir' can push into an existing dir.
323
 
 
324
 
        By default, 'bzr push' will not use an existing, non-versioned dir.
325
 
        """
326
 
        tree = self.create_simple_tree()
327
 
        self.build_tree(['target/'])
328
 
 
329
 
        self.run_bzr_error(['Target directory ../target already exists',
330
 
                            'Supply --use-existing-dir',
331
 
                           ],
332
 
                           'push ../target', working_dir='tree')
333
 
 
334
 
        self.run_bzr('push --use-existing-dir ../target',
335
 
                     working_dir='tree')
336
 
 
337
 
        new_tree = workingtree.WorkingTree.open('target')
338
 
        self.assertEqual(tree.last_revision(), new_tree.last_revision())
339
 
        # The push should have created target/a
340
 
        self.failUnlessExists('target/a')
341
 
 
342
 
    def test_push_use_existing_into_empty_bzrdir(self):
343
 
        """'bzr push --use-existing-dir' into a dir with an empty .bzr dir
344
 
        fails.
345
 
        """
346
 
        tree = self.create_simple_tree()
347
 
        self.build_tree(['target/', 'target/.bzr/'])
348
 
        self.run_bzr_error(
349
 
            ['Target directory ../target already contains a .bzr directory, '
350
 
             'but it is not valid.'],
351
 
            'push ../target --use-existing-dir', working_dir='tree')
352
 
 
353
 
    def test_push_onto_repo(self):
354
 
        """We should be able to 'bzr push' into an existing bzrdir."""
355
 
        tree = self.create_simple_tree()
356
 
        repo = self.make_repository('repo', shared=True)
357
 
 
358
 
        self.run_bzr('push ../repo',
359
 
                     working_dir='tree')
360
 
 
361
 
        # Pushing onto an existing bzrdir will create a repository and
362
 
        # branch as needed, but will only create a working tree if there was
363
 
        # no BzrDir before.
364
 
        self.assertRaises(errors.NoWorkingTree,
365
 
                          workingtree.WorkingTree.open, 'repo')
366
 
        new_branch = branch.Branch.open('repo')
367
 
        self.assertEqual(tree.last_revision(), new_branch.last_revision())
368
 
 
369
 
    def test_push_onto_just_bzrdir(self):
370
 
        """We don't handle when the target is just a bzrdir.
371
 
 
372
 
        Because you shouldn't be able to create *just* a bzrdir in the wild.
373
 
        """
374
 
        # TODO: jam 20070109 Maybe it would be better to create the repository
375
 
        #       if at this point
376
 
        tree = self.create_simple_tree()
377
 
        a_bzrdir = self.make_bzrdir('dir')
378
 
 
379
 
        self.run_bzr_error(['At ../dir you have a valid .bzr control'],
380
 
                'push ../dir',
381
 
                working_dir='tree')
382
 
 
383
 
    def test_push_with_revisionspec(self):
384
 
        """We should be able to push a revision older than the tip."""
385
 
        tree_from = self.make_branch_and_tree('from')
386
 
        tree_from.commit("One.", rev_id="from-1")
387
 
        tree_from.commit("Two.", rev_id="from-2")
388
 
 
389
 
        self.run_bzr('push -r1 ../to', working_dir='from')
390
 
 
391
 
        tree_to = workingtree.WorkingTree.open('to')
392
 
        repo_to = tree_to.branch.repository
393
 
        self.assertTrue(repo_to.has_revision('from-1'))
394
 
        self.assertFalse(repo_to.has_revision('from-2'))
395
 
        self.assertEqual(tree_to.branch.last_revision_info()[1], 'from-1')
396
 
 
397
 
        self.run_bzr_error(
398
 
            ['bzr: ERROR: bzr push --revision '
399
 
             'takes exactly one revision identifier\n'],
400
 
            'push -r0..2 ../to', working_dir='from')
401
 
 
402
 
    def create_trunk_and_feature_branch(self):
403
 
        # We have a mainline
404
 
        trunk_tree = self.make_branch_and_tree('target',
405
 
            format='1.9')
406
 
        trunk_tree.commit('mainline')
407
 
        # and a branch from it
408
 
        branch_tree = self.make_branch_and_tree('branch',
409
 
            format='1.9')
410
 
        branch_tree.pull(trunk_tree.branch)
411
 
        branch_tree.branch.set_parent(trunk_tree.branch.base)
412
 
        # with some work on it
413
 
        branch_tree.commit('moar work plz')
414
 
        return trunk_tree, branch_tree
415
 
 
416
 
    def assertPublished(self, branch_revid, stacked_on):
417
 
        """Assert that the branch 'published' has been published correctly."""
418
 
        published_branch = branch.Branch.open('published')
419
 
        # The published branch refers to the mainline
420
 
        self.assertEqual(stacked_on, published_branch.get_stacked_on_url())
421
 
        # and the branch's work was pushed
422
 
        self.assertTrue(published_branch.repository.has_revision(branch_revid))
423
 
 
424
 
    def test_push_new_branch_stacked_on(self):
425
 
        """Pushing a new branch with --stacked-on creates a stacked branch."""
426
 
        trunk_tree, branch_tree = self.create_trunk_and_feature_branch()
427
 
        # we publish branch_tree with a reference to the mainline.
428
 
        out, err = self.run_bzr(['push', '--stacked-on', trunk_tree.branch.base,
429
 
            self.get_url('published')], working_dir='branch')
430
 
        self.assertEqual('', out)
431
 
        self.assertEqual('Created new stacked branch referring to %s.\n' %
432
 
            trunk_tree.branch.base, err)
433
 
        self.assertPublished(branch_tree.last_revision(),
434
 
            trunk_tree.branch.base)
435
 
 
436
 
    def test_push_new_branch_stacked_uses_parent_when_no_public_url(self):
437
 
        """When the parent has no public url the parent is used as-is."""
438
 
        trunk_tree, branch_tree = self.create_trunk_and_feature_branch()
439
 
        # now we do a stacked push, which should determine the public location
440
 
        # for us.
441
 
        out, err = self.run_bzr(['push', '--stacked',
442
 
            self.get_url('published')], working_dir='branch')
443
 
        self.assertEqual('', out)
444
 
        self.assertEqual('Created new stacked branch referring to %s.\n' %
445
 
            trunk_tree.branch.base, err)
446
 
        self.assertPublished(branch_tree.last_revision(),
447
 
                             trunk_tree.branch.base)
448
 
 
449
 
    def test_push_new_branch_stacked_uses_parent_public(self):
450
 
        """Pushing a new branch with --stacked creates a stacked branch."""
451
 
        trunk_tree, branch_tree = self.create_trunk_and_feature_branch()
452
 
        # the trunk is published on a web server
453
 
        self.transport_readonly_server = http_server.HttpServer
454
 
        trunk_public = self.make_branch('public_trunk', format='1.9')
455
 
        trunk_public.pull(trunk_tree.branch)
456
 
        trunk_public_url = self.get_readonly_url('public_trunk')
457
 
        trunk_tree.branch.set_public_branch(trunk_public_url)
458
 
        # now we do a stacked push, which should determine the public location
459
 
        # for us.
460
 
        out, err = self.run_bzr(['push', '--stacked',
461
 
            self.get_url('published')], working_dir='branch')
462
 
        self.assertEqual('', out)
463
 
        self.assertEqual('Created new stacked branch referring to %s.\n' %
464
 
            trunk_public_url, err)
465
 
        self.assertPublished(branch_tree.last_revision(), trunk_public_url)
466
 
 
467
 
    def test_push_new_branch_stacked_no_parent(self):
468
 
        """Pushing with --stacked and no parent branch errors."""
469
 
        branch = self.make_branch_and_tree('branch', format='1.9')
470
 
        # now we do a stacked push, which should fail as the place to refer too
471
 
        # cannot be determined.
472
 
        out, err = self.run_bzr_error(
473
 
            ['Could not determine branch to refer to\\.'], ['push', '--stacked',
474
 
            self.get_url('published')], working_dir='branch')
475
 
        self.assertEqual('', out)
476
 
        self.assertFalse(self.get_transport('published').has('.'))
477
 
 
478
 
    def test_push_notifies_default_stacking(self):
479
 
        self.make_branch('stack_on', format='1.6')
480
 
        self.make_bzrdir('.').get_config().set_default_stack_on('stack_on')
481
 
        self.make_branch('from', format='1.6')
482
 
        out, err = self.run_bzr('push -d from to')
483
 
        self.assertContainsRe(err,
484
 
                              'Using default stacking branch stack_on at .*')
485
 
 
486
 
    def test_push_stacks_with_default_stacking_if_target_is_stackable(self):
487
 
        self.make_branch('stack_on', format='1.6')
488
 
        self.make_bzrdir('.').get_config().set_default_stack_on('stack_on')
489
 
        self.make_branch('from', format='pack-0.92')
490
 
        out, err = self.run_bzr('push -d from to')
491
 
        b = branch.Branch.open('to')
492
 
        self.assertEqual('../stack_on', b.get_stacked_on_url())
493
 
 
494
 
    def test_push_does_not_change_format_with_default_if_target_cannot(self):
495
 
        self.make_branch('stack_on', format='pack-0.92')
496
 
        self.make_bzrdir('.').get_config().set_default_stack_on('stack_on')
497
 
        self.make_branch('from', format='pack-0.92')
498
 
        out, err = self.run_bzr('push -d from to')
499
 
        b = branch.Branch.open('to')
500
 
        self.assertRaises(errors.UnstackableBranchFormat, b.get_stacked_on_url)
501
 
 
502
 
    def test_push_doesnt_create_broken_branch(self):
503
 
        """Pushing a new standalone branch works even when there's a default
504
 
        stacking policy at the destination.
505
 
 
506
 
        The new branch will preserve the repo format (even if it isn't the
507
 
        default for the branch), and will be stacked when the repo format
508
 
        allows (which means that the branch format isn't necessarly preserved).
509
 
        """
510
 
        self.make_repository('repo', shared=True, format='1.6')
511
 
        builder = self.make_branch_builder('repo/local', format='pack-0.92')
512
 
        builder.start_series()
513
 
        builder.build_snapshot('rev-1', None, [
514
 
            ('add', ('', 'root-id', 'directory', '')),
515
 
            ('add', ('filename', 'f-id', 'file', 'content\n'))])
516
 
        builder.build_snapshot('rev-2', ['rev-1'], [])
517
 
        builder.build_snapshot('rev-3', ['rev-2'],
518
 
            [('modify', ('f-id', 'new-content\n'))])
519
 
        builder.finish_series()
520
 
        branch = builder.get_branch()
521
 
        # Push rev-1 to "trunk", so that we can stack on it.
522
 
        self.run_bzr('push -d repo/local trunk -r 1')
523
 
        # Set a default stacking policy so that new branches will automatically
524
 
        # stack on trunk.
525
 
        self.make_bzrdir('.').get_config().set_default_stack_on('trunk')
526
 
        # Push rev-2 to a new branch "remote".  It will be stacked on "trunk".
527
 
        out, err = self.run_bzr('push -d repo/local remote -r 2')
528
 
        self.assertContainsRe(
529
 
            err, 'Using default stacking branch trunk at .*')
530
 
        # Push rev-3 onto "remote".  If "remote" not stacked and is missing the
531
 
        # fulltext record for f-id @ rev-1, then this will fail.
532
 
        out, err = self.run_bzr('push -d repo/local remote -r 3')
533
 
 
534
 
    def test_push_verbose_shows_log(self):
535
 
        tree = self.make_branch_and_tree('source')
536
 
        tree.commit('rev1')
537
 
        out, err = self.run_bzr('push -v -d source target')
538
 
        # initial push contains log
539
 
        self.assertContainsRe(out, 'rev1')
540
 
        tree.commit('rev2')
541
 
        out, err = self.run_bzr('push -v -d source target')
542
 
        # subsequent push contains log
543
 
        self.assertContainsRe(out, 'rev2')
544
 
        # subsequent log is accurate
545
 
        self.assertNotContainsRe(out, 'rev1')
546
 
 
547
 
    def test_push_from_subdir(self):
548
 
        t = self.make_branch_and_tree('tree')
549
 
        self.build_tree(['tree/dir/', 'tree/dir/file'])
550
 
        t.add('dir', 'dir/file')
551
 
        t.commit('r1')
552
 
        out, err = self.run_bzr('push ../../pushloc', working_dir='tree/dir')
553
 
        self.assertEqual('', out)
554
 
        self.assertEqual('Created new branch.\n', err)
555
 
 
556
 
 
557
 
class RedirectingMemoryTransport(memory.MemoryTransport):
558
 
 
559
 
    def mkdir(self, relpath, mode=None):
560
 
        if self._cwd == '/source/':
561
 
            raise errors.RedirectRequested(self.abspath(relpath),
562
 
                                           self.abspath('../target'),
563
 
                                           is_permanent=True)
564
 
        elif self._cwd == '/infinite-loop/':
565
 
            raise errors.RedirectRequested(self.abspath(relpath),
566
 
                                           self.abspath('../infinite-loop'),
567
 
                                           is_permanent=True)
568
 
        else:
569
 
            return super(RedirectingMemoryTransport, self).mkdir(
570
 
                relpath, mode)
571
 
 
572
 
    def get(self, relpath):
573
 
        if self.clone(relpath)._cwd == '/infinite-loop/':
574
 
            raise errors.RedirectRequested(self.abspath(relpath),
575
 
                                           self.abspath('../infinite-loop'),
576
 
                                           is_permanent=True)
577
 
        else:
578
 
            return super(RedirectingMemoryTransport, self).get(relpath)
579
 
 
580
 
    def _redirected_to(self, source, target):
581
 
        # We do accept redirections
582
 
        return transport.get_transport(target)
583
 
 
584
 
 
585
 
class RedirectingMemoryServer(memory.MemoryServer):
586
 
 
587
 
    def start_server(self):
588
 
        self._dirs = {'/': None}
589
 
        self._files = {}
590
 
        self._locks = {}
591
 
        self._scheme = 'redirecting-memory+%s:///' % id(self)
592
 
        transport.register_transport(self._scheme, self._memory_factory)
593
 
 
594
 
    def _memory_factory(self, url):
595
 
        result = RedirectingMemoryTransport(url)
596
 
        result._dirs = self._dirs
597
 
        result._files = self._files
598
 
        result._locks = self._locks
599
 
        return result
600
 
 
601
 
    def stop_server(self):
602
 
        transport.unregister_transport(self._scheme, self._memory_factory)
603
 
 
604
 
 
605
 
class TestPushRedirect(tests.TestCaseWithTransport):
606
 
 
607
 
    def setUp(self):
608
 
        tests.TestCaseWithTransport.setUp(self)
609
 
        self.memory_server = RedirectingMemoryServer()
610
 
        self.start_server(self.memory_server)
611
 
        # Make the branch and tree that we'll be pushing.
612
 
        t = self.make_branch_and_tree('tree')
613
 
        self.build_tree(['tree/file'])
614
 
        t.add('file')
615
 
        t.commit('commit 1')
616
 
 
617
 
    def test_push_redirects_on_mkdir(self):
618
 
        """If the push requires a mkdir, push respects redirect requests.
619
 
 
620
 
        This is added primarily to handle lp:/ URI support, so that users can
621
 
        push to new branches by specifying lp:/ URIs.
622
 
        """
623
 
        destination_url = self.memory_server.get_url() + 'source'
624
 
        self.run_bzr(['push', '-d', 'tree', destination_url])
625
 
 
626
 
        local_revision = branch.Branch.open('tree').last_revision()
627
 
        remote_revision = branch.Branch.open(
628
 
            self.memory_server.get_url() + 'target').last_revision()
629
 
        self.assertEqual(remote_revision, local_revision)
630
 
 
631
 
    def test_push_gracefully_handles_too_many_redirects(self):
632
 
        """Push fails gracefully if the mkdir generates a large number of
633
 
        redirects.
634
 
        """
635
 
        destination_url = self.memory_server.get_url() + 'infinite-loop'
636
 
        out, err = self.run_bzr_error(
637
 
            ['Too many redirections trying to make %s\\.\n'
638
 
             % re.escape(destination_url)],
639
 
            ['push', '-d', 'tree', destination_url], retcode=3)
640
 
        self.assertEqual('', out)
641
 
 
642
 
 
643
 
class TestPushStrictMixin(object):
644
 
 
645
 
    def make_local_branch_and_tree(self):
646
 
        self.tree = self.make_branch_and_tree('local')
647
 
        self.build_tree_contents([('local/file', 'initial')])
648
 
        self.tree.add('file')
649
 
        self.tree.commit('adding file', rev_id='added')
650
 
        self.build_tree_contents([('local/file', 'modified')])
651
 
        self.tree.commit('modify file', rev_id='modified')
652
 
 
653
 
    def set_config_push_strict(self, value):
654
 
        # set config var (any of bazaar.conf, locations.conf, branch.conf
655
 
        # should do)
656
 
        conf = self.tree.branch.get_config()
657
 
        conf.set_user_option('push_strict', value)
658
 
 
659
 
    _default_command = ['push', '../to']
660
 
    _default_wd = 'local'
661
 
    _default_errors = ['Working tree ".*/local/" has uncommitted '
662
 
                       'changes \(See bzr status\)\.',]
663
 
    _default_pushed_revid = 'modified'
664
 
 
665
 
    def assertPushFails(self, args):
666
 
        self.run_bzr_error(self._default_errors, self._default_command + args,
667
 
                           working_dir=self._default_wd, retcode=3)
668
 
 
669
 
    def assertPushSucceeds(self, args, pushed_revid=None):
670
 
        self.run_bzr(self._default_command + args,
671
 
                     working_dir=self._default_wd)
672
 
        if pushed_revid is None:
673
 
            pushed_revid = self._default_pushed_revid
674
 
        tree_to = workingtree.WorkingTree.open('to')
675
 
        repo_to = tree_to.branch.repository
676
 
        self.assertTrue(repo_to.has_revision(pushed_revid))
677
 
        self.assertEqual(tree_to.branch.last_revision_info()[1], pushed_revid)
678
 
 
679
 
 
680
 
 
681
 
class TestPushStrictWithoutChanges(tests.TestCaseWithTransport,
682
 
                                   TestPushStrictMixin):
683
 
 
684
 
    def setUp(self):
685
 
        super(TestPushStrictWithoutChanges, self).setUp()
686
 
        self.make_local_branch_and_tree()
687
 
 
688
 
    def test_push_default(self):
689
 
        self.assertPushSucceeds([])
690
 
 
691
 
    def test_push_strict(self):
692
 
        self.assertPushSucceeds(['--strict'])
693
 
 
694
 
    def test_push_no_strict(self):
695
 
        self.assertPushSucceeds(['--no-strict'])
696
 
 
697
 
    def test_push_config_var_strict(self):
698
 
        self.set_config_push_strict('true')
699
 
        self.assertPushSucceeds([])
700
 
 
701
 
    def test_push_config_var_no_strict(self):
702
 
        self.set_config_push_strict('false')
703
 
        self.assertPushSucceeds([])
704
 
 
705
 
 
706
 
class TestPushStrictWithChanges(tests.TestCaseWithTransport,
707
 
                                TestPushStrictMixin):
708
 
 
709
 
    _changes_type = None # Set by load_tests
710
 
 
711
 
    def setUp(self):
712
 
        super(TestPushStrictWithChanges, self).setUp()
713
 
        # Apply the changes defined in load_tests: one of _uncommitted_changes,
714
 
        # _pending_merges or _out_of_sync_trees
715
 
        getattr(self, self._changes_type)()
716
 
 
717
 
    def _uncommitted_changes(self):
718
 
        self.make_local_branch_and_tree()
719
 
        # Make a change without committing it
720
 
        self.build_tree_contents([('local/file', 'in progress')])
721
 
 
722
 
    def _pending_merges(self):
723
 
        self.make_local_branch_and_tree()
724
 
        # Create 'other' branch containing a new file
725
 
        other_bzrdir = self.tree.bzrdir.sprout('other')
726
 
        other_tree = other_bzrdir.open_workingtree()
727
 
        self.build_tree_contents([('other/other-file', 'other')])
728
 
        other_tree.add('other-file')
729
 
        other_tree.commit('other commit', rev_id='other')
730
 
        # Merge and revert, leaving a pending merge
731
 
        self.tree.merge_from_branch(other_tree.branch)
732
 
        self.tree.revert(filenames=['other-file'], backups=False)
733
 
 
734
 
    def _out_of_sync_trees(self):
735
 
        self.make_local_branch_and_tree()
736
 
        self.run_bzr(['checkout', '--lightweight', 'local', 'checkout'])
737
 
        # Make a change and commit it
738
 
        self.build_tree_contents([('local/file', 'modified in local')])
739
 
        self.tree.commit('modify file', rev_id='modified-in-local')
740
 
        # Exercise commands from the checkout directory
741
 
        self._default_wd = 'checkout'
742
 
        self._default_errors = ["Working tree is out of date, please run"
743
 
                                " 'bzr update'\.",]
744
 
        self._default_pushed_revid = 'modified-in-local'
745
 
 
746
 
    def test_push_default(self):
747
 
        self.assertPushFails([])
748
 
 
749
 
    def test_push_with_revision(self):
750
 
        self.assertPushSucceeds(['-r', 'revid:added'], pushed_revid='added')
751
 
 
752
 
    def test_push_no_strict(self):
753
 
        self.assertPushSucceeds(['--no-strict'])
754
 
 
755
 
    def test_push_strict_with_changes(self):
756
 
        self.assertPushFails(['--strict'])
757
 
 
758
 
    def test_push_respect_config_var_strict(self):
759
 
        self.set_config_push_strict('true')
760
 
        self.assertPushFails([])
761
 
 
762
 
    def test_push_bogus_config_var_ignored(self):
763
 
        self.set_config_push_strict("I don't want you to be strict")
764
 
        self.assertPushFails([])
765
 
 
766
 
    def test_push_no_strict_command_line_override_config(self):
767
 
        self.set_config_push_strict('yES')
768
 
        self.assertPushFails([])
769
 
        self.assertPushSucceeds(['--no-strict'])
770
 
 
771
 
    def test_push_strict_command_line_override_config(self):
772
 
        self.set_config_push_strict('oFF')
773
 
        self.assertPushFails(['--strict'])
774
 
        self.assertPushSucceeds([])
775
 
 
776
 
 
777
 
class TestPushForeign(blackbox.ExternalBase):
778
 
 
779
 
    def setUp(self):
780
 
        super(TestPushForeign, self).setUp()
781
 
        test_foreign.register_dummy_foreign_for_test(self)
782
 
 
783
 
    def make_dummy_builder(self, relpath):
784
 
        builder = self.make_branch_builder(
785
 
            relpath, format=test_foreign.DummyForeignVcsDirFormat())
786
 
        builder.build_snapshot('revid', None,
787
 
            [('add', ('', 'TREE_ROOT', 'directory', None)),
788
 
             ('add', ('foo', 'fooid', 'file', 'bar'))])
789
 
        return builder
790
 
 
791
 
    def test_no_roundtripping(self):
792
 
        target_branch = self.make_dummy_builder('dp').get_branch()
793
 
        source_tree = self.make_branch_and_tree("dc")
794
 
        output, error = self.run_bzr("push -d dc dp", retcode=3)
795
 
        self.assertEquals("", output)
796
 
        self.assertEquals(error, "bzr: ERROR: It is not possible to losslessly"
797
 
            " push to dummy. You may want to use dpush instead.\n")