1
# Copyright (C) 2005, 2007 Canonical Ltd
 
 
3
# This program is free software; you can redistribute it and/or modify
 
 
4
# it under the terms of the GNU General Public License as published by
 
 
5
# the Free Software Foundation; either version 2 of the License, or
 
 
6
# (at your option) any later version.
 
 
8
# This program is distributed in the hope that it will be useful,
 
 
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
 
 
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 
 
11
# GNU General Public License for more details.
 
 
13
# You should have received a copy of the GNU General Public License
 
 
14
# along with this program; if not, write to the Free Software
 
 
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 
 
18
"""Black-box tests for bzr push."""
 
 
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.transport import register_transport, unregister_transport
 
 
33
from bzrlib.transport.memory import MemoryServer, MemoryTransport
 
 
34
from bzrlib.uncommit import uncommit
 
 
35
from bzrlib.urlutils import local_path_from_url
 
 
36
from bzrlib.workingtree import WorkingTree
 
 
39
class TestPush(ExternalBase):
 
 
41
    def test_push_remember(self):
 
 
42
        """Push changes from one branch to another and test push location."""
 
 
43
        transport = self.get_transport()
 
 
44
        tree_a = self.make_branch_and_tree('branch_a')
 
 
45
        branch_a = tree_a.branch
 
 
46
        self.build_tree(['branch_a/a'])
 
 
48
        tree_a.commit('commit a')
 
 
49
        tree_b = branch_a.bzrdir.sprout('branch_b').open_workingtree()
 
 
50
        branch_b = tree_b.branch
 
 
51
        tree_c = branch_a.bzrdir.sprout('branch_c').open_workingtree()
 
 
52
        branch_c = tree_c.branch
 
 
53
        self.build_tree(['branch_a/b'])
 
 
55
        tree_a.commit('commit b')
 
 
56
        self.build_tree(['branch_b/c'])
 
 
58
        tree_b.commit('commit c')
 
 
59
        # initial push location must be empty
 
 
60
        self.assertEqual(None, branch_b.get_push_location())
 
 
62
        # test push for failure without push location set
 
 
64
        out = self.run_bzr('push', retcode=3)
 
 
65
        self.assertEquals(out,
 
 
66
                ('','bzr: ERROR: No push location known or specified.\n'))
 
 
68
        # test not remembered if cannot actually push
 
 
69
        self.run_bzr('push ../path/which/doesnt/exist', retcode=3)
 
 
70
        out = self.run_bzr('push', retcode=3)
 
 
72
                ('', 'bzr: ERROR: No push location known or specified.\n'),
 
 
75
        # test implicit --remember when no push location set, push fails
 
 
76
        out = self.run_bzr('push ../branch_b', retcode=3)
 
 
77
        self.assertEquals(out,
 
 
78
                ('','bzr: ERROR: These branches have diverged.  '
 
 
79
                    'Try using "merge" and then "push".\n'))
 
 
80
        self.assertEquals(abspath(branch_a.get_push_location()),
 
 
81
                          abspath(branch_b.bzrdir.root_transport.base))
 
 
83
        # test implicit --remember after resolving previous failure
 
 
84
        uncommit(branch=branch_b, tree=tree_b)
 
 
85
        transport.delete('branch_b/c')
 
 
86
        out, err = self.run_bzr('push')
 
 
87
        path = branch_a.get_push_location()
 
 
88
        self.assertEquals(out,
 
 
89
                          'Using saved location: %s\n' 
 
 
90
                          'Pushed up to revision 2.\n'
 
 
91
                          % local_path_from_url(path))
 
 
93
                         'All changes applied successfully.\n')
 
 
94
        self.assertEqual(path,
 
 
95
                         branch_b.bzrdir.root_transport.base)
 
 
96
        # test explicit --remember
 
 
97
        self.run_bzr('push ../branch_c --remember')
 
 
98
        self.assertEquals(branch_a.get_push_location(),
 
 
99
                          branch_c.bzrdir.root_transport.base)
 
 
101
    def test_push_without_tree(self):
 
 
102
        # bzr push from a branch that does not have a checkout should work.
 
 
103
        b = self.make_branch('.')
 
 
104
        out, err = self.run_bzr('push pushed-location')
 
 
105
        self.assertEqual('', out)
 
 
106
        self.assertEqual('Created new branch.\n', err)
 
 
107
        b2 = Branch.open('pushed-location')
 
 
108
        self.assertEndsWith(b2.base, 'pushed-location/')
 
 
110
    def test_push_new_branch_revision_count(self):
 
 
111
        # bzr push of a branch with revisions to a new location 
 
 
112
        # should print the number of revisions equal to the length of the 
 
 
114
        t = self.make_branch_and_tree('tree')
 
 
115
        self.build_tree(['tree/file'])
 
 
119
        out, err = self.run_bzr('push pushed-to')
 
 
