1
# Copyright (C) 2009 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)."""
21
from bzrlib import errors
22
from bzrlib.bzrdir import BzrDir
23
from bzrlib.smart import request
24
from bzrlib.tests import TestCase, TestCaseWithMemoryTransport
25
from bzrlib.transport import get_transport
28
class NoBodyRequest(request.SmartServerRequest):
29
"""A request that does not implement do_body."""
32
return request.SuccessfulSmartServerResponse(('ok',))
35
class DoErrorRequest(request.SmartServerRequest):
36
"""A request that raises an error from self.do()."""
39
raise errors.NoSuchFile('xyzzy')
42
class ChunkErrorRequest(request.SmartServerRequest):
43
"""A request that raises an error from self.do_chunk()."""
49
def do_chunk(self, bytes):
50
raise errors.NoSuchFile('xyzzy')
53
class EndErrorRequest(request.SmartServerRequest):
54
"""A request that raises an error from self.do_end()."""
60
def do_chunk(self, bytes):
65
raise errors.NoSuchFile('xyzzy')
68
class CheckJailRequest(request.SmartServerRequest):
70
def __init__(self, *args):
71
request.SmartServerRequest.__init__(self, *args)
72
self.jail_transports_log = []
75
self.jail_transports_log.append(request.jail_info.transports)
77
def do_chunk(self, bytes):
78
self.jail_transports_log.append(request.jail_info.transports)
81
self.jail_transports_log.append(request.jail_info.transports)
84
class TestSmartRequest(TestCase):
86
def test_request_class_without_do_body(self):
87
"""If a request has no body data, and the request's implementation does
88
not override do_body, then no exception is raised.
90
# Create a SmartServerRequestHandler with a SmartServerRequest subclass
91
# that does not implement do_body.
92
handler = request.SmartServerRequestHandler(
93
None, {'foo': NoBodyRequest}, '/')
94
# Emulate a request with no body (i.e. just args).
95
handler.args_received(('foo',))
96
handler.end_received()
97
# Request done, no exception was raised.
99
def test_only_request_code_is_jailed(self):
100
transport = 'dummy transport'
101
handler = request.SmartServerRequestHandler(
102
transport, {'foo': CheckJailRequest}, '/')
103
handler.args_received(('foo',))
104
self.assertEqual(None, request.jail_info.transports)
105
handler.accept_body('bytes')
106
self.assertEqual(None, request.jail_info.transports)
107
handler.end_received()
108
self.assertEqual(None, request.jail_info.transports)
110
[[transport]] * 3, handler._command.jail_transports_log)
114
class TestSmartRequestHandlerErrorTranslation(TestCase):
115
"""Tests that SmartServerRequestHandler will translate exceptions raised by
116
a SmartServerRequest into FailedSmartServerResponses.
119
def assertNoResponse(self, handler):
120
self.assertEqual(None, handler.response)
122
def assertResponseIsTranslatedError(self, handler):
123
expected_translation = ('NoSuchFile', 'xyzzy')
125
request.FailedSmartServerResponse(expected_translation),
128
def test_error_translation_from_args_received(self):
129
handler = request.SmartServerRequestHandler(
130
None, {'foo': DoErrorRequest}, '/')
131
handler.args_received(('foo',))
132
self.assertResponseIsTranslatedError(handler)
134
def test_error_translation_from_chunk_received(self):
135
handler = request.SmartServerRequestHandler(
136
None, {'foo': ChunkErrorRequest}, '/')
137
handler.args_received(('foo',))
138
self.assertNoResponse(handler)
139
handler.accept_body('bytes')
140
self.assertResponseIsTranslatedError(handler)
142
def test_error_translation_from_end_received(self):
143
handler = request.SmartServerRequestHandler(
144
None, {'foo': EndErrorRequest}, '/')
145
handler.args_received(('foo',))
146
self.assertNoResponse(handler)
147
handler.end_received()
148
self.assertResponseIsTranslatedError(handler)
151
class TestRequestHanderErrorTranslation(TestCase):
152
"""Tests for bzrlib.smart.request._translate_error."""
154
def assertTranslationEqual(self, expected_tuple, error):
155
self.assertEqual(expected_tuple, request._translate_error(error))
157
def test_NoSuchFile(self):
158
self.assertTranslationEqual(
159
('NoSuchFile', 'path'), errors.NoSuchFile('path'))
161
def test_LockContention(self):
162
# For now, LockContentions are always transmitted with no details.
163
# Eventually they should include a relpath or url or something else to
164
# identify which lock is busy.
165
self.assertTranslationEqual(
166
('LockContention',), errors.LockContention('lock', 'msg'))
168
def test_TokenMismatch(self):
169
self.assertTranslationEqual(
170
('TokenMismatch', 'some-token', 'actual-token'),
171
errors.TokenMismatch('some-token', 'actual-token'))
174
class TestRequestJail(TestCaseWithMemoryTransport):
177
transport = self.get_transport('blah')
178
req = request.SmartServerRequest(transport)
179
self.assertEqual(None, request.jail_info.transports)
181
self.assertEqual([transport], request.jail_info.transports)
183
self.assertEqual(None, request.jail_info.transports)
186
class TestJailHook(TestCaseWithMemoryTransport):
189
request.jail_info.transports = None
190
TestCaseWithMemoryTransport.tearDown(self)
192
def test_jail_hook(self):
193
request.jail_info.transports = None
194
_pre_open_hook = request._pre_open_hook
195
# Any transport is fine if jail_info.transports is None
196
t = self.get_transport('foo')
198
# A transport in jail_info.transports is allowed
199
request.jail_info.transports = [t]
201
# A child of a transport in jail_info is allowed
202
_pre_open_hook(t.clone('child'))
203
# A parent is not allowed
204
self.assertRaises(errors.JailBreak, _pre_open_hook, t.clone('..'))
205
# A completely unrelated transport is not allowed
207
errors.JailBreak, _pre_open_hook, get_transport('http://host/'))
209
def test_open_bzrdir_in_non_main_thread(self):
210
"""Opening a bzrdir in a non-main thread should work ok.
212
This makes sure that the globally-installed
213
bzrlib.smart.request._pre_open_hook, which uses a threading.local(),
214
works in a newly created thread.
216
bzrdir = self.make_bzrdir('.')
217
transport = bzrdir.root_transport
220
BzrDir.open_from_transport(transport)
221
thread_result.append('ok')
222
thread = threading.Thread(target=t)
225
self.assertEqual(['ok'], thread_result)
added
removed
bzrlib/tests/test_smart_request.py
