1
# Copyright (C) 2005, 2006, 2007, 2009, 2010 Robey Pointer <robey@lag.net>, 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
17
"""Tests for branches bound to an sftp branch."""
28
from bzrlib.tests import test_server
29
from bzrlib.transport import memory
32
class BoundSFTPBranch(tests.TestCaseWithTransport):
35
tests.TestCaseWithTransport.setUp(self)
36
self.vfs_transport_factory = memory.MemoryServer
37
if self.transport_server is test_server.LocalURLServer:
38
self.transport_server = None
40
def create_branches(self):
41
self.build_tree(['base/', 'base/a', 'base/b'])
42
format = bzrdir.format_registry.make_bzrdir('knit')
44
wt_base = bzrdir.BzrDir.create_standalone_workingtree(
45
self.get_url('base'), format=format)
46
except errors.NotLocalUrl:
47
raise tests.TestSkipped('Not a local URL')
49
b_base = wt_base.branch
53
wt_base.commit('first', rev_id='r@b-1')
55
wt_child = b_base.bzrdir.sprout('child').open_workingtree()
56
self.sftp_base = branch.Branch.open(self.get_url('base'))
57
wt_child.branch.bind(self.sftp_base)
58
# check the branch histories are ready for using in tests.
59
self.assertEqual(['r@b-1'], b_base.revision_history())
60
self.assertEqual(['r@b-1'], wt_child.branch.revision_history())
61
return b_base, wt_child
63
def test_simple_binding(self):
64
self.build_tree(['base/', 'base/a', 'base/b', 'child/'])
66
wt_base = bzrdir.BzrDir.create_standalone_workingtree(
68
except errors.NotLocalUrl:
69
raise tests.TestSkipped('Not a local URL')
73
wt_base.commit('first', rev_id='r@b-1')
75
b_base = wt_base.branch
76
# manually make a branch we can bind, because the default format
77
# may not be bindable-from, and we want to test the side effects etc
79
format = bzrdir.format_registry.make_bzrdir('knit')
80
b_child = bzrdir.BzrDir.create_branch_convenience(
81
'child', format=format)
82
self.assertEqual(None, b_child.get_bound_location())
83
self.assertEqual(None, b_child.get_master_branch())
85
sftp_b_base = branch.Branch.open(self.get_url('base'))
86
b_child.bind(sftp_b_base)
87
self.assertEqual(sftp_b_base.base, b_child.get_bound_location())
88
# the bind must not have given b_child history:
89
self.assertEqual([], b_child.revision_history())
90
# we should be able to update the branch at this point:
91
self.assertEqual(None, b_child.update())
92
# and now there must be history.
93
self.assertEqual(['r@b-1'], b_child.revision_history())
94
# this line is more of a working tree test line, but - what the hey,
96
b_child.bzrdir.open_workingtree().update()
97
self.failUnlessExists('child/a')
98
self.failUnlessExists('child/b')
101
self.assertEqual(None, b_child.get_bound_location())
103
def test_bound_commit(self):
104
b_base, wt_child = self.create_branches()
106
open('child/a', 'wb').write('new contents\n')
107
wt_child.commit('modified a', rev_id='r@c-2')
109
self.assertEqual(['r@b-1', 'r@c-2'], wt_child.branch.revision_history())
110
self.assertEqual(['r@b-1', 'r@c-2'], b_base.revision_history())
112
def test_bound_commit_fails_when_out_of_date(self):
113
# Make sure commit fails if out of date.
114
b_base, wt_child = self.create_branches()
116
open('base/a', 'wb').write('new base contents\n')
117
b_base.bzrdir.open_workingtree().commit('base', rev_id='r@b-2')
119
open('child/b', 'wb').write('new b child contents\n')
120
self.assertRaises(errors.BoundBranchOutOfDate,
121
wt_child.commit, 'child', rev_id='r@c-2')
123
sftp_b_base = branch.Branch.open(self.get_url('base'))
125
# This is all that cmd_update does
126
wt_child.pull(sftp_b_base, overwrite=False)
128
wt_child.commit('child', rev_id='r@c-3')
130
self.assertEqual(['r@b-1', 'r@b-2', 'r@c-3'],
131
wt_child.branch.revision_history())
132
self.assertEqual(['r@b-1', 'r@b-2', 'r@c-3'],
133
b_base.revision_history())
134
self.assertEqual(['r@b-1', 'r@b-2', 'r@c-3'],
135
sftp_b_base.revision_history())
137
def test_double_binding(self):
138
b_base, wt_child = self.create_branches()
140
wt_child2 = wt_child.branch.create_checkout('child2')
142
open('child2/a', 'wb').write('new contents\n')
143
self.assertRaises(errors.CommitToDoubleBoundBranch,
144
wt_child2.commit, 'child2', rev_id='r@d-2')
146
def test_unbinding(self):
147
from bzrlib import transport
148
b_base, wt_child = self.create_branches()
150
# TestCaseWithSFTPServer only allows you to connect one time
151
# to the SFTP server. So we have to create a connection and
152
# keep it around, so that it can be reused
153
__unused_t = transport.get_transport(self.get_url('.'))
155
wt_base = b_base.bzrdir.open_workingtree()
156
open('base/a', 'wb').write('new base contents\n')
157
wt_base.commit('base', rev_id='r@b-2')
159
open('child/b', 'wb').write('new b child contents\n')
160
self.assertRaises(errors.BoundBranchOutOfDate,
161
wt_child.commit, 'child', rev_id='r@c-2')
162
self.assertEqual(['r@b-1'], wt_child.branch.revision_history())
163
wt_child.branch.unbind()
164
wt_child.commit('child', rev_id='r@c-2')
165
self.assertEqual(['r@b-1', 'r@c-2'], wt_child.branch.revision_history())
166
self.assertEqual(['r@b-1', 'r@b-2'], b_base.revision_history())
168
sftp_b_base = branch.Branch.open(self.get_url('base'))
169
self.assertRaises(errors.DivergedBranches,
170
wt_child.branch.bind, sftp_b_base)
172
def test_commit_remote_bound(self):
173
# Make sure it is detected if the current base is bound during the
174
# objects lifetime, when the child goes to commit.
175
b_base, wt_child = self.create_branches()
177
b_base.bzrdir.sprout('newbase')
179
sftp_b_base = branch.Branch.open(self.get_url('base'))
180
sftp_b_newbase = branch.Branch.open(self.get_url('newbase'))
182
sftp_b_base.bind(sftp_b_newbase)
184
open('child/a', 'wb').write('new contents\n')
185
self.assertRaises(errors.CommitToDoubleBoundBranch,
186
wt_child.commit, 'failure', rev_id='r@c-2')
188
self.assertEqual(['r@b-1'], b_base.revision_history())
189
self.assertEqual(['r@b-1'], wt_child.branch.revision_history())
190
self.assertEqual(['r@b-1'], sftp_b_newbase.revision_history())
192
def test_bind_diverged(self):
193
b_base, wt_child = self.create_branches()
195
wt_child.branch.unbind()
196
open('child/a', 'ab').write('child contents\n')
197
wt_child_rev = wt_child.commit('child', rev_id='r@c-2')
199
self.assertEqual(['r@b-1', 'r@c-2'], wt_child.branch.revision_history())
200
self.assertEqual(['r@b-1'], b_base.revision_history())
202
open('base/b', 'ab').write('base contents\n')
203
b_base.bzrdir.open_workingtree().commit('base', rev_id='r@b-2')
204
self.assertEqual(['r@b-1', 'r@b-2'], b_base.revision_history())
206
sftp_b_base = branch.Branch.open(self.get_url('base'))
208
self.assertRaises(errors.DivergedBranches,
209
wt_child.branch.bind, sftp_b_base)
211
wt_child.merge_from_branch(sftp_b_base)
212
self.assertEqual([wt_child_rev, 'r@b-2'], wt_child.get_parent_ids())
213
wt_child.commit('merged', rev_id='r@c-3')
215
# After a merge, trying to bind again should succeed but not push the
217
wt_child.branch.bind(sftp_b_base)
219
self.assertEqual(['r@b-1', 'r@b-2'], b_base.revision_history())
220
self.assertEqual(['r@b-1', 'r@c-2', 'r@c-3'],
221
wt_child.branch.revision_history())
223
def test_bind_parent_ahead_preserves_parent(self):
224
b_base, wt_child = self.create_branches()
226
wt_child.branch.unbind()
228
open('a', 'ab').write('base changes\n')
229
wt_base = b_base.bzrdir.open_workingtree()
230
wt_base.commit('base', rev_id='r@b-2')
231
self.assertEqual(['r@b-1', 'r@b-2'], b_base.revision_history())
232
self.assertEqual(['r@b-1'], wt_child.branch.revision_history())
234
sftp_b_base = branch.Branch.open(self.get_url('base'))
235
wt_child.branch.bind(sftp_b_base)
237
self.assertEqual(['r@b-1'], wt_child.branch.revision_history())
239
wt_child.branch.unbind()
241
# Check and make sure it also works if parent is ahead multiple
242
wt_base.commit('base 3', rev_id='r@b-3', allow_pointless=True)
243
wt_base.commit('base 4', rev_id='r@b-4', allow_pointless=True)
244
wt_base.commit('base 5', rev_id='r@b-5', allow_pointless=True)
246
self.assertEqual(['r@b-1', 'r@b-2', 'r@b-3', 'r@b-4', 'r@b-5'],
247
b_base.revision_history())
249
self.assertEqual(['r@b-1'], wt_child.branch.revision_history())
251
wt_child.branch.bind(sftp_b_base)
252
self.assertEqual(['r@b-1'], wt_child.branch.revision_history())
254
def test_bind_child_ahead_preserves_child(self):
255
b_base, wt_child = self.create_branches()
257
wt_child.branch.unbind()
259
wt_child.commit('child', rev_id='r@c-2', allow_pointless=True)
260
self.assertEqual(['r@b-1', 'r@c-2'], wt_child.branch.revision_history())
261
self.assertEqual(['r@b-1'], b_base.revision_history())
263
sftp_b_base = branch.Branch.open(self.get_url('base'))
264
wt_child.branch.bind(sftp_b_base)
266
self.assertEqual(['r@b-1'], b_base.revision_history())
268
# Check and make sure it also works if child is ahead multiple
269
wt_child.branch.unbind()
270
wt_child.commit('child 3', rev_id='r@c-3', allow_pointless=True)
271
wt_child.commit('child 4', rev_id='r@c-4', allow_pointless=True)
272
wt_child.commit('child 5', rev_id='r@c-5', allow_pointless=True)
274
self.assertEqual(['r@b-1', 'r@c-2', 'r@c-3', 'r@c-4', 'r@c-5'],
275
wt_child.branch.revision_history())
276
self.assertEqual(['r@b-1'], b_base.revision_history())
278
wt_child.branch.bind(sftp_b_base)
279
self.assertEqual(['r@b-1'], b_base.revision_history())
281
def test_commit_after_merge(self):
282
b_base, wt_child = self.create_branches()
284
# We want merge to be able to be a local only
285
# operation, because it does not alter the branch data.
287
# But we can't fail afterwards
289
wt_other = wt_child.bzrdir.sprout('other').open_workingtree()
291
open('other/c', 'wb').write('file c\n')
293
wt_other.commit('adding c', rev_id='r@d-2')
295
self.failIf(wt_child.branch.repository.has_revision('r@d-2'))
296
self.failIf(b_base.repository.has_revision('r@d-2'))
298
wt_child.merge_from_branch(wt_other.branch)
300
self.failUnlessExists('child/c')
301
self.assertEqual(['r@d-2'], wt_child.get_parent_ids()[1:])
302
self.failUnless(wt_child.branch.repository.has_revision('r@d-2'))
303
self.failIf(b_base.repository.has_revision('r@d-2'))
305
# Commit should succeed, and cause merged revisions to
306
# be pushed into base
307
wt_child.commit('merge other', rev_id='r@c-2')
308
self.assertEqual(['r@b-1', 'r@c-2'], wt_child.branch.revision_history())
309
self.assertEqual(['r@b-1', 'r@c-2'], b_base.revision_history())
310
self.failUnless(b_base.repository.has_revision('r@d-2'))
312
def test_commit_fails(self):
313
b_base, wt_child = self.create_branches()
315
open('a', 'ab').write('child adds some text\n')
317
# this deletes the branch from memory
319
# and this moves it out of the way on disk
320
os.rename('base', 'hidden_base')
322
self.assertRaises(errors.BoundBranchConnectionFailure,
323
wt_child.commit, 'added text', rev_id='r@c-2')
325
# TODO: jam 20051231 We need invasive failure tests, so that we can show
326
# performance even when something fails.
added
removed
bzrlib/tests/per_branch/test_bound_sftp.py
