116
117
def test_remote_branch_revision_history(self):
117
118
b = BzrDir.open_from_transport(self.transport).open_branch()
118
self.assertEqual([], b.revision_history())
120
self.applyDeprecated(deprecated_in((2, 5, 0)), b.revision_history))
119
121
r1 = self.local_wt.commit('1st commit')
120
122
r2 = self.local_wt.commit('1st commit', rev_id=u'\xc8'.encode('utf8'))
121
self.assertEqual([r1, r2], b.revision_history())
123
self.assertEqual([r1, r2],
124
self.applyDeprecated(deprecated_in((2, 5, 0)), b.revision_history))
123
126
def test_find_correct_format(self):
124
127
"""Should open a RemoteBzrDir over a RemoteTransport"""
1183
1186
client.add_expected_call(
1184
1187
'Branch.last_revision_info', ('quack/',),
1185
1188
'success', ('ok', '1', 'rev-tip'))
1189
client.add_expected_call(
1190
'Branch.get_config_file', ('quack/',),
1191
'success', ('ok',), '')
1192
transport.mkdir('quack')
1193
transport = transport.clone('quack')
1194
branch = self.make_remote_branch(transport, client)
1195
result = branch.heads_to_fetch()
1196
self.assertFinished(client)
1197
self.assertEqual((set(['rev-tip']), set()), result)
1199
def test_uses_last_revision_info_and_tags_when_set(self):
1200
transport = MemoryTransport()
1201
client = FakeClient(transport.base)
1202
client.add_expected_call(
1203
'Branch.get_stacked_on_url', ('quack/',),
1204
'error', ('NotStacked',))
1205
client.add_expected_call(
1206
'Branch.last_revision_info', ('quack/',),
1207
'success', ('ok', '1', 'rev-tip'))
1208
client.add_expected_call(
1209
'Branch.get_config_file', ('quack/',),
1210
'success', ('ok',), 'branch.fetch_tags = True')
1186
1211
# XXX: this will break if the default format's serialization of tags
1187
1212
# changes, or if the RPC for fetching tags changes from get_tags_bytes.
1188
1213
client.add_expected_call(
1213
1238
self.assertFinished(client)
1214
1239
self.assertEqual((set(['tip']), set(['tagged-1', 'tagged-2'])), result)
1216
def test_backwards_compatible(self):
1241
def make_branch_with_tags(self):
1217
1242
self.setup_smart_server_with_call_log()
1218
1243
# Make a branch with a single revision.
1219
1244
builder = self.make_branch_builder('foo')
1225
1250
# Add two tags to that branch
1226
1251
branch.tags.set_tag('tag-1', 'rev-1')
1227
1252
branch.tags.set_tag('tag-2', 'rev-2')
1255
def test_backwards_compatible(self):
1256
branch = self.make_branch_with_tags()
1257
c = branch.get_config()
1258
c.set_user_option('branch.fetch_tags', 'True')
1228
1259
self.addCleanup(branch.lock_read().unlock)
1229
1260
# Disable the heads_to_fetch verb
1230
1261
verb = 'Branch.heads_to_fetch'
1233
1264
result = branch.heads_to_fetch()
1234
1265
self.assertEqual((set(['tip']), set(['rev-1', 'rev-2'])), result)
1235
1266
self.assertEqual(
1236
['Branch.last_revision_info', 'Branch.get_tags_bytes'],
1267
['Branch.last_revision_info', 'Branch.get_config_file',
1268
'Branch.get_tags_bytes'],
1269
[call.call.method for call in self.hpss_calls])
1271
def test_backwards_compatible_no_tags(self):
1272
branch = self.make_branch_with_tags()
1273
c = branch.get_config()
1274
c.set_user_option('branch.fetch_tags', 'False')
1275
self.addCleanup(branch.lock_read().unlock)
1276
# Disable the heads_to_fetch verb
1277
verb = 'Branch.heads_to_fetch'
1278
self.disable_verb(verb)
1279
self.reset_smart_call_log()
1280
result = branch.heads_to_fetch()
1281
self.assertEqual((set(['tip']), set()), result)
1283
['Branch.last_revision_info', 'Branch.get_config_file'],
1237
1284
[call.call.method for call in self.hpss_calls])
2103
2150
parents = repo.get_parent_map([rev_id])
2104
2151
self.assertEqual(
2105
2152
[('call_with_body_bytes_expecting_body',
2106
'Repository.get_parent_map', ('quack/', 'include-missing:',
2153
'Repository.get_parent_map',
2154
('quack/', 'include-missing:', rev_id), '\n\n0'),
2108
2155
('disconnect medium',),
2109
2156
('call_expecting_body', 'Repository.get_revision_graph',
2110
2157
('quack/', ''))],
2230
2277
self.assertEqual({}, repo.get_parent_map(['non-existant']))
2231
2278
self.assertLength(0, self.hpss_calls)
2280
def test_exposes_get_cached_parent_map(self):
2281
"""RemoteRepository exposes get_cached_parent_map from
2284
r1 = u'\u0e33'.encode('utf8')
2285
r2 = u'\u0dab'.encode('utf8')
2286
lines = [' '.join([r2, r1]), r1]
2287
encoded_body = bz2.compress('\n'.join(lines))
2289
transport_path = 'quack'
2290
repo, client = self.setup_fake_client_and_repository(transport_path)
2291
client.add_success_response_with_body(encoded_body, 'ok')
2293
# get_cached_parent_map should *not* trigger an RPC
2294
self.assertEqual({}, repo.get_cached_parent_map([r1]))
2295
self.assertEqual([], client._calls)
2296
self.assertEqual({r2: (r1,)}, repo.get_parent_map([r2]))
2297
self.assertEqual({r1: (NULL_REVISION,)},
2298
repo.get_cached_parent_map([r1]))
2300
[('call_with_body_bytes_expecting_body',
2301
'Repository.get_parent_map', ('quack/', 'include-missing:', r2),
2234
2307
class TestGetParentMapAllowsNew(tests.TestCaseWithTransport):
3359
3432
remote_branch_url = self.smart_server.get_url() + 'remote'
3360
3433
remote_branch = bzrdir.BzrDir.open(remote_branch_url).open_branch()
3361
3434
self.hpss_calls = []
3362
local.repository.fetch(remote_branch.repository,
3363
fetch_spec=_mod_graph.EverythingResult(remote_branch.repository))
3435
local.repository.fetch(
3436
remote_branch.repository,
3437
fetch_spec=_mod_graph.EverythingResult(remote_branch.repository))
3364
3438
self.assertEqual(['Repository.get_stream_1.19'], self.hpss_calls)
3366
3440
def override_verb(self, verb_name, verb):
3381
3455
"""A version of the Repository.get_stream_1.19 verb patched to
3382
3456
reject 'everything' searches the way 2.3 and earlier do.
3384
def recreate_search(self, repository, search_bytes, discard_excess=False):
3458
def recreate_search(self, repository, search_bytes,
3459
discard_excess=False):
3385
3460
verb_log.append(search_bytes.split('\n', 1)[0])
3386
3461
if search_bytes == 'everything':
3387
return (None, request.FailedSmartServerResponse(('BadSearch',)))
3463
request.FailedSmartServerResponse(('BadSearch',)))
3388
3464
return super(OldGetStreamVerb,
3389
3465
self).recreate_search(repository, search_bytes,
3390
3466
discard_excess=discard_excess)
3395
3471
remote_branch_url = self.smart_server.get_url() + 'remote'
3396
3472
remote_branch = bzrdir.BzrDir.open(remote_branch_url).open_branch()
3397
3473
self.hpss_calls = []
3398
local.repository.fetch(remote_branch.repository,
3399
fetch_spec=_mod_graph.EverythingResult(remote_branch.repository))
3474
local.repository.fetch(
3475
remote_branch.repository,
3476
fetch_spec=_mod_graph.EverythingResult(remote_branch.repository))
3400
3477
# make sure the overridden verb was used
3401
3478
self.assertLength(1, verb_log)
3402
3479
# more than one HPSS call is needed, but because it's a VFS callback
3403
3480
# its hard to predict exactly how many.
3404
3481
self.assertTrue(len(self.hpss_calls) > 1)
3484
class TestUpdateBoundBranchWithModifiedBoundLocation(
3485
tests.TestCaseWithTransport):
3486
"""Ensure correct handling of bound_location modifications.
3488
This is tested against a smart server as http://pad.lv/786980 was about a
3489
ReadOnlyError (write attempt during a read-only transaction) which can only
3490
happen in this context.
3494
super(TestUpdateBoundBranchWithModifiedBoundLocation, self).setUp()
3495
self.transport_server = test_server.SmartTCPServer_for_testing
3497
def make_master_and_checkout(self, master_name, checkout_name):
3498
# Create the master branch and its associated checkout
3499
self.master = self.make_branch_and_tree(master_name)
3500
self.checkout = self.master.branch.create_checkout(checkout_name)
3501
# Modify the master branch so there is something to update
3502
self.master.commit('add stuff')
3503
self.last_revid = self.master.commit('even more stuff')
3504
self.bound_location = self.checkout.branch.get_bound_location()
3506
def assertUpdateSucceeds(self, new_location):
3507
self.checkout.branch.set_bound_location(new_location)
3508
self.checkout.update()
3509
self.assertEquals(self.last_revid, self.checkout.last_revision())
3511
def test_without_final_slash(self):
3512
self.make_master_and_checkout('master', 'checkout')
3513
# For unclear reasons some users have a bound_location without a final
3514
# '/', simulate that by forcing such a value
3515
self.assertEndsWith(self.bound_location, '/')
3516
self.assertUpdateSucceeds(self.bound_location.rstrip('/'))
3518
def test_plus_sign(self):
3519
self.make_master_and_checkout('+master', 'checkout')
3520
self.assertUpdateSucceeds(self.bound_location.replace('%2B', '+', 1))
3522
def test_tilda(self):
3523
# Embed ~ in the middle of the path just to avoid any $HOME
3525
self.make_master_and_checkout('mas~ter', 'checkout')
3526
self.assertUpdateSucceeds(self.bound_location.replace('%2E', '~', 1))
added
removed
bzrlib/tests/test_remote.py
