/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_remote.py

  • Committer: Alexander Belchenko
  • Date: 2011-03-28 08:27:18 UTC
  • mfrom: (5742 +trunk)
  • mto: (5759.1.1 integration)
  • mto: This revision was merged to the branch mainline in revision 5760.
  • Revision ID: bialix@ukr.net-20110328082718-wb3emf38d4iffmcp
merge trunk

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_remote.py
52
52
    RemoteBranch,
53
53
    RemoteBranchFormat,
54
54
    RemoteBzrDir,
 
55
    RemoteBzrDirFormat,
55
56
    RemoteRepository,
56
57
    RemoteRepositoryFormat,
57
58
    )
95
96
        self.addCleanup(self.transport.disconnect)
96
97
 
97
98
    def test_create_remote_bzrdir(self):
98
 
        b = remote.RemoteBzrDir(self.transport, remote.RemoteBzrDirFormat())
 
99
        b = remote.RemoteBzrDir(self.transport, RemoteBzrDirFormat())
99
100
        self.assertIsInstance(b, BzrDir)
100
101
 
101
102
    def test_open_remote_branch(self):
102
103
        # open a standalone branch in the working directory
103
 
        b = remote.RemoteBzrDir(self.transport, remote.RemoteBzrDirFormat())
 
104
        b = remote.RemoteBzrDir(self.transport, RemoteBzrDirFormat())
104
105
        branch = b.open_branch()
105
106
        self.assertIsInstance(branch, Branch)
106
107
 
124
125
        fmt = BzrDirFormat.find_format(self.transport)
125
126
        self.assertTrue(bzrdir.RemoteBzrProber
126
127
                        in controldir.ControlDirFormat._server_probers)
127
 
        self.assertIsInstance(fmt, remote.RemoteBzrDirFormat)
 
128
        self.assertIsInstance(fmt, RemoteBzrDirFormat)
128
129
 
129
130
    def test_open_detected_smart_format(self):
130
131
        fmt = BzrDirFormat.find_format(self.transport)
450
451
        client.add_expected_call(
451
452
            'BzrDir.open_branchV3', ('quack/',),
452
453
            'success', ('ref', self.get_url('referenced'))),
