19
19
from cStringIO import StringIO
21
21
from bzrlib import tests
22
from bzrlib.smart import medium, message, protocol
22
from bzrlib.smart import protocol
23
23
from bzrlib.transport.http import wsgi
24
24
from bzrlib.transport import chroot, memory
27
class WSGITestMixin(object):
27
class TestWSGI(tests.TestCase):
30
tests.TestCase.setUp(self)
29
34
def build_environ(self, updates=None):
30
35
"""Builds an environ dict with all fields required by PEP 333.
245
242
protocol.RESPONSE_VERSION_TWO + 'success\nok\x012\n', response)
248
class TestWSGIJail(tests.TestCaseWithMemoryTransport, WSGITestMixin):
250
def make_hpss_wsgi_request(self, wsgi_relpath, *args):
251
write_buf = StringIO()
252
request_medium = medium.SmartSimplePipesClientMedium(
253
None, write_buf, 'fake:' + wsgi_relpath)
254
request_encoder = protocol.ProtocolThreeRequester(
255
request_medium.get_request())
256
request_encoder.call(*args)
258
environ = self.build_environ({
259
'REQUEST_METHOD': 'POST',
260
'CONTENT_LENGTH': len(write_buf.getvalue()),
261
'wsgi.input': write_buf,
262
'bzrlib.relpath': wsgi_relpath,
266
def test_jail_root(self):
267
"""The WSGI HPSS glue allows access to the whole WSGI backing
268
transport, regardless of which HTTP path the request was delivered
271
# make a branch in a shared repo
272
self.make_repository('repo', shared=True)
273
branch = self.make_bzrdir('repo/branch').create_branch()
274
# serve the repo via bzr+http WSGI
275
wsgi_app = wsgi.SmartWSGIApp(self.get_transport())
276
# send a request to /repo/branch that will have to access /repo.
277
environ = self.make_hpss_wsgi_request(
278
'/repo/branch', 'BzrDir.open_branchV2', '.')
279
iterable = wsgi_app(environ, self.start_response)
280
response_bytes = self.read_response(iterable)
281
self.assertEqual('200 OK', self.status)
282
# expect a successful response, rather than a jail break error
283
from bzrlib.tests.test_smart_transport import LoggingMessageHandler
284
message_handler = LoggingMessageHandler()
285
decoder = protocol.ProtocolThreeDecoder(
286
message_handler, expect_version_marker=True)
287
decoder.accept_bytes(response_bytes)
289
('structure', ('branch', branch._format.network_name()))
290
in message_handler.event_log)
293
245
class FakeRequest(object):
295
247
def __init__(self, transport, write_func):