← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_smart_request.py

  • Committer: Patch Queue Manager
  • Date: 2015-10-05 13:45:00 UTC
  • mfrom: (6603.3.1 bts794146)
  • Revision ID: pqm@pqm.ubuntu.com-20151005134500-v244rho557tv0ukd
(vila) Resolve Bug #1480015: Test failure: hexify removed from paramiko
 (Andrew Starr-Bochicchio) (Andrew Starr-Bochicchio)

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_smart_request.py
1
 
# Copyright (C) 2009 Canonical Ltd
 
1
# Copyright (C) 2009, 2010 Canonical Ltd
2
2
#
3
3
# This program is free software; you can redistribute it and/or modify
4
4
# it under the terms of the GNU General Public License as published by
18
18
 
19
19
import threading
20
20
 
21
 
from bzrlib import errors
 
21
from bzrlib import (
 
22
    errors,
 
23
    transport,
 
24
    )
22
25
from bzrlib.bzrdir import BzrDir
23
26
from bzrlib.smart import request
24
27
from bzrlib.tests import TestCase, TestCaseWithMemoryTransport
25
 
from bzrlib.transport import get_transport
26
28
 
27
29
 
28
30
class NoBodyRequest(request.SmartServerRequest):
34
36
 
35
37
class DoErrorRequest(request.SmartServerRequest):
36
38
    """A request that raises an error from self.do()."""
37
 
    
 
39
 
38
40
    def do(self):
39
41
        raise errors.NoSuchFile('xyzzy')
40
42
 
41
43
 
 
44
class DoUnexpectedErrorRequest(request.SmartServerRequest):
 
45
    """A request that encounters a generic error in self.do()"""
 
46
 
 
47
    def do(self):
 
48
        dict()[1]
 
49
 
 
50
 
42
51
class ChunkErrorRequest(request.SmartServerRequest):
43
52
    """A request that raises an error from self.do_chunk()."""
44
53
    
109
118
        self.assertEqual(
110
119
            [[transport]] * 3, handler._command.jail_transports_log)
111
120
 
 
121
    def test_all_registered_requests_are_safety_qualified(self):
 
122
        unclassified_requests = []
 
123
        allowed_info = ('read', 'idem', 'mutate', 'semivfs', 'semi', 'stream')
 
124
        for key in request.request_handlers.keys():
 
125
            info = request.request_handlers.get_info(key)
 
126
            if info is None or info not in allowed_info:
 
127
                unclassified_requests.append(key)
 
128
        if unclassified_requests:
 
129
            self.fail('These requests were not categorized as safe/unsafe'
 
130
                      ' to retry: %s'  % (unclassified_requests,))
112
131
 
113
132
 
114
133
class TestSmartRequestHandlerErrorTranslation(TestCase):
147
166
        handler.end_received()
148
167
        self.assertResponseIsTranslatedError(handler)
149
168
 
 
169
    def test_unexpected_error_translation(self):
 
170
        handler = request.SmartServerRequestHandler(
 
171
            None, {'foo': DoUnexpectedErrorRequest}, '/')
 
172
        handler.args_received(('foo',))
 
173
        self.assertEqual(
 
174
            request.FailedSmartServerResponse(('error', 'KeyError', "1")),
 
175
            handler.response)
 
176
 
150
177
 
151
178
class TestRequestHanderErrorTranslation(TestCase):
152
179
    """Tests for bzrlib.smart.request._translate_error."""
170
197
            ('TokenMismatch', 'some-token', 'actual-token'),
171
198
            errors.TokenMismatch('some-token', 'actual-token'))
172
199
 
 
200
    def test_MemoryError(self):
 
201
        self.assertTranslationEqual(("MemoryError",), MemoryError())
 
202
 
 
203
    def test_generic_Exception(self):
 
204
        self.assertTranslationEqual(('error', 'Exception', ""),
 
205
            Exception())
 
206
 
 
207
    def test_generic_BzrError(self):
 
208
        self.assertTranslationEqual(('error', 'BzrError', "some text"),
 
209
            errors.BzrError(msg="some text"))
 
210
 
 
211
    def test_generic_zlib_error(self):
 
212
        from zlib import error
 
213
        msg = "Error -3 while decompressing data: incorrect data check"
 
214
        self.assertTranslationEqual(('error', 'zlib.error', msg),
 
215
            error(msg))
 
216
 
173
217
 
174
218
class TestRequestJail(TestCaseWithMemoryTransport):
175
 
    
 
219
 
176
220
    def test_jail(self):
177
221
        transport = self.get_transport('blah')
178
222
        req = request.SmartServerRequest(transport)
185
229
 
186
230
class TestJailHook(TestCaseWithMemoryTransport):
187
231
 
188
 
    def tearDown(self):
189
 
        request.jail_info.transports = None
190
 
        TestCaseWithMemoryTransport.tearDown(self)
 
232
    def setUp(self):
 
233
        super(TestJailHook, self).setUp()
 
234
        def clear_jail_info():
 
235
            request.jail_info.transports = None
 
236
        self.addCleanup(clear_jail_info)
191
237
 
192
238
    def test_jail_hook(self):
193
239
        request.jail_info.transports = None
203
249
        # A parent is not allowed
204
250
        self.assertRaises(errors.JailBreak, _pre_open_hook, t.clone('..'))
205
251
        # A completely unrelated transport is not allowed
206
 
        self.assertRaises(
207
 
            errors.JailBreak, _pre_open_hook, get_transport('http://host/'))
 
252
        self.assertRaises(errors.JailBreak, _pre_open_hook,
 
253
                          transport.get_transport_from_url('http://host/'))
208
254
 
209
255
    def test_open_bzrdir_in_non_main_thread(self):
210
256
        """Opening a bzrdir in a non-main thread should work ok.