121
        self.assertEqual('', out)
 
 
122
        self.assertEqual('Created new branch.\n', err)
 
 
124
    def test_push_only_pushes_history(self):
 
 
125
        # Knit branches should only push the history for the current revision.
 
 
126
        format = BzrDirMetaFormat1()
 
 
127
        format.repository_format = RepositoryFormatKnit1()
 
 
128
        shared_repo = self.make_repository('repo', format=format, shared=True)
 
 
129
        shared_repo.set_make_working_trees(True)
 
 
131
        def make_shared_tree(path):
 
 
132
            shared_repo.bzrdir.root_transport.mkdir(path)
 
 
133
            shared_repo.bzrdir.create_branch_convenience('repo/' + path)
 
 
134
            return WorkingTree.open('repo/' + path)
 
 
135
        tree_a = make_shared_tree('a')
 
 
136
        self.build_tree(['repo/a/file'])
 
 
138
        tree_a.commit('commit a-1', rev_id='a-1')
 
 
139
        f = open('repo/a/file', 'ab')
 
 
140
        f.write('more stuff\n')
 
 
142
        tree_a.commit('commit a-2', rev_id='a-2')
 
 
144
        tree_b = make_shared_tree('b')
 
 
145
        self.build_tree(['repo/b/file'])
 
 
147
        tree_b.commit('commit b-1', rev_id='b-1')
 
 
149
        self.assertTrue(shared_repo.has_revision('a-1'))
 
 
150
        self.assertTrue(shared_repo.has_revision('a-2'))
 
 
151
        self.assertTrue(shared_repo.has_revision('b-1'))
 
 
153
        # Now that we have a repository with shared files, make sure
 
 
154
        # that things aren't copied out by a 'push'
 
 
156
        self.run_bzr('push ../../push-b')
 
 
157
        pushed_tree = WorkingTree.open('../../push-b')
 
 
158
        pushed_repo = pushed_tree.branch.repository
 
 
159
        self.assertFalse(pushed_repo.has_revision('a-1'))
 
 
160
        self.assertFalse(pushed_repo.has_revision('a-2'))
 
 
161
        self.assertTrue(pushed_repo.has_revision('b-1'))
 
 
163
    def test_push_funky_id(self):
 
 
164
        t = self.make_branch_and_tree('tree')
 
 
166
        self.build_tree(['filename'])
 
 
167
        t.add('filename', 'funky-chars<>%&;"\'')
 
 
168
        t.commit('commit filename')
 
 
169
        self.run_bzr('push ../new-tree')
 
 
171
    def test_push_dash_d(self):
 
 
172
        t = self.make_branch_and_tree('from')
 
 
173
        t.commit(allow_pointless=True,
 
 
174
                message='first commit')
 
 
175
        self.run_bzr('push -d from to-one')
 
 
176
        self.failUnlessExists('to-one')
 
 
177
        self.run_bzr('push -d %s %s' 
 
 
178
            % tuple(map(urlutils.local_path_to_url, ['from', 'to-two'])))
 
 
179
        self.failUnlessExists('to-two')
 
 
181
    def create_simple_tree(self):
 
 
182
        tree = self.make_branch_and_tree('tree')
 
 
183
        self.build_tree(['tree/a'])
 
 
184
        tree.add(['a'], ['a-id'])
 
 
185
        tree.commit('one', rev_id='r1')
 
 
188
    def test_push_create_prefix(self):
 
 
189
        """'bzr push --create-prefix' will create leading directories."""
 
 
190
        tree = self.create_simple_tree()
 
 
