1
# Copyright (C) 2009, 2010 Canonical Ltd
3
# This program is free software; you can redistribute it and/or modify
4
# it under the terms of the GNU General Public License as published by
5
# the Free Software Foundation; either version 2 of the License, or
6
# (at your option) any later version.
8
# This program is distributed in the hope that it will be useful,
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
11
# GNU General Public License for more details.
13
# You should have received a copy of the GNU General Public License
14
# along with this program; if not, write to the Free Software
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
17
"""Tests for smart server request infrastructure (bzrlib.smart.request)."""
25
from bzrlib.bzrdir import BzrDir
26
from bzrlib.smart import request
27
from bzrlib.tests import TestCase, TestCaseWithMemoryTransport
30
class NoBodyRequest(request.SmartServerRequest):
31
"""A request that does not implement do_body."""
34
return request.SuccessfulSmartServerResponse(('ok',))
37
class DoErrorRequest(request.SmartServerRequest):
38
"""A request that raises an error from self.do()."""
41
raise errors.NoSuchFile('xyzzy')
44
class DoUnexpectedErrorRequest(request.SmartServerRequest):
45
"""A request that encounters a generic error in self.do()"""
51
class ChunkErrorRequest(request.SmartServerRequest):
52
"""A request that raises an error from self.do_chunk()."""
58
def do_chunk(self, bytes):
59
raise errors.NoSuchFile('xyzzy')
62
class EndErrorRequest(request.SmartServerRequest):
63
"""A request that raises an error from self.do_end()."""
69
def do_chunk(self, bytes):
74
raise errors.NoSuchFile('xyzzy')
77
class CheckJailRequest(request.SmartServerRequest):
79
def __init__(self, *args):
80
request.SmartServerRequest.__init__(self, *args)
81
self.jail_transports_log = []
84
self.jail_transports_log.append(request.jail_info.transports)
86
def do_chunk(self, bytes):
87
self.jail_transports_log.append(request.jail_info.transports)
90
self.jail_transports_log.append(request.jail_info.transports)
93
class TestSmartRequest(TestCase):
95
def test_request_class_without_do_body(self):
96
"""If a request has no body data, and the request's implementation does
97
not override do_body, then no exception is raised.
99
# Create a SmartServerRequestHandler with a SmartServerRequest subclass
100
# that does not implement do_body.
101
handler = request.SmartServerRequestHandler(
102
None, {'foo': NoBodyRequest}, '/')
103
# Emulate a request with no body (i.e. just args).
104
handler.args_received(('foo',))
105
handler.end_received()
106
# Request done, no exception was raised.
108
def test_only_request_code_is_jailed(self):
109
transport = 'dummy transport'
110
handler = request.SmartServerRequestHandler(
111
transport, {'foo': CheckJailRequest}, '/')
112
handler.args_received(('foo',))
113
self.assertEqual(None, request.jail_info.transports)
114
handler.accept_body('bytes')
115
self.assertEqual(None, request.jail_info.transports)
116
handler.end_received()
117
self.assertEqual(None, request.jail_info.transports)
119
[[transport]] * 3, handler._command.jail_transports_log)
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,))
133
class TestSmartRequestHandlerErrorTranslation(TestCase):
134
"""Tests that SmartServerRequestHandler will translate exceptions raised by
135
a SmartServerRequest into FailedSmartServerResponses.
138
def assertNoResponse(self, handler):
139
self.assertEqual(None, handler.response)
141
def assertResponseIsTranslatedError(self, handler):
142
expected_translation = ('NoSuchFile', 'xyzzy')
144
request.FailedSmartServerResponse(expected_translation),
147
def test_error_translation_from_args_received(self):
148
handler = request.SmartServerRequestHandler(
149
None, {'foo': DoErrorRequest}, '/')
150
handler.args_received(('foo',))
151
self.assertResponseIsTranslatedError(handler)
153
def test_error_translation_from_chunk_received(self):
154
handler = request.SmartServerRequestHandler(
155
None, {'foo': ChunkErrorRequest}, '/')
156
handler.args_received(('foo',))
157
self.assertNoResponse(handler)
158
handler.accept_body('bytes')
159
self.assertResponseIsTranslatedError(handler)
161
def test_error_translation_from_end_received(self):
162
handler = request.SmartServerRequestHandler(
163
None, {'foo': EndErrorRequest}, '/')
164
handler.args_received(('foo',))
165
self.assertNoResponse(handler)
166
handler.end_received()
167
self.assertResponseIsTranslatedError(handler)
169
def test_unexpected_error_translation(self):
170
handler = request.SmartServerRequestHandler(
171
None, {'foo': DoUnexpectedErrorRequest}, '/')
172
handler.args_received(('foo',))
174
request.FailedSmartServerResponse(('error', 'KeyError', "1")),
178
class TestRequestHanderErrorTranslation(TestCase):
179
"""Tests for bzrlib.smart.request._translate_error."""
181
def assertTranslationEqual(self, expected_tuple, error):
182
self.assertEqual(expected_tuple, request._translate_error(error))
184
def test_NoSuchFile(self):
185
self.assertTranslationEqual(
186
('NoSuchFile', 'path'), errors.NoSuchFile('path'))
188
def test_LockContention(self):
189
# For now, LockContentions are always transmitted with no details.
190
# Eventually they should include a relpath or url or something else to
191
# identify which lock is busy.
192
self.assertTranslationEqual(
193
('LockContention',), errors.LockContention('lock', 'msg'))
195
def test_TokenMismatch(self):
196
self.assertTranslationEqual(
197
('TokenMismatch', 'some-token', 'actual-token'),
198
errors.TokenMismatch('some-token', 'actual-token'))
200
def test_MemoryError(self):
201
self.assertTranslationEqual(("MemoryError",), MemoryError())
203
def test_generic_Exception(self):
204
self.assertTranslationEqual(('error', 'Exception', ""),
207
def test_generic_BzrError(self):
208
self.assertTranslationEqual(('error', 'BzrError', "some text"),
209
errors.BzrError(msg="some text"))
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),
218
class TestRequestJail(TestCaseWithMemoryTransport):
221
transport = self.get_transport('blah')
222
req = request.SmartServerRequest(transport)
223
self.assertEqual(None, request.jail_info.transports)
225
self.assertEqual([transport], request.jail_info.transports)
227
self.assertEqual(None, request.jail_info.transports)
230
class TestJailHook(TestCaseWithMemoryTransport):
233
super(TestJailHook, self).setUp()
234
def clear_jail_info():
235
request.jail_info.transports = None
236
self.addCleanup(clear_jail_info)
238
def test_jail_hook(self):
239
request.jail_info.transports = None
240
_pre_open_hook = request._pre_open_hook
241
# Any transport is fine if jail_info.transports is None
242
t = self.get_transport('foo')
244
# A transport in jail_info.transports is allowed
245
request.jail_info.transports = [t]
247
# A child of a transport in jail_info is allowed
248
_pre_open_hook(t.clone('child'))
249
# A parent is not allowed
250
self.assertRaises(errors.JailBreak, _pre_open_hook, t.clone('..'))
251
# A completely unrelated transport is not allowed
252
self.assertRaises(errors.JailBreak, _pre_open_hook,
253
transport.get_transport_from_url('http://host/'))
255
def test_open_bzrdir_in_non_main_thread(self):
256
"""Opening a bzrdir in a non-main thread should work ok.
258
This makes sure that the globally-installed
259
bzrlib.smart.request._pre_open_hook, which uses a threading.local(),
260
works in a newly created thread.
262
bzrdir = self.make_bzrdir('.')
263
transport = bzrdir.root_transport
266
BzrDir.open_from_transport(transport)
267
thread_result.append('ok')
268
thread = threading.Thread(target=t)
271
self.assertEqual(['ok'], thread_result)
added
removed
bzrlib/tests/test_smart_request.py
