17
17
"""Tests for WSGI application"""
20
from ..sixish import (
23
from ..bzr.smart import medium, protocol
24
from ..transport.http import wsgi
25
from ..transport import chroot, memory
19
from cStringIO import StringIO
21
from bzrlib import tests
22
from bzrlib.smart import medium, protocol
23
from bzrlib.transport.http import wsgi
24
from bzrlib.transport import chroot, memory
28
27
class WSGITestMixin(object):
43
42
'SERVER_PROTOCOL': 'HTTP/1.0',
45
44
# Required WSGI variables
46
'wsgi.version': (1, 0),
45
'wsgi.version': (1,0),
47
46
'wsgi.url_scheme': 'http',
48
'wsgi.input': BytesIO(b''),
49
'wsgi.errors': BytesIO(),
47
'wsgi.input': StringIO(''),
48
'wsgi.errors': StringIO(),
50
49
'wsgi.multithread': False,
51
50
'wsgi.multiprocess': False,
52
51
'wsgi.run_once': True,
96
95
def test_smart_wsgi_app_uses_given_relpath(self):
97
# The SmartWSGIApp should use the "breezy.relpath" field from the
96
# The SmartWSGIApp should use the "bzrlib.relpath" field from the
98
97
# WSGI environ to clone from its backing transport to get a specific
99
98
# transport for this request.
100
99
transport = FakeTransport()
101
100
wsgi_app = wsgi.SmartWSGIApp(transport)
102
101
wsgi_app.backing_transport = transport
103
102
wsgi_app.make_request = self._fake_make_request
104
fake_input = BytesIO(b'fake request')
103
fake_input = StringIO('fake request')
105
104
environ = self.build_environ({
106
105
'REQUEST_METHOD': 'POST',
107
106
'CONTENT_LENGTH': len(fake_input.getvalue()),
108
107
'wsgi.input': fake_input,
109
'breezy.relpath': 'foo/bar',
108
'bzrlib.relpath': 'foo/bar',
111
110
iterable = wsgi_app(environ, self.start_response)
112
111
response = self.read_response(iterable)
113
self.assertEqual([('clone', 'foo/bar/')], transport.calls)
112
self.assertEqual([('clone', 'foo/bar/')] , transport.calls)
115
114
def test_smart_wsgi_app_request_and_response(self):
116
115
# SmartWSGIApp reads the smart request from the 'wsgi.input' file-like
120
119
transport.put_bytes('foo', 'some bytes')
121
120
wsgi_app = wsgi.SmartWSGIApp(transport)
122
121
wsgi_app.make_request = self._fake_make_request
123
fake_input = BytesIO(b'fake request')
122
fake_input = StringIO('fake request')
124
123
environ = self.build_environ({
125
124
'REQUEST_METHOD': 'POST',
126
125
'CONTENT_LENGTH': len(fake_input.getvalue()),
127
126
'wsgi.input': fake_input,
128
'breezy.relpath': 'foo',
127
'bzrlib.relpath': 'foo',
130
129
iterable = wsgi_app(environ, self.start_response)
131
130
response = self.read_response(iterable)
133
132
self.assertEqual('got bytes: fake request', response)
135
134
def test_relpath_setter(self):
136
# wsgi.RelpathSetter is WSGI "middleware" to set the 'breezy.relpath'
135
# wsgi.RelpathSetter is WSGI "middleware" to set the 'bzrlib.relpath'
139
138
def fake_app(environ, start_response):
140
calls.append(environ['breezy.relpath'])
139
calls.append(environ['bzrlib.relpath'])
141
140
wrapped_app = wsgi.RelpathSetter(
142
141
fake_app, prefix='/abc/', path_var='FOO')
143
142
wrapped_app({'FOO': '/abc/xyz/.bzr/smart'}, None)
197
196
wsgi_app.make_request = make_request
199
fake_input = BytesIO(b'incomplete request')
198
fake_input = StringIO('incomplete request')
200
199
environ = self.build_environ({
201
200
'REQUEST_METHOD': 'POST',
202
201
'CONTENT_LENGTH': len(fake_input.getvalue()),
203
202
'wsgi.input': fake_input,
204
'breezy.relpath': 'foo/bar',
203
'bzrlib.relpath': 'foo/bar',
206
205
iterable = wsgi_app(environ, self.start_response)
207
206
response = self.read_response(iterable)
213
212
# REQUEST_VERSION_TWO as version one.
214
213
transport = memory.MemoryTransport()
215
214
wsgi_app = wsgi.SmartWSGIApp(transport)
216
fake_input = BytesIO(b'hello\n')
215
fake_input = StringIO('hello\n')
217
216
environ = self.build_environ({
218
217
'REQUEST_METHOD': 'POST',
219
218
'CONTENT_LENGTH': len(fake_input.getvalue()),
220
219
'wsgi.input': fake_input,
221
'breezy.relpath': 'foo',
220
'bzrlib.relpath': 'foo',
223
222
iterable = wsgi_app(environ, self.start_response)
224
223
response = self.read_response(iterable)
231
230
# as version two.
232
231
transport = memory.MemoryTransport()
233
232
wsgi_app = wsgi.SmartWSGIApp(transport)
234
fake_input = BytesIO(protocol.REQUEST_VERSION_TWO + 'hello\n')
233
fake_input = StringIO(protocol.REQUEST_VERSION_TWO + 'hello\n')
235
234
environ = self.build_environ({
236
235
'REQUEST_METHOD': 'POST',
237
236
'CONTENT_LENGTH': len(fake_input.getvalue()),
238
237
'wsgi.input': fake_input,
239
'breezy.relpath': 'foo',
238
'bzrlib.relpath': 'foo',
241
240
iterable = wsgi_app(environ, self.start_response)
242
241
response = self.read_response(iterable)
249
248
class TestWSGIJail(tests.TestCaseWithMemoryTransport, WSGITestMixin):
251
250
def make_hpss_wsgi_request(self, wsgi_relpath, *args):
252
write_buf = BytesIO()
251
write_buf = StringIO()
253
252
request_medium = medium.SmartSimplePipesClientMedium(
254
253
None, write_buf, 'fake:' + wsgi_relpath)
255
254
request_encoder = protocol.ProtocolThreeRequester(
260
259
'REQUEST_METHOD': 'POST',
261
260
'CONTENT_LENGTH': len(write_buf.getvalue()),
262
261
'wsgi.input': write_buf,
263
'breezy.relpath': wsgi_relpath,
262
'bzrlib.relpath': wsgi_relpath,
272
271
# make a branch in a shared repo
273
272
self.make_repository('repo', shared=True)
274
branch = self.make_controldir('repo/branch').create_branch()
273
branch = self.make_bzrdir('repo/branch').create_branch()
275
274
# serve the repo via bzr+http WSGI
276
275
wsgi_app = wsgi.SmartWSGIApp(self.get_transport())
277
276
# send a request to /repo/branch that will have to access /repo.
281
280
response_bytes = self.read_response(iterable)
282
281
self.assertEqual('200 OK', self.status)
283
282
# expect a successful response, rather than a jail break error
284
from breezy.tests.test_smart_transport import LoggingMessageHandler
283
from bzrlib.tests.test_smart_transport import LoggingMessageHandler
285
284
message_handler = LoggingMessageHandler()
286
285
decoder = protocol.ProtocolThreeDecoder(
287
286
message_handler, expect_version_marker=True)