3244
3244
_, stacked = branch_factory()
3245
3245
source = stacked.repository._get_source(target_repository_format)
3246
3246
tip = stacked.last_revision()
3247
revs = stacked.repository.get_ancestry(tip)
3248
search = graph.PendingAncestryResult([tip], stacked.repository)
3247
stacked.repository._ensure_real()
3248
graph = stacked.repository.get_graph()
3249
revs = [r for (r,ps) in graph.iter_ancestry([tip])
3250
if r != NULL_REVISION]
3252
search = _mod_graph.PendingAncestryResult([tip], stacked.repository)
3249
3253
self.reset_smart_call_log()
3250
3254
stream = source.get_stream(search)
3253
3255
# We trust that if a revision is in the stream the rest of the new
3254
3256
# content for it is too, as per our main fetch tests; here we are
3255
3257
# checking that the revisions are actually included at all, and their
3294
3296
self.assertEqual(expected_revs, rev_ord)
3295
3297
# Getting topological sort requires VFS calls still - one of which is
3296
3298
# pushing up from the bound branch.
3297
self.assertLength(13, self.hpss_calls)
3299
self.assertLength(14, self.hpss_calls)
3299
3301
def test_stacked_get_stream_groupcompress(self):
3300
3302
# Repository._get_source.get_stream() from a stacked repository with
3357
3359
remote_branch_url = self.smart_server.get_url() + 'remote'
3358
3360
remote_branch = bzrdir.BzrDir.open(remote_branch_url).open_branch()
3359
3361
self.hpss_calls = []
3360
local.repository.fetch(remote_branch.repository,
3361
fetch_spec=graph.EverythingResult(remote_branch.repository))
3362
local.repository.fetch(
3363
remote_branch.repository,
3364
fetch_spec=_mod_graph.EverythingResult(remote_branch.repository))
3362
3365
self.assertEqual(['Repository.get_stream_1.19'], self.hpss_calls)
3364
3367
def override_verb(self, verb_name, verb):
3379
3382
"""A version of the Repository.get_stream_1.19 verb patched to
3380
3383
reject 'everything' searches the way 2.3 and earlier do.
3382
def recreate_search(self, repository, search_bytes, discard_excess=False):
3385
def recreate_search(self, repository, search_bytes,
3386
discard_excess=False):
3383
3387
verb_log.append(search_bytes.split('\n', 1)[0])
3384
3388
if search_bytes == 'everything':
3385
return (None, request.FailedSmartServerResponse(('BadSearch',)))
3390
request.FailedSmartServerResponse(('BadSearch',)))
3386
3391
return super(OldGetStreamVerb,
3387
3392
self).recreate_search(repository, search_bytes,
3388
3393
discard_excess=discard_excess)
3393
3398
remote_branch_url = self.smart_server.get_url() + 'remote'
3394
3399
remote_branch = bzrdir.BzrDir.open(remote_branch_url).open_branch()
3395
3400
self.hpss_calls = []
3396
local.repository.fetch(remote_branch.repository,
3397
fetch_spec=graph.EverythingResult(remote_branch.repository))
3401
local.repository.fetch(
3402
remote_branch.repository,
3403
fetch_spec=_mod_graph.EverythingResult(remote_branch.repository))
3398
3404
# make sure the overridden verb was used
3399
3405
self.assertLength(1, verb_log)
3400
3406
# more than one HPSS call is needed, but because it's a VFS callback
3401
3407
# its hard to predict exactly how many.
3402
3408
self.assertTrue(len(self.hpss_calls) > 1)
3411
class TestUpdateBoundBranchWithModifiedBoundLocation(
3412
tests.TestCaseWithTransport):
3413
"""Ensure correct handling of bound_location modifications.
3415
This is tested against a smart server as http://pad.lv/786980 was about a
3416
ReadOnlyError (write attempt during a read-only transaction) which can only
3417
happen in this context.
3421
super(TestUpdateBoundBranchWithModifiedBoundLocation, self).setUp()
3422
self.transport_server = test_server.SmartTCPServer_for_testing
3424
def make_master_and_checkout(self, master_name, checkout_name):
3425
# Create the master branch and its associated checkout
3426
self.master = self.make_branch_and_tree(master_name)
3427
self.checkout = self.master.branch.create_checkout(checkout_name)
3428
# Modify the master branch so there is something to update
3429
self.master.commit('add stuff')
3430
self.last_revid = self.master.commit('even more stuff')
3431
self.bound_location = self.checkout.branch.get_bound_location()
3433
def assertUpdateSucceeds(self, new_location):
3434
self.checkout.branch.set_bound_location(new_location)
3435
self.checkout.update()
3436
self.assertEquals(self.last_revid, self.checkout.last_revision())
3438
def test_without_final_slash(self):
3439
self.make_master_and_checkout('master', 'checkout')
3440
# For unclear reasons some users have a bound_location without a final
3441
# '/', simulate that by forcing such a value
3442
self.assertEndsWith(self.bound_location, '/')
3443
self.assertUpdateSucceeds(self.bound_location.rstrip('/'))
3445
def test_plus_sign(self):
3446
self.make_master_and_checkout('+master', 'checkout')
3447
self.assertUpdateSucceeds(self.bound_location.replace('%2B', '+', 1))
3449
def test_tilda(self):
3450
# Embed ~ in the middle of the path just to avoid any $HOME
3452
self.make_master_and_checkout('mas~ter', 'checkout')
3453
self.assertUpdateSucceeds(self.bound_location.replace('%2E', '~', 1))
added
removed
bzrlib/tests/test_remote.py
