~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

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

  • Committer: Robert Collins
  • Date: 2005-08-23 06:52:09 UTC
  • mto: (974.1.50) (1185.1.10) (1092.3.1)
  • mto: This revision was merged to the branch mainline in revision 1139.
  • Revision ID: robertc@robertcollins.net-20050823065209-81cd5962c401751b
move io redirection into each test case from the global runner

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., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
16
 
 
17
 
 
18
 
"""Black-box tests for bzr push."""
19
 
 
20
 
import os
21
 
import re
22
 
 
23
 
from bzrlib import (
24
 
    errors,
25
 
    urlutils,
26
 
    )
27
 
from bzrlib.branch import Branch
28
 
from bzrlib.bzrdir import BzrDirMetaFormat1
29
 
from bzrlib.osutils import abspath
30
 
from bzrlib.repofmt.knitrepo import RepositoryFormatKnit1
31
 
from bzrlib.tests.blackbox import ExternalBase
32
 
from bzrlib.tests.http_server import HttpServer
33
 
from bzrlib.transport import register_transport, unregister_transport
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
38
 
 
39
 
 
40
 
class TestPush(ExternalBase):
41
 
 
42
 
    def test_push_remember(self):
43
 
        """Push changes from one branch to another and test push location."""
44
 
        transport = self.get_transport()
45
 
        tree_a = self.make_branch_and_tree('branch_a')
46
 
        branch_a = tree_a.branch
47
 
        self.build_tree(['branch_a/a'])
48
 
        tree_a.add('a')
49
 
        tree_a.commit('commit a')
50
 
        tree_b = branch_a.bzrdir.sprout('branch_b').open_workingtree()
51
 
        branch_b = tree_b.branch
52
 
        tree_c = branch_a.bzrdir.sprout('branch_c').open_workingtree()
53
 
        branch_c = tree_c.branch
54
 
        self.build_tree(['branch_a/b'])
55
 
        tree_a.add('b')
56
 
        tree_a.commit('commit b')
57
 
        self.build_tree(['branch_b/c'])
58
 
        tree_b.add('c')
59
 
        tree_b.commit('commit c')
60
 
        # initial push location must be empty
61
 
        self.assertEqual(None, branch_b.get_push_location())
62
 
 
63
 
        # test push for failure without push location set
64
 
        os.chdir('branch_a')
65
 
        out = self.run_bzr('push', retcode=3)
66
 
        self.assertEquals(out,
67
 
                ('','bzr: ERROR: No push location known or specified.\n'))
68
 
 
69
 
        # test not remembered if cannot actually push
70
 
        self.run_bzr('push ../path/which/doesnt/exist', retcode=3)
71
 
        out = self.run_bzr('push', retcode=3)
72
 
        self.assertEquals(
73
 
                ('', 'bzr: ERROR: No push location known or specified.\n'),
74
 
                out)
75
 
 
76
 
        # test implicit --remember when no push location set, push fails
77
 
        out = self.run_bzr('push ../branch_b', retcode=3)
78
 
        self.assertEquals(out,
79
 
                ('','bzr: ERROR: These branches have diverged.  '
80
 
                    'Try using "merge" and then "push".\n'))
81
 
        self.assertEquals(abspath(branch_a.get_push_location()),
82
 
                          abspath(branch_b.bzrdir.root_transport.base))
83
 
 
84
 
        # test implicit --remember after resolving previous failure
85
 
        uncommit(branch=branch_b, tree=tree_b)
86
 
        transport.delete('branch_b/c')
87
 
        out, err = self.run_bzr('push')
88
 
        path = branch_a.get_push_location()
89
 
        self.assertEquals(out,
90
 
                          'Using saved push location: %s\n' 
91
 
                          'Pushed up to revision 2.\n'
92
 
                          % local_path_from_url(path))
93
 
        self.assertEqual(err,
94
 
                         'All changes applied successfully.\n')
95
 
        self.assertEqual(path,
96
 
                         branch_b.bzrdir.root_transport.base)
97
 
        # test explicit --remember
98
 
        self.run_bzr('push ../branch_c --remember')
99
 
        self.assertEquals(branch_a.get_push_location(),
100
 
                          branch_c.bzrdir.root_transport.base)
