/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_smart.py

  • Committer: Martin Pool
  • Date: 2009-03-13 07:54:48 UTC
  • mfrom: (4144 +trunk)
  • mto: This revision was merged to the branch mainline in revision 4189.
  • Revision ID: mbp@sourcefrog.net-20090313075448-jlz1t7baz7gzipqn
merge trunk

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_smart.py
38
38
    )
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,
47
48
    SuccessfulSmartServerResponse,
48
49
    )
49
50
from bzrlib.tests import (
50
 
    iter_suite_tests,
51
51
    split_suite_by_re,
52
 
    TestScenarioApplier,
53
52
    )
54
53
from bzrlib.transport import chroot, get_transport
55
54
from bzrlib.util import bencode
59
58
    """Multiply tests version and protocol consistency."""
60
59
    # FindRepository tests.
61
60
    bzrdir_mod = bzrlib.smart.bzrdir
62
 
    applier = TestScenarioApplier()
63
 
    applier.scenarios = [
 
61
    scenarios = [
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}),
68
68
        ]
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,
72
72
        "_v2")
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
 
75
    # since.
 
76
    tests.multiply_tests(v2_only, scenarios[1:], result)
78
77
    return result
79
78
 
80
79
 
160
159
            request.transport_from_client_path('foo/').base)
161
160
 
162
161
 
 
162
class TestSmartServerBzrDirRequestCloningMetaDir(
 
163
    tests.TestCaseWithMemoryTransport):
 
164
    """Tests for BzrDir.cloning_metadir."""
 
165
 
 
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'))
 
178
 
 
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'))
 
197
 
 
198
 
 
199
class TestSmartServerRequestCreateRepository(tests.TestCaseWithMemoryTransport):
 
200
    """Tests for BzrDir.create_repository."""
 
201
 
 
202
    def test_makes_repository(self):
 
203
        """When there is a bzrdir present, the call succeeds."""
 
204
        backing = self.get_transport()
 
205
        self.make_bzrdir('.')
 
206
        request_class = bzrlib.smart.bzrdir.SmartServerRequestCreateRepository
 
207
        request = request_class(backing)
 
208
        reference_bzrdir_format = bzrdir.format_registry.get('default')()
 
209
        reference_format = reference_bzrdir_format.repository_format
 
210
        network_name = reference_format.network_name()
 
211
        expected = SuccessfulSmartServerResponse(
 
212
            ('ok', 'no', 'no', 'no', network_name))
 
213
        self.assertEqual(expected, request.execute('', network_name, 'True'))
 
214
 
 
215
 
163
216
class TestSmartServerRequestFindRepository(tests.TestCaseWithMemoryTransport):
164
217
    """Tests for BzrDir.find_repository."""
165
218
 
172
225
            request.execute(''))
173
226
 
174
227
    def test_nonshared_repository(self):
175
 
        # nonshared repositorys only allow 'find' to return a handle when the 
176
 
        # path the repository is being searched on is the same as that that 
 
228
        # nonshared repositorys only allow 'find' to return a handle when the
 
229
        # path the repository is being searched on is the same as that that
177
230
        # the repository is at.
178
231
        backing = self.get_transport()
179
232
        request = self._request_class(backing)
197
250
            subtrees = 'yes'
198
251
        else:
199
252
            subtrees = 'no'
200
 
        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 ==
201
259
            self._request_class):
202
260
            # All tests so far are on formats, and for non-external
203
261
            # repositories.
250
308
        self.assertEqual(SmartServerResponse(('ok', )),
251
309
            request.execute(''))
252
310
        made_dir = bzrdir.BzrDir.open_from_transport(backing)
253
 
        # no branch, tree or repository is expected with the current 
 
311
        # no branch, tree or repository is expected with the current
254
312
        # default formart.
255
313
        self.assertRaises(errors.NoWorkingTree, made_dir.open_workingtree)
256
314
        self.assertRaises(errors.NotBranchError, made_dir.open_branch)
302
360
            request.execute('reference'))
303
361
 
304
362
 
 
363
class TestSmartServerRequestOpenBranchV2(TestCaseWithChrootedTransport):
 
364
 
 
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', )),
 
371
            request.execute(''))
 
372
 
 
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)),
 
379
            request.execute(''))
 
380
 
 
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'))
 
391
 
 
392
 
305
393
class TestSmartServerRequestRevisionHistory(tests.TestCaseWithMemoryTransport):
306
394
 
307
395
    def test_empty(self):
388
476
    def test_with_content(self):
389
477
        # SmartServerBranchGetConfigFile should return the content from