192
        self.run_bzr_error(['Parent directory of ../new/tree does not exist'],
 
 
195
        self.run_bzr('push ../new/tree --create-prefix',
 
 
197
        new_tree = WorkingTree.open('new/tree')
 
 
198
        self.assertEqual(tree.last_revision(), new_tree.last_revision())
 
 
199
        self.failUnlessExists('new/tree/a')
 
 
201
    def test_push_use_existing(self):
 
 
202
        """'bzr push --use-existing-dir' can push into an existing dir.
 
 
204
        By default, 'bzr push' will not use an existing, non-versioned dir.
 
 
206
        tree = self.create_simple_tree()
 
 
207
        self.build_tree(['target/'])
 
 
209
        self.run_bzr_error(['Target directory ../target already exists',
 
 
210
                            'Supply --use-existing-dir',
 
 
212
                           'push ../target', working_dir='tree')
 
 
214
        self.run_bzr('push --use-existing-dir ../target',
 
 
217
        new_tree = WorkingTree.open('target')
 
 
218
        self.assertEqual(tree.last_revision(), new_tree.last_revision())
 
 
219
        # The push should have created target/a
 
 
220
        self.failUnlessExists('target/a')
 
 
222
    def test_push_onto_repo(self):
 
 
223
        """We should be able to 'bzr push' into an existing bzrdir."""
 
 
224
        tree = self.create_simple_tree()
 
 
225
        repo = self.make_repository('repo', shared=True)
 
 
227
        self.run_bzr('push ../repo',
 
 
230
        # Pushing onto an existing bzrdir will create a repository and
 
 
231
        # branch as needed, but will only create a working tree if there was
 
 
233
        self.assertRaises(errors.NoWorkingTree, WorkingTree.open, 'repo')
 
 
234
        new_branch = Branch.open('repo')
 
 
235
        self.assertEqual(tree.last_revision(), new_branch.last_revision())
 
 
237
    def test_push_onto_just_bzrdir(self):
 
 
238
        """We don't handle when the target is just a bzrdir.
 
 
240
        Because you shouldn't be able to create *just* a bzrdir in the wild.
 
 
242
        # TODO: jam 20070109 Maybe it would be better to create the repository
 
 
244
        tree = self.create_simple_tree()
 
 
245
        a_bzrdir = self.make_bzrdir('dir')
 
 
247
        self.run_bzr_error(['At ../dir you have a valid .bzr control'],
 
 
251
    def test_push_with_revisionspec(self):
 
 
252
        """We should be able to push a revision older than the tip."""
 
 
253
        tree_from = self.make_branch_and_tree('from')
 
 
254
        tree_from.commit("One.", rev_id="from-1")
 
 
255
        tree_from.commit("Two.", rev_id="from-2")
 
 
257
        self.run_bzr('push -r1 ../to', working_dir='from')
 
 
259
        tree_to = WorkingTree.open('to')
 
 
260
        repo_to = tree_to.branch.repository
 
 
261
        self.assertTrue(repo_to.has_revision('from-1'))
 
 
262
        self.assertFalse(repo_to.has_revision('from-2'))
 
 
263
        self.assertEqual(tree_to.branch.last_revision_info()[1], 'from-1')
 
 
266
            "bzr: ERROR: bzr push --revision takes one value.\n",
 
 
267
            'push -r0..2 ../to', working_dir='from')
 
 
270
class RedirectingMemoryTransport(MemoryTransport):
 
 
272
    def mkdir(self, path, mode=None):
 
 
273
        path = self.abspath(path)[len(self._scheme):]
 
 
274
        if path == '/source':
 
 
275
            raise errors.RedirectRequested(
 
 
276
                path, self._scheme + '/target', is_permanent=True)
 
 
277
        elif path == '/infinite-loop':
 
 
278
            raise errors.RedirectRequested(
 
 
279
                path, self._scheme + '/infinite-loop', is_permanent=True)
 
 
281
            return super(RedirectingMemoryTransport, self).mkdir(
 
 
285
class RedirectingMemoryServer(MemoryServer):
 
 
288
        self._dirs = {'/': None}
 
 
291
        self._scheme = 'redirecting-memory+%s:///' % id(self)
 
 
292
        register_transport(self._scheme, self._memory_factory)
 
 
294
    def _memory_factory(self, url):
 
 
295
        result = RedirectingMemoryTransport(url)
 
 
296
        result._dirs = self._dirs
 
 
297
        result._files = self._files
 
 
298
        result._locks = self._locks
 
 
302
        unregister_transport(self._scheme, self._memory_factory)
 
 
305
class TestPushRedirect(ExternalBase):
 
 
308
        ExternalBase.setUp(self)
 
 
309
        self.memory_server = RedirectingMemoryServer()
 
 
310
        self.memory_server.setUp()
 
 
311
        self.addCleanup(self.memory_server.tearDown)
 
 
313
        # Make the branch and tree that we'll be pushing.
 
 
314
        t = self.make_branch_and_tree('tree')
 
 
315
        self.build_tree(['tree/file'])
 
 
319
    def test_push_redirects_on_mkdir(self):
 
 
320
        """If the push requires a mkdir, push respects redirect requests.
 
 
322
        This is added primarily to handle lp:/ URI support, so that users can
 
 
323
        push to new branches by specifying lp:/ URIs.
 
 
326
        destination_url = self.memory_server.get_url() + 'source'
 
 
327
        self.run_bzr('push %s' % destination_url)
 
 
330
        local_revision = Branch.open('tree').last_revision()
 
 
331
        remote_revision = Branch.open(
 
 
332
            self.memory_server.get_url() + 'target').last_revision()
 
 
333
        self.assertEqual(remote_revision, local_revision)
 
 
335
    def test_push_gracefully_handles_too_many_redirects(self):
 
 
336
        """Push fails gracefully if the mkdir generates a large number of
 
 
340
        destination_url = self.memory_server.get_url() + 'infinite-loop'
 
 
341
        out, err = self.run_bzr_error(
 
 
342
            ['Too many redirections trying to make %s\\.\n'
 
 
343
             % re.escape(destination_url)],
 
 
344
            'push %s' % destination_url, retcode=3)
 
 
346
        self.assertEqual('', out)