101
 
    
102
 
    def test_push_without_tree(self):
103
 
        # bzr push from a branch that does not have a checkout should work.
104
 
        b = self.make_branch('.')
105
 
        out, err = self.run_bzr('push pushed-location')
106
 
        self.assertEqual('', out)
107
 
        self.assertEqual('Created new branch.\n', err)
108
 
        b2 = Branch.open('pushed-location')
109
 
        self.assertEndsWith(b2.base, 'pushed-location/')
110
 
 
111
 
    def test_push_new_branch_revision_count(self):
112
 
        # bzr push of a branch with revisions to a new location 
113
 
        # should print the number of revisions equal to the length of the 
114
 
        # local branch.
115
 
        t = self.make_branch_and_tree('tree')
116
 
        self.build_tree(['tree/file'])
117
 
        t.add('file')
118
 
        t.commit('commit 1')
119
 
        os.chdir('tree')
120
 
        out, err = self.run_bzr('push pushed-to')
121
 
        os.chdir('..')
122
 
        self.assertEqual('', out)
123
 
        self.assertEqual('Created new branch.\n', err)
124
 
 
125
 
    def test_push_only_pushes_history(self):
126
 
        # Knit branches should only push the history for the current revision.
127
 
        format = BzrDirMetaFormat1()
128
 
        format.repository_format = RepositoryFormatKnit1()
129
 
        shared_repo = self.make_repository('repo', format=format, shared=True)
130
 
        shared_repo.set_make_working_trees(True)
131
 
 
132
 
        def make_shared_tree(path):
133
 
            shared_repo.bzrdir.root_transport.mkdir(path)
134
 
            shared_repo.bzrdir.create_branch_convenience('repo/' + path)
135
 
            return WorkingTree.open('repo/' + path)
136
 
        tree_a = make_shared_tree('a')
137
 
        self.build_tree(['repo/a/file'])
138
 
        tree_a.add('file')
139
 
        tree_a.commit('commit a-1', rev_id='a-1')
140
 
        f = open('repo/a/file', 'ab')
141
 
        f.write('more stuff\n')
142
 
        f.close()
143
 
        tree_a.commit('commit a-2', rev_id='a-2')
144
 
 
145
 
        tree_b = make_shared_tree('b')
146
 
        self.build_tree(['repo/b/file'])
147
 
        tree_b.add('file')
148
 
        tree_b.commit('commit b-1', rev_id='b-1')
149
 
 
150
 
        self.assertTrue(shared_repo.has_revision('a-1'))
151
 
        self.assertTrue(shared_repo.has_revision('a-2'))
152
 
        self.assertTrue(shared_repo.has_revision('b-1'))
153
 
 
154
 
        # Now that we have a repository with shared files, make sure
155
 
        # that things aren't copied out by a 'push'
156
 
        os.chdir('repo/b')
157
 
        self.run_bzr('push ../../push-b')
158
 
        pushed_tree = WorkingTree.open('../../push-b')
159
 
        pushed_repo = pushed_tree.branch.repository
160
 
        self.assertFalse(pushed_repo.has_revision('a-1'))
161
 
        self.assertFalse(pushed_repo.has_revision('a-2'))
162
 
        self.assertTrue(pushed_repo.has_revision('b-1'))
163
 
 
164
 
    def test_push_funky_id(self):
165
 
        t = self.make_branch_and_tree('tree')
166
 
        os.chdir('tree')
167
 
        self.build_tree(['filename'])
168
 
        t.add('filename', 'funky-chars<>%&;"\'')
169
 
        t.commit('commit filename')
170
 
        self.run_bzr('push ../new-tree')
171
 
 
172
 
    def test_push_dash_d(self):
173
 
        t = self.make_branch_and_tree('from')
174
 
        t.commit(allow_pointless=True,
175
 
                message='first commit')
176
 
        self.run_bzr('push -d from to-one')
177
 
        self.failUnlessExists('to-one')
178
 
        self.run_bzr('push -d %s %s' 
179
 
            % tuple(map(urlutils.local_path_to_url, ['from', 'to-two'])))