453
 
        a_bzrdir = RemoteBzrDir(transport, remote.RemoteBzrDirFormat(),
 
454
        a_bzrdir = RemoteBzrDir(transport, RemoteBzrDirFormat(),
454
455
            _client=client)
455
456
        result = a_bzrdir.cloning_metadir()
456
457
        # We should have got a control dir matching the referenced branch.
469
470
        client.add_expected_call(
470
471
            'BzrDir.cloning_metadir', ('quack/', 'False'),
471
472
            'success', (control_name, '', ('branch', ''))),
472
 
        a_bzrdir = RemoteBzrDir(transport, remote.RemoteBzrDirFormat(),
 
473
        a_bzrdir = RemoteBzrDir(transport, RemoteBzrDirFormat(),
473
474
            _client=client)
474
475
        result = a_bzrdir.cloning_metadir()
475
476
        # We should have got a reference control dir with default branch and
495
496
        client.add_expected_call(
496
497
            'BzrDir.open_2.1', ('quack/',), 'success', ('no',))
497
498
        self.assertRaises(errors.NotBranchError, RemoteBzrDir, transport,
498
 
                remote.RemoteBzrDirFormat(), _client=client, _force_probe=True)
 
499
                RemoteBzrDirFormat(), _client=client, _force_probe=True)
499
500
        self.assertFinished(client)
500
501
 
501
502
    def test_present_without_workingtree(self):
502
503
        client, transport = self.make_fake_client_and_transport()
503
504
        client.add_expected_call(
504
505
            'BzrDir.open_2.1', ('quack/',), 'success', ('yes', 'no'))
505
 
        bd = RemoteBzrDir(transport, remote.RemoteBzrDirFormat(),
 
506
        bd = RemoteBzrDir(transport, RemoteBzrDirFormat(),
506
507
            _client=client, _force_probe=True)
507
508
        self.assertIsInstance(bd, RemoteBzrDir)
508
509
        self.assertFalse(bd.has_workingtree())
513
514
        client, transport = self.make_fake_client_and_transport()
514
515
        client.add_expected_call(
515
516
            'BzrDir.open_2.1', ('quack/',), 'success', ('yes', 'yes'))
516
 
        bd = RemoteBzrDir(transport, remote.RemoteBzrDirFormat(),
 
517
        bd = RemoteBzrDir(transport, RemoteBzrDirFormat(),
517
518
            _client=client, _force_probe=True)
518
519
        self.assertIsInstance(bd, RemoteBzrDir)
519
520
        self.assertTrue(bd.has_workingtree())
526
527
            'BzrDir.open_2.1', ('quack/',), 'unknown', ('BzrDir.open_2.1',))
527
528
        client.add_expected_call(
528
529
            'BzrDir.open', ('quack/',), 'success', ('yes',))
529
 
        bd = RemoteBzrDir(transport, remote.RemoteBzrDirFormat(),
 
530
        bd = RemoteBzrDir(transport, RemoteBzrDirFormat(),
530
531
            _client=client, _force_probe=True)
531
532
        self.assertIsInstance(bd, RemoteBzrDir)
532
533
        self.assertFinished(client)
548
549
            'BzrDir.open_2.1', ('quack/',), 'unknown', ('BzrDir.open_2.1',))
549
550
        client.add_expected_call(
550
551
            'BzrDir.open', ('quack/',), 'success', ('yes',))
551
 
        bd = RemoteBzrDir(transport, remote.RemoteBzrDirFormat(),
 
552
        bd = RemoteBzrDir(transport, RemoteBzrDirFormat(),
552
553
            _client=client, _force_probe=True)
553
554
        self.assertIsInstance(bd, RemoteBzrDir)
554
555
        self.assertFinished(client)
585
586
        client.add_expected_call(
586
587
            'Branch.get_stacked_on_url', ('quack/',),
587
588
            'error', ('NotStacked',))
588
 
        bzrdir = RemoteBzrDir(transport, remote.RemoteBzrDirFormat(),
 
589
        bzrdir = RemoteBzrDir(transport, RemoteBzrDirFormat(),
589
590
            _client=client)
590
591
        result = bzrdir.open_branch()
591
592
        self.assertIsInstance(result, RemoteBranch)
598
599
        transport = transport.clone('quack')
599
600
        client = FakeClient(transport.base)
600
601
        client.add_error_response('nobranch')
601
 
        bzrdir = RemoteBzrDir(transport, remote.RemoteBzrDirFormat(),
 
602
        bzrdir = RemoteBzrDir(transport, RemoteBzrDirFormat(),
602
603
            _client=client)
603
604
        self.assertRaises(errors.NotBranchError, bzrdir.open_branch)
604
605
        self.assertEqual(
615
616
        transport = MemoryTransport()
616
617
        # no requests on the network - catches other api calls being made.
617
618
        client = FakeClient(transport.base)
618
 
        bzrdir = RemoteBzrDir(transport, remote.RemoteBzrDirFormat(),
 
619
        bzrdir = RemoteBzrDir(transport, RemoteBzrDirFormat(),
619
620
            _client=client)
620
621
        # patch the open_branch call to record that it was called.
621
622
        bzrdir.open_branch = open_branch
640
641
        client.add_expected_call(
641
642
            'Branch.get_stacked_on_url', ('~hello/',),
642
643
            'error', ('NotStacked',))
643
 
        bzrdir = RemoteBzrDir(transport, remote.RemoteBzrDirFormat(),
 
644
        bzrdir = RemoteBzrDir(transport, RemoteBzrDirFormat(),
644
645
            _client=client)
645
646
        result = bzrdir.open_branch()
646
647
        self.assertFinished(client)
663
664
        client.add_success_response(
664
665
            'ok', '', rich_response, subtree_response, external_lookup,
665
666
            network_name)
666
 
        bzrdir = RemoteBzrDir(transport, remote.RemoteBzrDirFormat(),
 
667
        bzrdir = RemoteBzrDir(transport, RemoteBzrDirFormat(),
667
668
            _client=client)
668
669
        result = bzrdir.open_repository()
669
670
        self.assertEqual(
715
716
            'BzrDir.create_branch', ('quack/', network_name),
716
717
            'success', ('ok', network_name, '', 'no', 'no', 'yes',
717
718
            reference_repo_name))
718
 
        a_bzrdir = RemoteBzrDir(transport, remote.RemoteBzrDirFormat(),
 
719
        a_bzrdir = RemoteBzrDir(transport, RemoteBzrDirFormat(),
719
720
            _client=client)
720
721
        branch = a_bzrdir.create_branch()
721
722
        # We should have got a remote branch
724
725
        format = branch._format
725
726
        self.assertEqual(network_name, format.network_name())
726
727
 
 
728
    def test_already_open_repo_and_reused_medium(self):
 
729
        """Bug 726584: create_branch(..., repository=repo) should work
 
730
        regardless of what the smart medium's base URL is.
 
731
        """
 
732
        self.transport_server = test_server.SmartTCPServer_for_testing
 
733
        transport = self.get_transport('.')
 
734
        repo = self.make_repository('quack')
 
735
        # Client's medium rooted a transport root (not at the bzrdir)
 
736
        client = FakeClient(transport.base)
 
737
        transport = transport.clone('quack')
 
738
        reference_bzrdir_format = bzrdir.format_registry.get('default')()
 
739
        reference_format = reference_bzrdir_format.get_branch_format()
 
740
        network_name = reference_format.network_name()
 
741
        reference_repo_fmt = reference_bzrdir_format.repository_format
 
742
        reference_repo_name = reference_repo_fmt.network_name()
 
743
        client.add_expected_call(
 
744
            'BzrDir.create_branch', ('extra/quack/', network_name),
 
745
            'success', ('ok', network_name, '', 'no', 'no', 'yes',
 
746
            reference_repo_name))
 
747
        a_bzrdir = RemoteBzrDir(transport, RemoteBzrDirFormat(),
 
748
            _client=client)
 
749
        branch = a_bzrdir.create_branch(repository=repo)
 
750
        # We should have got a remote branch
 
751
        self.assertIsInstance(branch, remote.RemoteBranch)
 
752
        # its format should have the settings from the response
 
753
        format = branch._format
 
754
        self.assertEqual(network_name, format.network_name())
 
755
 
727
756
 
728
757
class TestBzrDirCreateRepository(TestRemote):
729
758
 
750
779
                'Bazaar repository format 2a (needs bzr 1.16 or later)\n',
751
780
                'False'),
752
781
            'success', ('ok', 'yes', 'yes', 'yes', network_name))
753
 
        a_bzrdir = RemoteBzrDir(transport, remote.RemoteBzrDirFormat(),
 
782
        a_bzrdir = RemoteBzrDir(transport, RemoteBzrDirFormat(),
754
783
            _client=client)
755
784
        repo = a_bzrdir.create_repository()
756
785
        # We should have got a remote repository
785
814
        client.add_success_response('stat', '0', '65535')
786
815
        remote_transport = RemoteTransport(server_url + 'quack/', medium=False,
787
816
            _client=client)
788
 
        bzrdir = RemoteBzrDir(remote_transport, remote.RemoteBzrDirFormat(),
 
817
        bzrdir = RemoteBzrDir(remote_transport, RemoteBzrDirFormat(),
789
818
            _client=client)
790
819
        repo = bzrdir.open_repository()
791
820
        self.assertEqual(
818
847
        client.add_success_response('stat', '0', '65535')
819
848
        remote_transport = RemoteTransport(server_url + 'quack/', medium=False,
820
849
            _client=client)
821
 
        bzrdir = RemoteBzrDir(remote_transport, remote.RemoteBzrDirFormat(),
 
850
        bzrdir = RemoteBzrDir(remote_transport, RemoteBzrDirFormat(),
822
851
            _client=client)
823
852
        repo = bzrdir.open_repository()
824
853
        self.assertEqual(
839
868
        transport = transport.clone('quack')
840
869
        client = FakeClient(transport.base)
841
870
        client.add_success_response('ok', '', 'no', 'no', 'no', network_name)
842
 
        bzrdir = RemoteBzrDir(transport, remote.RemoteBzrDirFormat(),
 
871
        bzrdir = RemoteBzrDir(transport, RemoteBzrDirFormat(),
843
872
            _client=client)
844
873
        repo = bzrdir.open_repository()
845
874
        self.assertEqual(
852
881
 
853
882
    def test_success(self):
854
883
        """Simple test for typical successful call."""
855
 
        fmt = bzrdir.RemoteBzrDirFormat()
 
884
        fmt = RemoteBzrDirFormat()
856
885
        default_format_name = BzrDirFormat.get_default_format().network_name()
857
886
        transport = self.get_transport()
858
887
        client = FakeClient(transport.base)
874
903
        """Error responses are translated, e.g. 'PermissionDenied' raises the
875
904
        corresponding error from the client.
876
905
        """
877
 
        fmt = bzrdir.RemoteBzrDirFormat()
 
906
        fmt = RemoteBzrDirFormat()
878
907
        default_format_name = BzrDirFormat.get_default_format().network_name()
879
908
        transport = self.get_transport()
880
909
        client = FakeClient(transport.base)
898
927
        """Integration test for error translation."""
899
928
        transport = self.make_smart_server('foo')
900
929
        transport = transport.clone('no-such-path')
901
 
        fmt = bzrdir.RemoteBzrDirFormat()
 
930
        fmt = RemoteBzrDirFormat()
902
931
        err = self.assertRaises(errors.NoSuchFile,
903
932
            fmt.initialize_on_transport_ex, transport, create_prefix=False)
904
933
 
935
964
 
936
965
    def make_remote_bzrdir(self, transport, client):
937
966
        """Make a RemotebzrDir using 'client' as the _client."""
938
 
        return RemoteBzrDir(transport, remote.RemoteBzrDirFormat(),
 
967
        return RemoteBzrDir(transport, RemoteBzrDirFormat(),
939
968
            _client=client)
940
969
 
941
970
 
1143
1172
            [('set_tags_bytes', 'tags bytes')] * 2, real_branch.calls)
1144
1173
 
1145
1174
 
 
1175
class TestBranchHeadsToFetch(RemoteBranchTestCase):
 
1176
 
 
1177
    def test_uses_last_revision_info_and_tags_by_default(self):
 
1178
        transport = MemoryTransport()
 
1179
        client = FakeClient(transport.base)
 
1180
        client.add_expected_call(
 
1181
            'Branch.get_stacked_on_url', ('quack/',),
 
1182
            'error', ('NotStacked',))
 
1183
        client.add_expected_call(
 
1184
            'Branch.last_revision_info', ('quack/',),
 
1185
            'success', ('ok', '1', 'rev-tip'))
 
1186
        # XXX: this will break if the default format's serialization of tags
 
1187
        # changes, or if the RPC for fetching tags changes from get_tags_bytes.
 
1188
        client.add_expected_call(
 
1189
            'Branch.get_tags_bytes', ('quack/',),
 
1190
            'success', ('d5:tag-17:rev-foo5:tag-27:rev-bare',))
 
1191
        transport.mkdir('quack')
 
1192
        transport = transport.clone('quack')
 
1193
        branch = self.make_remote_branch(transport, client)
 
1194
        result = branch.heads_to_fetch()
 
1195
        self.assertFinished(client)
 
1196
        self.assertEqual(
 
1197
            (set(['rev-tip']), set(['rev-foo', 'rev-bar'])), result)
 
1198
 
 
1199
    def test_uses_rpc_for_formats_with_non_default_heads_to_fetch(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.heads_to_fetch', ('quack/',),
 
1207
            'success', (['tip'], ['tagged-1', 'tagged-2']))
 
1208
        transport.mkdir('quack')
 
1209
        transport = transport.clone('quack')
 
1210
        branch = self.make_remote_branch(transport, client)
 
1211
        branch._format._use_default_local_heads_to_fetch = lambda: False
 
1212
        result = branch.heads_to_fetch()
 
1213
        self.assertFinished(client)
 
1214
        self.assertEqual((set(['tip']), set(['tagged-1', 'tagged-2'])), result)
 
1215
 
 
1216
    def test_backwards_compatible(self):
 
1217
        self.setup_smart_server_with_call_log()
 
1218
        # Make a branch with a single revision.
 
1219
        builder = self.make_branch_builder('foo')
 
1220
        builder.start_series()
 
1221
        builder.build_snapshot('tip', None, [
 
1222
            ('add', ('', 'root-id', 'directory', ''))])
 
1223
        builder.finish_series()
 
1224
        branch = builder.get_branch()
 
1225
        # Add two tags to that branch
 
1226
        branch.tags.set_tag('tag-1', 'rev-1')
 
1227
        branch.tags.set_tag('tag-2', 'rev-2')
 
1228
        self.addCleanup(branch.lock_read().unlock)
 
1229
        # Disable the heads_to_fetch verb
 
1230
        verb = 'Branch.heads_to_fetch'
 
1231
        self.disable_verb(verb)
 
1232
        self.reset_smart_call_log()
 
1233
        result = branch.heads_to_fetch()
 
1234
        self.assertEqual((set(['tip']), set(['rev-1', 'rev-2'])), result)
 
1235
        self.assertEqual(
 
1236
            ['Branch.last_revision_info', 'Branch.get_tags_bytes'],
 
1237
            [call.call.method for call in self.hpss_calls])
 
1238
 
 
1239
 
1146
1240
class TestBranchLastRevisionInfo(RemoteBranchTestCase):
1147
1241
 
1148
1242
    def test_empty_branch(self):
1203
1297
        client.add_expected_call(
1204
1298
            'Branch.get_stacked_on_url', ('stacked/',),
1205
1299
            'success', ('ok', vfs_url))
1206
 
        bzrdir = RemoteBzrDir(transport, remote.RemoteBzrDirFormat(),
 
1300
        bzrdir = RemoteBzrDir(transport, RemoteBzrDirFormat(),
1207
1301
            _client=client)
1208
1302
        repo_fmt = remote.RemoteRepositoryFormat()
1209
1303
        repo_fmt._custom_format = stacked_branch.repository._format
1236
1330
        # this will also do vfs access, but that goes direct to the transport
1237
1331
        # and isn't seen by the FakeClient.
1238
1332
        bzrdir = RemoteBzrDir(self.get_transport('stacked'),
1239
 
            remote.RemoteBzrDirFormat(), _client=client)
 
1333
            RemoteBzrDirFormat(), _client=client)
1240
1334
        branch = bzrdir.open_branch()
1241
1335
        result = branch.get_stacked_on_url()
1242
1336
        self.assertEqual('../base', result)
1269
1363
            'Branch.get_stacked_on_url', ('stacked/',),
1270
1364
            'success', ('ok', '../base'))
1271
1365
        bzrdir = RemoteBzrDir(self.get_transport('stacked'),
1272
 
            remote.RemoteBzrDirFormat(), _client=client)
 
1366
            RemoteBzrDirFormat(), _client=client)
1273
1367
        branch = bzrdir.open_branch()
1274
1368
        result = branch.get_stacked_on_url()
1275
1369
        self.assertEqual('../base', result)
1845
1939
        client = FakeClient(transport.base)
1846
1940
        transport = transport.clone(transport_path)
1847
1941
        # we do not want bzrdir to make any remote calls
1848
 
        bzrdir = RemoteBzrDir(transport, remote.RemoteBzrDirFormat(),
 
1942
        bzrdir = RemoteBzrDir(transport, RemoteBzrDirFormat(),
1849
1943
            _client=False)
1850
1944
        repo = RemoteRepository(bzrdir, None, _client=client)
1851
1945
        return repo, client
1859
1953
 
1860
1954
    def test_get_format_description(self):
1861
1955
        remote_format = RemoteBranchFormat()
1862
 
        real_format = branch.BranchFormat.get_default_format()
 
1956
        real_format = branch.format_registry.get_default()
1863
1957
        remote_format._network_name = real_format.network_name()
1864
1958
        self.assertEqual(remoted_description(real_format),
1865
1959
            remote_format.get_format_description())
2775
2869
             ('pack collection autopack',)],
2776
2870
            client._calls)
2777
2871
 
 
2872
    def test_oom_error_reporting(self):
 
2873
        """An out-of-memory condition on the server is reported clearly"""
 
2874
        transport_path = 'quack'
 
2875
        repo, client = self.setup_fake_client_and_repository(transport_path)
 
2876
        client.add_expected_call(
 
2877
            'PackRepository.autopack', ('quack/',),
 
2878
            'error', ('MemoryError',))
 
2879
        err = self.assertRaises(errors.BzrError, repo.autopack)
 
2880
        self.assertContainsRe(str(err), "^remote server out of mem")
 
2881
 
2778
2882
 
2779
2883
class TestErrorTranslationBase(tests.TestCaseWithMemoryTransport):
2780
2884
    """Base class for unit tests for bzrlib.remote._translate_error."""
2853
2957
            detail='extra detail')
2854
2958
        self.assertEqual(expected_error, translated_error)
2855
2959
 
 
2960
    def test_norepository(self):
 
2961
        bzrdir = self.make_bzrdir('')
 
2962
        translated_error = self.translateTuple(('norepository',),
 
2963
            bzrdir=bzrdir)
 
2964
        expected_error = errors.NoRepositoryPresent(bzrdir)
 
2965
        self.assertEqual(expected_error, translated_error)
 
2966
 
2856
2967
    def test_LockContention(self):
2857
2968
        translated_error = self.translateTuple(('LockContention',))
2858
2969
        expected_error = errors.LockContention('(remote lock)')
2886
2997
        expected_error = errors.DivergedBranches(branch, other_branch)
2887
2998
        self.assertEqual(expected_error, translated_error)
2888
2999
 
 
3000
    def test_NotStacked(self):
 
3001
        branch = self.make_branch('')
 
3002
        translated_error = self.translateTuple(('NotStacked',), branch=branch)
 
3003
        expected_error = errors.NotStacked(branch)
 
3004
        self.assertEqual(expected_error, translated_error)
 
3005
 
2889
3006
    def test_ReadError_no_args(self):
2890
3007
        path = 'a path'
2891
3008
        translated_error = self.translateTuple(('ReadError',), path=path)
2907
3024
 
2908
3025
    def test_PermissionDenied_no_args(self):
2909
3026
        path = 'a path'
2910
 
        translated_error = self.translateTuple(('PermissionDenied',), path=path)
 
3027
        translated_error = self.translateTuple(('PermissionDenied',),
 
3028
            path=path)
2911
3029
        expected_error = errors.PermissionDenied(path)
2912
3030
        self.assertEqual(expected_error, translated_error)
2913
3031
 
2936
3054
        expected_error = errors.PermissionDenied(path, extra)
2937
3055
        self.assertEqual(expected_error, translated_error)
2938
3056
 
 
3057
    # GZ 2011-03-02: TODO test for PermissionDenied with non-ascii 'extra'
 
3058
 
 
3059
    def test_NoSuchFile_context_path(self):
 
3060
        local_path = "local path"
 
3061
        translated_error = self.translateTuple(('ReadError', "remote path"),
 
3062
            path=local_path)
 
3063
        expected_error = errors.ReadError(local_path)
 
3064
        self.assertEqual(expected_error, translated_error)
 
3065
 
 
3066
    def test_NoSuchFile_without_context(self):
 
3067
        remote_path = "remote path"
 
3068
        translated_error = self.translateTuple(('ReadError', remote_path))
 
3069
        expected_error = errors.ReadError(remote_path)
 
3070
        self.assertEqual(expected_error, translated_error)
 
3071
 
 
3072
    def test_ReadOnlyError(self):
 
3073
        translated_error = self.translateTuple(('ReadOnlyError',))
 
3074
        expected_error = errors.TransportNotPossible("readonly transport")
 
3075
        self.assertEqual(expected_error, translated_error)
 
3076
 
 
3077
    def test_MemoryError(self):
 
3078
        translated_error = self.translateTuple(('MemoryError',))
 
3079
        self.assertStartsWith(str(translated_error),
 
3080
            "remote server out of memory")
 
3081
 
 
3082
    def test_generic_IndexError_no_classname(self):
 
3083
        err = errors.ErrorFromSmartServer(('error', "list index out of range"))
 
3084
        translated_error = self.translateErrorFromSmartServer(err)
 
3085
        expected_error = errors.UnknownErrorFromSmartServer(err)
 
3086
        self.assertEqual(expected_error, translated_error)
 
3087
 
 
3088
    # GZ 2011-03-02: TODO test generic non-ascii error string
 
3089
 
 
3090
    def test_generic_KeyError(self):
 
3091
        err = errors.ErrorFromSmartServer(('error', 'KeyError', "1"))
 
3092
        translated_error = self.translateErrorFromSmartServer(err)
 
3093
        expected_error = errors.UnknownErrorFromSmartServer(err)
 
3094
        self.assertEqual(expected_error, translated_error)
 
3095
 
2939
3096
 
2940
3097
class TestErrorTranslationRobustness(TestErrorTranslationBase):
2941
3098
    """Unit tests for bzrlib.remote._translate_error's robustness.