39
39
from bzrlib.branch import Branch, BranchReferenceFormat
40
40
import bzrlib.smart.branch
41
import bzrlib.smart.bzrdir
41
import bzrlib.smart.bzrdir, bzrlib.smart.bzrdir as smart_dir
42
import bzrlib.smart.packrepository
42
43
import bzrlib.smart.repository
43
44
from bzrlib.smart.request import (
44
45
FailedSmartServerResponse,
59
58
"""Multiply tests version and protocol consistency."""
60
59
# FindRepository tests.
61
60
bzrdir_mod = bzrlib.smart.bzrdir
62
applier = TestScenarioApplier()
64
62
("find_repository", {
65
63
"_request_class":bzrdir_mod.SmartServerRequestFindRepositoryV1}),
66
64
("find_repositoryV2", {
67
65
"_request_class":bzrdir_mod.SmartServerRequestFindRepositoryV2}),
66
("find_repositoryV3", {
67
"_request_class":bzrdir_mod.SmartServerRequestFindRepositoryV3}),
69
69
to_adapt, result = split_suite_by_re(standard_tests,
70
70
"TestSmartServerRequestFindRepository")
71
71
v2_only, v1_and_2 = split_suite_by_re(to_adapt,
73
for test in iter_suite_tests(v1_and_2):
74
result.addTests(applier.adapt(test))
75
del applier.scenarios[0]
76
for test in iter_suite_tests(v2_only):
77
result.addTests(applier.adapt(test))
73
tests.multiply_tests(v1_and_2, scenarios, result)
74
# The first scenario is only applicable to v1 protocols, it is deleted
76
tests.multiply_tests(v2_only, scenarios[1:], result)
160
159
request.transport_from_client_path('foo/').base)
162
class TestSmartServerBzrDirRequestCloningMetaDir(
163
tests.TestCaseWithMemoryTransport):
164
"""Tests for BzrDir.cloning_metadir."""
166
def test_cloning_metadir(self):
167
"""When there is a bzrdir present, the call succeeds."""
168
backing = self.get_transport()
169
dir = self.make_bzrdir('.')
170
local_result = dir.cloning_metadir()
171
request_class = smart_dir.SmartServerBzrDirRequestCloningMetaDir
172
request = request_class(backing)
173
expected = SuccessfulSmartServerResponse(
174
(local_result.network_name(),
175
local_result.repository_format.network_name(),
176
('branch', local_result.get_branch_format().network_name())))
177
self.assertEqual(expected, request.execute('', 'False'))
179
def test_cloning_metadir_reference(self):
180
"""The request works when bzrdir contains a branch reference."""
181
backing = self.get_transport()
182
referenced_branch = self.make_branch('referenced')
183
dir = self.make_bzrdir('.')
184
local_result = dir.cloning_metadir()
185
reference = BranchReferenceFormat().initialize(dir, referenced_branch)
186
reference_url = BranchReferenceFormat().get_reference(dir)
187
# The server shouldn't try to follow the branch reference, so it's fine
188
# if the referenced branch isn't reachable.
189
backing.rename('referenced', 'moved')
190
request_class = smart_dir.SmartServerBzrDirRequestCloningMetaDir
191
request = request_class(backing)
192
expected = SuccessfulSmartServerResponse(
193
(local_result.network_name(),
194
local_result.repository_format.network_name(),
195
('ref', reference_url)))
196
self.assertEqual(expected, request.execute('', 'False'))
163
199
class TestSmartServerRequestCreateRepository(tests.TestCaseWithMemoryTransport):
164
200
"""Tests for BzrDir.create_repository."""
217
if (smart.bzrdir.SmartServerRequestFindRepositoryV2 ==
253
if (smart.bzrdir.SmartServerRequestFindRepositoryV3 ==
254
self._request_class):
255
return SuccessfulSmartServerResponse(
256
('ok', '', rich_root, subtrees, 'no',
257
repo._format.network_name()))
258
elif (smart.bzrdir.SmartServerRequestFindRepositoryV2 ==
218
259
self._request_class):
219
260
# All tests so far are on formats, and for non-external
319
360
request.execute('reference'))
363
class TestSmartServerRequestOpenBranchV2(TestCaseWithChrootedTransport):
365
def test_no_branch(self):
366
"""When there is no branch, ('nobranch', ) is returned."""
367
backing = self.get_transport()
368
self.make_bzrdir('.')
369
request = smart.bzrdir.SmartServerRequestOpenBranchV2(backing)
370
self.assertEqual(SmartServerResponse(('nobranch', )),
373
def test_branch(self):
374
"""When there is a branch, 'ok' is returned."""
375
backing = self.get_transport()
376
expected = self.make_branch('.')._format.network_name()
377
request = smart.bzrdir.SmartServerRequestOpenBranchV2(backing)
378
self.assertEqual(SuccessfulSmartServerResponse(('branch', expected)),
381
def test_branch_reference(self):
382
"""When there is a branch reference, the reference URL is returned."""
383
backing = self.get_transport()
384
request = smart.bzrdir.SmartServerRequestOpenBranchV2(backing)
385
branch = self.make_branch('branch')
386
checkout = branch.create_checkout('reference',lightweight=True)
387
reference_url = BranchReferenceFormat().get_reference(checkout.bzrdir)
388
self.assertFileEqual(reference_url, 'reference/.bzr/branch/location')
389
self.assertEqual(SuccessfulSmartServerResponse(('ref', reference_url)),
390
request.execute('reference'))
322
393
class TestSmartServerRequestRevisionHistory(tests.TestCaseWithMemoryTransport):
324
395
def test_empty(self):
632
703
self.assertEqual('child-1', self.tree.branch.last_revision())
706
class TestSmartServerBranchRequestGetParent(tests.TestCaseWithMemoryTransport):
708
def test_get_parent_none(self):
709
base_branch = self.make_branch('base')
710
request = smart.branch.SmartServerBranchGetParent(self.get_transport())
711
response = request.execute('base')
713
SuccessfulSmartServerResponse(('',)), response)
715
def test_get_parent_something(self):
716
base_branch = self.make_branch('base')
717
base_branch.set_parent(self.get_url('foo'))
718
request = smart.branch.SmartServerBranchGetParent(self.get_transport())
719
response = request.execute('base')
721
SuccessfulSmartServerResponse(("../foo",)),
725
class TestSmartServerBranchRequestGetTagsBytes(tests.TestCaseWithMemoryTransport):
726
# Only called when the branch format and tags match [yay factory
727
# methods] so only need to test straight forward cases.
729
def test_get_bytes(self):
730
base_branch = self.make_branch('base')
731
request = smart.branch.SmartServerBranchGetTagsBytes(
732
self.get_transport())
733
response = request.execute('base')
735
SuccessfulSmartServerResponse(('',)), response)
635
738
class TestSmartServerBranchRequestGetStackedOnURL(tests.TestCaseWithMemoryTransport):
637
740
def test_get_stacked_on_url(self):
867
970
request.execute('', 'missingrevision'))
973
class TestSmartServerRepositoryGetStream(tests.TestCaseWithMemoryTransport):
975
def make_two_commit_repo(self):
976
tree = self.make_branch_and_memory_tree('.')
979
r1 = tree.commit('1st commit')
980
r2 = tree.commit('2nd commit', rev_id=u'\xc8'.encode('utf-8'))
982
repo = tree.branch.repository
985
def test_ancestry_of(self):
986
"""The search argument may be a 'ancestry-of' some heads'."""
987
backing = self.get_transport()
988
request = smart.repository.SmartServerRepositoryGetStream(backing)
989
repo, r1, r2 = self.make_two_commit_repo()
990
fetch_spec = ['ancestry-of', r2]
991
lines = '\n'.join(fetch_spec)
992
request.execute('', repo._format.network_name())
993
response = request.do_body(lines)
994
self.assertEqual(('ok',), response.args)
995
stream_bytes = ''.join(response.body_stream)
996
self.assertStartsWith(stream_bytes, 'Bazaar pack format 1')
998
def test_search(self):
999
"""The search argument may be a 'search' of some explicit keys."""
1000
backing = self.get_transport()
1001
request = smart.repository.SmartServerRepositoryGetStream(backing)
1002
repo, r1, r2 = self.make_two_commit_repo()
1003
fetch_spec = ['search', '%s %s' % (r1, r2), 'null:', '2']
1004
lines = '\n'.join(fetch_spec)
1005
request.execute('', repo._format.network_name())
1006
response = request.do_body(lines)
1007
self.assertEqual(('ok',), response.args)
1008
stream_bytes = ''.join(response.body_stream)
1009
self.assertStartsWith(stream_bytes, 'Bazaar pack format 1')
870
1012
class TestSmartServerRequestHasRevision(tests.TestCaseWithMemoryTransport):
872
1014
def test_missing_revision(self):
1142
1284
smart.request.request_handlers.get('Branch.get_config_file'),
1143
1285
smart.branch.SmartServerBranchGetConfigFile)
1144
1286
self.assertEqual(
1287
smart.request.request_handlers.get('Branch.get_parent'),
1288
smart.branch.SmartServerBranchGetParent)
1290
smart.request.request_handlers.get('Branch.get_tags_bytes'),
1291
smart.branch.SmartServerBranchGetTagsBytes)
1145
1293
smart.request.request_handlers.get('Branch.lock_write'),
1146
1294
smart.branch.SmartServerBranchRequestLockWrite)
1147
1295
self.assertEqual(
1169
1317
smart.request.request_handlers.get('BzrDirFormat.initialize'),
1170
1318
smart.bzrdir.SmartServerRequestInitializeBzrDir)
1171
1319
self.assertEqual(
1320
smart.request.request_handlers.get('BzrDir.cloning_metadir'),
1321
smart.bzrdir.SmartServerBzrDirRequestCloningMetaDir)
1172
1323
smart.request.request_handlers.get('BzrDir.open_branch'),
1173
1324
smart.bzrdir.SmartServerRequestOpenBranch)
1174
1325
self.assertEqual(
1326
smart.request.request_handlers.get('BzrDir.open_branchV2'),
1327
smart.bzrdir.SmartServerRequestOpenBranchV2)
1175
1329
smart.request.request_handlers.get('PackRepository.autopack'),
1176
1330
smart.packrepository.SmartServerPackRepositoryAutopack)
1177
1331
self.assertEqual(
1194
1348
smart.request.request_handlers.get('Repository.lock_write'),
1195
1349
smart.repository.SmartServerRepositoryLockWrite)
1196
1350
self.assertEqual(
1351
smart.request.request_handlers.get('Repository.get_stream'),
1352
smart.repository.SmartServerRepositoryGetStream)
1197
1354
smart.request.request_handlers.get('Repository.tarball'),
1198
1355
smart.repository.SmartServerRepositoryTarball)
1199
1356
self.assertEqual(
added
removed
bzrlib/tests/test_smart.py