180
 
        self.failUnlessExists('to-two')
181
 
 
182
 
    def create_simple_tree(self):
183
 
        tree = self.make_branch_and_tree('tree')
184
 
        self.build_tree(['tree/a'])
185
 
        tree.add(['a'], ['a-id'])
186
 
        tree.commit('one', rev_id='r1')
187
 
        return tree
188
 
 
189
 
    def test_push_create_prefix(self):
190
 
        """'bzr push --create-prefix' will create leading directories."""
191
 
        tree = self.create_simple_tree()
192
 
 
193
 
        self.run_bzr_error(['Parent directory of ../new/tree does not exist'],
194
 
                           'push ../new/tree',
195
 
                           working_dir='tree')
196
 
        self.run_bzr('push ../new/tree --create-prefix',
197
 
                     working_dir='tree')
198
 
        new_tree = WorkingTree.open('new/tree')
199
 
        self.assertEqual(tree.last_revision(), new_tree.last_revision())
200
 
        self.failUnlessExists('new/tree/a')
201
 
 
202
 
    def test_push_use_existing(self):
203
 
        """'bzr push --use-existing-dir' can push into an existing dir.
204
 
 
205
 
        By default, 'bzr push' will not use an existing, non-versioned dir.
206
 
        """
207
 
        tree = self.create_simple_tree()
208
 
        self.build_tree(['target/'])
209
 
 
210
 
        self.run_bzr_error(['Target directory ../target already exists',
211
 
                            'Supply --use-existing-dir',
212
 
                           ],
213
 
                           'push ../target', working_dir='tree')
214
 
 
215
 
        self.run_bzr('push --use-existing-dir ../target',
216
 
                     working_dir='tree')
217
 
 
218
 
        new_tree = WorkingTree.open('target')
219
 
        self.assertEqual(tree.last_revision(), new_tree.last_revision())
220
 
        # The push should have created target/a
221
 
        self.failUnlessExists('target/a')
222
 
 
223
 
    def test_push_onto_repo(self):
224
 
        """We should be able to 'bzr push' into an existing bzrdir."""
225
 
        tree = self.create_simple_tree()
226
 
        repo = self.make_repository('repo', shared=True)
227
 
 
228
 
        self.run_bzr('push ../repo',
229
 
                     working_dir='tree')
230
 
 
231
 
        # Pushing onto an existing bzrdir will create a repository and
232
 
        # branch as needed, but will only create a working tree if there was
233
 
        # no BzrDir before.
234
 
        self.assertRaises(errors.NoWorkingTree, WorkingTree.open, 'repo')
235
 
        new_branch = Branch.open('repo')
236
 
        self.assertEqual(tree.last_revision(), new_branch.last_revision())
237
 
 
238
 
    def test_push_onto_just_bzrdir(self):
239
 
        """We don't handle when the target is just a bzrdir.
240
 
 
241
 
        Because you shouldn't be able to create *just* a bzrdir in the wild.
242
 
        """
243
 
        # TODO: jam 20070109 Maybe it would be better to create the repository
244
 
        #       if at this point
245
 
        tree = self.create_simple_tree()
246
 
        a_bzrdir = self.make_bzrdir('dir')
247
 
 
248
 
        self.run_bzr_error(['At ../dir you have a valid .bzr control'],
249
 
                'push ../dir',
250
 
                working_dir='tree')
251
 
 
252
 
    def test_push_with_revisionspec(self):
253
 
        """We should be able to push a revision older than the tip."""
254
 
        tree_from = self.make_branch_and_tree('from')
255
 
        tree_from.commit("One.", rev_id="from-1")
256
 
        tree_from.commit("Two.", rev_id="from-2")
257
 
 
258
 
        self.run_bzr('push -r1 ../to', working_dir='from')
259
 
 
260
 
        tree_to = WorkingTree.open('to')
261
 
        repo_to = tree_to.branch.repository
262
 
        self.assertTrue(repo_to.has_revision('from-1'))
263
 
        self.assertFalse(repo_to.has_revision('from-2'))
264
 
        self.assertEqual(tree_to.branch.last_revision_info()[1], 'from-1')