390
478
        # branch.control_files.get('branch.conf') for now - in the future it may
391
 
        # perform more complex processing. 
 
479
        # perform more complex processing.
392
480
        backing = self.get_transport()
393
481
        request = smart.branch.SmartServerBranchGetConfigFile(backing)
394
482
        branch = self.make_branch('.')
415
503
 
416
504
    def unlock_branch(self):
417
505
        self.tree.branch.unlock()
418
 
        
 
506
 
419
507
    def set_last_revision(self, revision_id, revno):
420
508
        branch_token, repo_token = self.lock_branch()
421
509
        response = self._set_last_revision(
427
515
        response = self.set_last_revision(revision_id, revno)
428
516
        self.assertEqual(SuccessfulSmartServerResponse(('ok',)), response)
429
517
 
430
 
        
 
518
 
431
519
class TestSetLastRevisionVerbMixin(object):
432
520
    """Mixin test case for verbs that implement set_last_revision."""
433
521
 
538
626
        self.assertEqual(
539
627
            SuccessfulSmartServerResponse(('ok', revno, revision_id)),
540
628
            response)
541
 
        
 
629
 
542
630
    def test_branch_last_revision_info_rewind(self):
543
631
        """A branch's tip can be set to a revision that is an ancestor of the
544
632
        current tip, but only if allow_overwrite_descendant is passed.
587
675
        # child-1.
588
676
        new_r2 = self.tree.commit('2nd commit', rev_id='child-2')
589
677
        self.tree.unlock()
590
 
        
 
678
 
591
679
    def test_not_allow_diverged(self):
592
680
        """If allow_diverged is not passed, then setting a divergent history
593
681
        returns a Diverged error.
615
703
        self.assertEqual('child-1', self.tree.branch.last_revision())
616
704
 
617
705
 
 
706
class TestSmartServerBranchRequestGetParent(tests.TestCaseWithMemoryTransport):
 
707
 
 
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')
 
712
        self.assertEquals(
 
713
            SuccessfulSmartServerResponse(('',)), response)
 
714
 
 
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')
 
720
        self.assertEquals(
 
721
            SuccessfulSmartServerResponse(("../foo",)),
 
722
            response)
 
723
 
 
724
 
 
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.
 
728
 
 
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')
 
734
        self.assertEquals(
 
735
            SuccessfulSmartServerResponse(('',)), response)
 
736
 
 
737
 
618
738
class TestSmartServerBranchRequestGetStackedOnURL(tests.TestCaseWithMemoryTransport):
619
739
 
620
740
    def test_get_stacked_on_url(self):
834
954
 
835
955
        self.assertEqual(SmartServerResponse(('ok', ), rev_id_utf8),
836
956
            request.execute('', rev_id_utf8))
837
 
    
 
957
 
838
958
    def test_no_such_revision(self):
839
959
        backing = self.get_transport()
840
960
        request = smart.repository.SmartServerRepositoryGetRevisionGraph(backing)
850
970
            request.execute('', 'missingrevision'))
851
971
 
852
972
 
 
973
class TestSmartServerRepositoryGetStream(tests.TestCaseWithMemoryTransport):
 
974
 
 
975
    def make_two_commit_repo(self):
 
976
        tree = self.make_branch_and_memory_tree('.')
 
977
        tree.lock_write()
 
978
        tree.add('')
 
979
        r1 = tree.commit('1st commit')
 
980
        r2 = tree.commit('2nd commit', rev_id=u'\xc8'.encode('utf-8'))
 
981
        tree.unlock()
 
982
        repo = tree.branch.repository
 
983
        return repo, r1, r2
 
984
 
 
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')
 
997
 
 
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')
 
1010
 
 
1011
 
853
1012
class TestSmartServerRequestHasRevision(tests.TestCaseWithMemoryTransport):
854
1013
 
855
1014
    def test_missing_revision(self):
1037
1196
            SmartServerResponse(('yes',)), response)
1038
1197
 
1039
1198
 
 
1199
class TestSmartServerRepositorySetMakeWorkingTrees(tests.TestCaseWithMemoryTransport):
 
1200
 
 
1201
    def test_set_false(self):
 
1202
        backing = self.get_transport()
 
1203
        repo = self.make_repository('.', shared=True)
 
1204
        repo.set_make_working_trees(True)
 
1205
        request_class = smart.repository.SmartServerRepositorySetMakeWorkingTrees
 
1206
        request = request_class(backing)
 
1207
        self.assertEqual(SuccessfulSmartServerResponse(('ok',)),
 
1208
            request.execute('', 'False'))
 
