1
# Copyright (C) 2006 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., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
17
"""Tests for WSGI application"""
19
from cStringIO import StringIO
21
from bzrlib import tests
22
from bzrlib.transport.http import wsgi
23
from bzrlib.transport import chroot, memory
26
class TestWSGI(tests.TestCase):
29
tests.TestCase.setUp(self)
33
def build_environ(self, updates=None):
34
"""Builds an environ dict with all fields required by PEP 333.
36
:param updates: a dict to that will be incorporated into the returned
37
dict using dict.update(updates).
40
# Required CGI variables
41
'REQUEST_METHOD': 'GET',
42
'SCRIPT_NAME': '/script/name/',
43
'PATH_INFO': 'path/info',
44
'SERVER_NAME': 'test',
45
'SERVER_PORT': '9999',
46
'SERVER_PROTOCOL': 'HTTP/1.0',
48
# Required WSGI variables
49
'wsgi.version': (1,0),
50
'wsgi.url_scheme': 'http',
51
'wsgi.input': StringIO(''),
52
'wsgi.errors': StringIO(),
53
'wsgi.multithread': False,
54
'wsgi.multiprocess': False,
55
'wsgi.run_once': True,
57
if updates is not None:
58
environ.update(updates)
61
def read_response(self, iterable):
63
for string in iterable:
67
def start_response(self, status, headers):
69
self.headers = headers
71
def test_construct(self):
72
app = wsgi.SmartWSGIApp(FakeTransport())
73
self.assertIsInstance(
74
app.backing_transport, chroot.ChrootTransport)
76
def test_http_get_rejected(self):
77
# GET requests are rejected.
78
app = wsgi.SmartWSGIApp(FakeTransport())
79
environ = self.build_environ({'REQUEST_METHOD': 'GET'})
80
iterable = app(environ, self.start_response)
81
self.read_response(iterable)
82
self.assertEqual('405 Method not allowed', self.status)
83
self.assertTrue(('Allow', 'POST') in self.headers)
85
def _fake_make_request(self, transport, write_func, bytes):
86
request = FakeRequest(transport, write_func)
87
request.accept_bytes(bytes)
88
self.request = request
91
def test_smart_wsgi_app_uses_given_relpath(self):
92
# The SmartWSGIApp should use the "bzrlib.relpath" field from the
93
# WSGI environ to clone from its backing transport to get a specific
94
# transport for this request.
95
transport = FakeTransport()
96
wsgi_app = wsgi.SmartWSGIApp(transport)
97
wsgi_app.backing_transport = transport
98
wsgi_app.make_request = self._fake_make_request
99
fake_input = StringIO('fake request')
100
environ = self.build_environ({
101
'REQUEST_METHOD': 'POST',
102
'CONTENT_LENGTH': len(fake_input.getvalue()),
103
'wsgi.input': fake_input,
104
'bzrlib.relpath': 'foo/bar',
106
iterable = wsgi_app(environ, self.start_response)
107
response = self.read_response(iterable)
108
self.assertEqual([('clone', 'foo/bar')] , transport.calls)
110
def test_smart_wsgi_app_request_and_response(self):
111
# SmartWSGIApp reads the smart request from the 'wsgi.input' file-like
112
# object in the environ dict, and returns the response via the iterable
113
# returned to the WSGI handler.
114
transport = memory.MemoryTransport()
115
transport.put_bytes('foo', 'some bytes')
116
wsgi_app = wsgi.SmartWSGIApp(transport)
117
wsgi_app.make_request = self._fake_make_request
118
fake_input = StringIO('fake request')
119
environ = self.build_environ({
120
'REQUEST_METHOD': 'POST',
121
'CONTENT_LENGTH': len(fake_input.getvalue()),
122
'wsgi.input': fake_input,
123
'bzrlib.relpath': 'foo',
125
iterable = wsgi_app(environ, self.start_response)
126
response = self.read_response(iterable)
127
self.assertEqual('200 OK', self.status)
128
self.assertEqual('got bytes: fake request', response)
130
def test_relpath_setter(self):
131
# wsgi.RelpathSetter is WSGI "middleware" to set the 'bzrlib.relpath'
134
def fake_app(environ, start_response):
135
calls.append(environ['bzrlib.relpath'])
136
wrapped_app = wsgi.RelpathSetter(
137
fake_app, prefix='/abc/', path_var='FOO')
138
wrapped_app({'FOO': '/abc/xyz/.bzr/smart'}, None)
139
self.assertEqual(['xyz'], calls)
141
def test_relpath_setter_bad_path_prefix(self):
142
# wsgi.RelpathSetter will reject paths with that don't match the prefix
143
# with a 404. This is probably a sign of misconfiguration; a server
144
# shouldn't ever be invoking our WSGI application with bad paths.
145
def fake_app(environ, start_response):
146
self.fail('The app should never be called when the path is wrong')
147
wrapped_app = wsgi.RelpathSetter(
148
fake_app, prefix='/abc/', path_var='FOO')
149
iterable = wrapped_app(
150
{'FOO': 'AAA/abc/xyz/.bzr/smart'}, self.start_response)
151
self.read_response(iterable)
152
self.assertTrue(self.status.startswith('404'))
154
def test_relpath_setter_bad_path_suffix(self):
155
# Similar to test_relpath_setter_bad_path_prefix: wsgi.RelpathSetter
156
# will reject paths with that don't match the suffix '.bzr/smart' with a
157
# 404 as well. Again, this shouldn't be seen by our WSGI application if
158
# the server is configured correctly.
159
def fake_app(environ, start_response):
160
self.fail('The app should never be called when the path is wrong')
161
wrapped_app = wsgi.RelpathSetter(
162
fake_app, prefix='/abc/', path_var='FOO')
163
iterable = wrapped_app(
164
{'FOO': '/abc/xyz/.bzr/AAA'}, self.start_response)
165
self.read_response(iterable)
166
self.assertTrue(self.status.startswith('404'))
168
def test_make_app(self):
169
# The make_app helper constructs a SmartWSGIApp wrapped in a
174
path_var='a path_var')
175
self.assertIsInstance(app, wsgi.RelpathSetter)
176
self.assertIsInstance(app.app, wsgi.SmartWSGIApp)
177
self.assertStartsWith(app.app.backing_transport.base, 'chroot-')
178
backing_transport = app.app.backing_transport
179
chroot_backing_transport = backing_transport.server.backing_transport
180
self.assertEndsWith(chroot_backing_transport.base, 'a%20root/')
181
self.assertEqual(app.prefix, 'a prefix')
182
self.assertEqual(app.path_var, 'a path_var')
184
def test_incomplete_request(self):
185
transport = FakeTransport()
186
wsgi_app = wsgi.SmartWSGIApp(transport)
187
def make_request(transport, write_func, bytes):
188
request = IncompleteRequest(transport, write_func)
189
request.accept_bytes(bytes)
190
self.request = request
192
wsgi_app.make_request = make_request
194
fake_input = StringIO('incomplete request')
195
environ = self.build_environ({
196
'REQUEST_METHOD': 'POST',
197
'CONTENT_LENGTH': len(fake_input.getvalue()),
198
'wsgi.input': fake_input,
199
'bzrlib.relpath': 'foo/bar',
201
iterable = wsgi_app(environ, self.start_response)
202
response = self.read_response(iterable)
203
self.assertEqual('200 OK', self.status)
204
self.assertEqual('error\x01incomplete request\n', response)
206
def test_protocol_version_detection_one(self):
207
# SmartWSGIApp detects requests that don't start with '2\n' as version
209
transport = memory.MemoryTransport()
210
wsgi_app = wsgi.SmartWSGIApp(transport)
211
fake_input = StringIO('hello\n')
212
environ = self.build_environ({
213
'REQUEST_METHOD': 'POST',
214
'CONTENT_LENGTH': len(fake_input.getvalue()),
215
'wsgi.input': fake_input,
216
'bzrlib.relpath': 'foo',
218
iterable = wsgi_app(environ, self.start_response)
219
response = self.read_response(iterable)
220
self.assertEqual('200 OK', self.status)
221
# Expect a version 1-encoded response.
222
self.assertEqual('ok\x012\n', response)
224
def test_protocol_version_detection_two(self):
225
# SmartWSGIApp detects requests that start with '2\n' as version two.
226
transport = memory.MemoryTransport()
227
wsgi_app = wsgi.SmartWSGIApp(transport)
228
fake_input = StringIO('2\nhello\n')
229
environ = self.build_environ({
230
'REQUEST_METHOD': 'POST',
231
'CONTENT_LENGTH': len(fake_input.getvalue()),
232
'wsgi.input': fake_input,
233
'bzrlib.relpath': 'foo',
235
iterable = wsgi_app(environ, self.start_response)
236
response = self.read_response(iterable)
237
self.assertEqual('200 OK', self.status)
238
# Expect a version 2-encoded response.
239
self.assertEqual('2\nok\x012\n', response)
242
class FakeRequest(object):
244
def __init__(self, transport, write_func):
245
self.transport = transport
246
self.write_func = write_func
247
self.accepted_bytes = ''
249
def accept_bytes(self, bytes):
250
self.accepted_bytes = bytes
251
self.write_func('got bytes: ' + bytes)
253
def next_read_size(self):
257
class FakeTransport(object):
261
self.base = 'fake:///'
263
def abspath(self, relpath):
264
return 'fake:///' + relpath
266
def clone(self, relpath):
267
self.calls.append(('clone', relpath))
271
class IncompleteRequest(FakeRequest):
272
"""A request-like object that always expects to read more bytes."""
274
def next_read_size(self):
275
# this request always asks for more
added
removed
bzrlib/tests/test_wsgi.py