265
 
 
266
 
        self.run_bzr_error(
267
 
            "bzr: ERROR: bzr push --revision takes one value.\n",
268
 
            'push -r0..2 ../to', working_dir='from')
269
 
 
270
 
    def create_trunk_and_feature_branch(self):
271
 
        # We have a mainline
272
 
        trunk_tree = self.make_branch_and_tree('target',
273
 
            format='development')
274
 
        trunk_tree.commit('mainline')
275
 
        # and a branch from it
276
 
        branch_tree = self.make_branch_and_tree('branch',
277
 
            format='development')
278
 
        branch_tree.pull(trunk_tree.branch)
279
 
        branch_tree.branch.set_parent(trunk_tree.branch.base)
280
 
        # with some work on it
281
 
        branch_tree.commit('moar work plz')
282
 
        return trunk_tree, branch_tree
283
 
 
284
 
    def assertPublished(self, branch_revid, stacked_on):
285
 
        """Assert that the branch 'published' has been published correctly."""
286
 
        published_branch = Branch.open('published')
287
 
        # The published branch refers to the mainline
288
 
        self.assertEqual(stacked_on, published_branch.get_stacked_on_url())
289
 
        # and the branch's work was pushed
290
 
        self.assertTrue(published_branch.repository.has_revision(branch_revid))
291
 
 
292
 
    def test_push_new_branch_stacked_on(self):
293
 
        """Pushing a new branch with --stacked-on creates a stacked branch."""
294
 
        trunk_tree, branch_tree = self.create_trunk_and_feature_branch()
295
 
        # we publish branch_tree with a reference to the mainline.
296
 
        out, err = self.run_bzr(['push', '--stacked-on', trunk_tree.branch.base,
297
 
            self.get_url('published')], working_dir='branch')
298
 
        self.assertEqual('', out)
299
 
        self.assertEqual('Created new stacked branch referring to %s.\n' %
300
 
            trunk_tree.branch.base, err)
301
 
        self.assertPublished(branch_tree.last_revision(),
302
 
            trunk_tree.branch.base)
303
 
 
304
 
    def test_push_new_branch_stacked_uses_parent_when_no_public_url(self):
305
 
        """When the parent has no public url the parent is used as-is."""
306
 
        trunk_tree, branch_tree = self.create_trunk_and_feature_branch()
307
 
        # now we do a stacked push, which should determine the public location
308
 
        # for us.
309
 
        out, err = self.run_bzr(['push', '--stacked',
310
 
            self.get_url('published')], working_dir='branch')
311
 
        self.assertEqual('', out)
312
 
        self.assertEqual('Created new stacked branch referring to %s.\n' %
313
 
            trunk_tree.branch.base, err)
314
 
        self.assertPublished(branch_tree.last_revision(), trunk_tree.branch.base)
315
 
 
316
 
    def test_push_new_branch_stacked_uses_parent_public(self):
317
 
        """Pushing a new branch with --stacked creates a stacked branch."""
318
 
        trunk_tree, branch_tree = self.create_trunk_and_feature_branch()
319
 
        # the trunk is published on a web server
320
 
        self.transport_readonly_server = HttpServer
321
 
        trunk_public = self.make_branch('public_trunk', format='development')
322
 
        trunk_public.pull(trunk_tree.branch)
323
 
        trunk_public_url = self.get_readonly_url('public_trunk')
324
 
        trunk_tree.branch.set_public_branch(trunk_public_url)
325
 
        # now we do a stacked push, which should determine the public location
326
 
        # for us.
327
 
        out, err = self.run_bzr(['push', '--stacked',
328
 
            self.get_url('published')], working_dir='branch')
329
 
        self.assertEqual('', out)
330
 
        self.assertEqual('Created new stacked branch referring to %s.\n' %
331
 
            trunk_public_url, err)
332
 
        self.assertPublished(branch_tree.last_revision(), trunk_public_url)
333
 
 
334
 
    def test_push_new_branch_stacked_no_parent(self):
335
 
        """Pushing with --stacked and no parent branch errors."""
336
 
        branch = self.make_branch_and_tree('branch', format='development')
337
 
        # now we do a stacked push, which should fail as the place to refer too
338
 
        # cannot be determined.