1209
        repo = repo.bzrdir.open_repository()
 
1210
        self.assertFalse(repo.make_working_trees())
 
1211
 
 
1212
    def test_set_true(self):
 
1213
        backing = self.get_transport()
 
1214
        repo = self.make_repository('.', shared=True)
 
1215
        repo.set_make_working_trees(False)
 
1216
        request_class = smart.repository.SmartServerRepositorySetMakeWorkingTrees
 
1217
        request = request_class(backing)
 
1218
        self.assertEqual(SuccessfulSmartServerResponse(('ok',)),
 
1219
            request.execute('', 'True'))
 
1220
        repo = repo.bzrdir.open_repository()
 
1221
        self.assertTrue(repo.make_working_trees())
 
1222
 
 
1223
 
1040
1224
class TestSmartServerPackRepositoryAutopack(tests.TestCaseWithTransport):
1041
1225
 
1042
1226
    def make_repo_needing_autopacking(self, path='.'):
1060
1244
        self.assertEqual(SmartServerResponse(('ok',)), response)
1061
1245
        repo._pack_collection.reload_pack_names()
1062
1246
        self.assertEqual(1, len(repo._pack_collection.names()))
1063
 
    
 
1247
 
1064
1248
    def test_autopack_not_needed(self):
1065
1249
        tree = self.make_branch_and_tree('.', format='pack-0.92')
1066
1250
        repo = tree.branch.repository
1073
1257
        self.assertEqual(SmartServerResponse(('ok',)), response)
1074
1258
        repo._pack_collection.reload_pack_names()
1075
1259
        self.assertEqual(9, len(repo._pack_collection.names()))
1076
 
    
 
1260
 
1077
1261
    def test_autopack_on_nonpack_format(self):
1078
1262
        """A request to autopack a non-pack repo is a no-op."""
1079
1263
        repo = self.make_repository('.', format='knit')
1082
1266
            backing)
1083
1267
        response = request.execute('')
1084
1268
        self.assertEqual(SmartServerResponse(('ok',)), response)
1085
 
        
 
1269
 
1086
1270
 
1087
1271
class TestHandlers(tests.TestCase):
1088
1272
    """Tests for the request.request_handlers object."""
1100
1284
            smart.request.request_handlers.get('Branch.get_config_file'),
1101
1285
            smart.branch.SmartServerBranchGetConfigFile)
1102
1286
        self.assertEqual(
 
1287
            smart.request.request_handlers.get('Branch.get_parent'),
 
1288
            smart.branch.SmartServerBranchGetParent)
 
1289
        self.assertEqual(
 
1290
            smart.request.request_handlers.get('Branch.get_tags_bytes'),
 
1291
            smart.branch.SmartServerBranchGetTagsBytes)
 
1292
        self.assertEqual(
1103
1293
            smart.request.request_handlers.get('Branch.lock_write'),
1104
1294
            smart.branch.SmartServerBranchRequestLockWrite)
1105
1295
        self.assertEqual(
1127
1317
            smart.request.request_handlers.get('BzrDirFormat.initialize'),
1128
1318
            smart.bzrdir.SmartServerRequestInitializeBzrDir)
1129
1319
        self.assertEqual(
 
1320
            smart.request.request_handlers.get('BzrDir.cloning_metadir'),
 
1321
            smart.bzrdir.SmartServerBzrDirRequestCloningMetaDir)
 
1322
        self.assertEqual(
1130
1323
            smart.request.request_handlers.get('BzrDir.open_branch'),
1131
1324
            smart.bzrdir.SmartServerRequestOpenBranch)
1132
1325
        self.assertEqual(
 
1326
            smart.request.request_handlers.get('BzrDir.open_branchV2'),
 
1327
            smart.bzrdir.SmartServerRequestOpenBranchV2)
 
1328
        self.assertEqual(
1133
1329
            smart.request.request_handlers.get('PackRepository.autopack'),
1134
1330
            smart.packrepository.SmartServerPackRepositoryAutopack)
1135
1331
        self.assertEqual(
1152
1348
            smart.request.request_handlers.get('Repository.lock_write'),
1153
1349
            smart.repository.SmartServerRepositoryLockWrite)
1154
1350
        self.assertEqual(
 
1351
            smart.request.request_handlers.get('Repository.get_stream'),
 
1352
            smart.repository.SmartServerRepositoryGetStream)
 
1353
        self.assertEqual(
1155
1354
            smart.request.request_handlers.get('Repository.tarball'),
1156
1355
            smart.repository.SmartServerRepositoryTarball)
1157
1356
        self.assertEqual(