339
 
        out, err = self.run_bzr_error(
340
 
            ['Could not determine branch to refer to\\.'], ['push', '--stacked',
341
 
            self.get_url('published')], working_dir='branch')
342
 
        self.assertEqual('', out)
343
 
        self.assertFalse(self.get_transport('published').has('.'))
344
 
 
345
 
    def test_push_notifies_default_stacking(self):
346
 
        self.make_branch('stack_on', format='1.6')
347
 
        self.make_bzrdir('.').get_config().set_default_stack_on('stack_on')
348
 
        self.make_branch('from', format='1.6')
349
 
        out, err = self.run_bzr('push -d from to')
350
 
        self.assertContainsRe(err,
351
 
                              'Using default stacking branch stack_on at .*')
352
 
 
353
 
 
354
 
class RedirectingMemoryTransport(MemoryTransport):
355
 
 
356
 
    def mkdir(self, path, mode=None):
357
 
        path = self.abspath(path)[len(self._scheme):]
358
 
        if path == '/source':
359
 
            raise errors.RedirectRequested(
360
 
                path, self._scheme + '/target', is_permanent=True)
361
 
        elif path == '/infinite-loop':
362
 
            raise errors.RedirectRequested(
363
 
                path, self._scheme + '/infinite-loop', is_permanent=True)
364
 
        else:
365
 
            return super(RedirectingMemoryTransport, self).mkdir(
366
 
                path, mode)
367
 
 
368
 
 
369
 
class RedirectingMemoryServer(MemoryServer):
370
 
 
371
 
    def setUp(self):
372
 
        self._dirs = {'/': None}
373
 
        self._files = {}
374
 
        self._locks = {}
375
 
        self._scheme = 'redirecting-memory+%s:///' % id(self)
376
 
        register_transport(self._scheme, self._memory_factory)
377
 
 
378
 
    def _memory_factory(self, url):
379
 
        result = RedirectingMemoryTransport(url)
380
 
        result._dirs = self._dirs
381
 
        result._files = self._files
382
 
        result._locks = self._locks
383
 
        return result
384
 
 
385
 
    def tearDown(self):
386
 
        unregister_transport(self._scheme, self._memory_factory)
387
 
 
388
 
 
389
 
class TestPushRedirect(ExternalBase):
390
 
 
391
 
    def setUp(self):
392
 
        ExternalBase.setUp(self)
393
 
        self.memory_server = RedirectingMemoryServer()
394
 
        self.memory_server.setUp()
395
 
        self.addCleanup(self.memory_server.tearDown)
396
 
 
397
 
        # Make the branch and tree that we'll be pushing.
398
 
        t = self.make_branch_and_tree('tree')
399
 
        self.build_tree(['tree/file'])
400
 
        t.add('file')
401
 
        t.commit('commit 1')
402
 
 
403
 
    def test_push_redirects_on_mkdir(self):
404
 
        """If the push requires a mkdir, push respects redirect requests.
405
 
 
406
 
        This is added primarily to handle lp:/ URI support, so that users can
407
 
        push to new branches by specifying lp:/ URIs.
408
 
        """
409
 
        os.chdir('tree')
410
 
        destination_url = self.memory_server.get_url() + 'source'
411
 
        self.run_bzr('push %s' % destination_url)
412
 
        os.chdir('..')
413
 
 
414
 
        local_revision = Branch.open('tree').last_revision()
415
 
        remote_revision = Branch.open(
416
 
            self.memory_server.get_url() + 'target').last_revision()
417
 
        self.assertEqual(remote_revision, local_revision)
418
 
 
419
 
    def test_push_gracefully_handles_too_many_redirects(self):
420
 
        """Push fails gracefully if the mkdir generates a large number of
421
 
        redirects.
422
 
        """
423
 
        os.chdir('tree')
424
 
        destination_url = self.memory_server.get_url() + 'infinite-loop'
425
 
        out, err = self.run_bzr_error(
426
 
            ['Too many redirections trying to make %s\\.\n'
427
 
             % re.escape(destination_url)],
428
 
            'push %s' % destination_url, retcode=3)
429
 
        os.chdir('..')
430
 
        self.assertEqual('', out)