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 test_smart_wsgi_app_uses_given_relpath(self):
86
# The SmartWSGIApp should use the "bzrlib.relpath" field from the
87
# WSGI environ to clone from its backing transport to get a specific
88
# transport for this request.
89
transport = FakeTransport()
90
wsgi_app = wsgi.SmartWSGIApp(transport)
91
wsgi_app.backing_transport = transport
92
def make_request(transport, write_func):
93
request = FakeRequest(transport, write_func)
94
self.request = request
96
wsgi_app.make_request = make_request
97
fake_input = StringIO('fake request')
98
environ = self.build_environ({
99
'REQUEST_METHOD': 'POST',
100
'CONTENT_LENGTH': len(fake_input.getvalue()),
101
'wsgi.input': fake_input,
102
'bzrlib.relpath': 'foo/bar',
104
iterable = wsgi_app(environ, self.start_response)
105
response = self.read_response(iterable)
106
self.assertEqual([('clone', 'foo/bar')] , transport.calls)
108
def test_smart_wsgi_app_request_and_response(self):
109
# SmartWSGIApp reads the smart request from the 'wsgi.input' file-like
110
# object in the environ dict, and returns the response via the iterable
111
# returned to the WSGI handler.
112
transport = memory.MemoryTransport()
113
transport.put_bytes('foo', 'some bytes')
114
wsgi_app = wsgi.SmartWSGIApp(transport)
115
def make_request(transport, write_func):
116
request = FakeRequest(transport, write_func)
117
self.request = request
119
wsgi_app.make_request = make_request
120
fake_input = StringIO('fake request')
121
environ = self.build_environ({
122
'REQUEST_METHOD': 'POST',
123
'CONTENT_LENGTH': len(fake_input.getvalue()),
124
'wsgi.input': fake_input,
125
'bzrlib.relpath': 'foo',
127
iterable = wsgi_app(environ, self.start_response)
128
response = self.read_response(iterable)
129
self.assertEqual('200 OK', self.status)
130
self.assertEqual('got bytes: fake request', response)
132
def test_relpath_setter(self):
133
# wsgi.RelpathSetter is WSGI "middleware" to set the 'bzrlib.relpath'
136
def fake_app(environ, start_response):
137
calls.append(environ['bzrlib.relpath'])
138
wrapped_app = wsgi.RelpathSetter(
139
fake_app, prefix='/abc/', path_var='FOO')
140
wrapped_app({'FOO': '/abc/xyz/.bzr/smart'}, None)
141
self.assertEqual(['xyz'], calls)
143
def test_relpath_setter_bad_path_prefix(self):
144
# wsgi.RelpathSetter will reject paths with that don't match the prefix
145
# with a 404. This is probably a sign of misconfiguration; a server
146
# shouldn't ever be invoking our WSGI application with bad paths.
147
def fake_app(environ, start_response):
148
self.fail('The app should never be called when the path is wrong')
149
wrapped_app = wsgi.RelpathSetter(
150
fake_app, prefix='/abc/', path_var='FOO')
151
iterable = wrapped_app(
152
{'FOO': 'AAA/abc/xyz/.bzr/smart'}, self.start_response)
153
self.read_response(iterable)
154
self.assertTrue(self.status.startswith('404'))
156
def test_relpath_setter_bad_path_suffix(self):
157
# Similar to test_relpath_setter_bad_path_prefix: wsgi.RelpathSetter
158
# will reject paths with that don't match the suffix '.bzr/smart' with a
159
# 404 as well. Again, this shouldn't be seen by our WSGI application if
160
# the server is configured correctly.
161
def fake_app(environ, start_response):
162
self.fail('The app should never be called when the path is wrong')
163
wrapped_app = wsgi.RelpathSetter(
164
fake_app, prefix='/abc/', path_var='FOO')
165
iterable = wrapped_app(
166
{'FOO': '/abc/xyz/.bzr/AAA'}, self.start_response)
167
self.read_response(iterable)
168
self.assertTrue(self.status.startswith('404'))
170
def test_make_app(self):
171
# The make_app helper constructs a SmartWSGIApp wrapped in a
176
path_var='a path_var')
177
self.assertIsInstance(app, wsgi.RelpathSetter)
178
self.assertIsInstance(app.app, wsgi.SmartWSGIApp)
179
self.assertStartsWith(app.app.backing_transport.base, 'chroot-')
180
backing_transport = app.app.backing_transport
181
chroot_backing_transport = backing_transport.server.backing_transport
182
self.assertEndsWith(chroot_backing_transport.base, 'a%20root/')
183
self.assertEqual(app.prefix, 'a prefix')
184
self.assertEqual(app.path_var, 'a path_var')
186
def test_incomplete_request(self):
187
transport = FakeTransport()
188
wsgi_app = wsgi.SmartWSGIApp(transport)
189
def make_request(transport, write_func):
190
request = IncompleteRequest(transport, write_func)
191
self.request = request
193
wsgi_app.make_request = make_request
195
fake_input = StringIO('incomplete request')
196
environ = self.build_environ({
197
'REQUEST_METHOD': 'POST',
198
'CONTENT_LENGTH': len(fake_input.getvalue()),
199
'wsgi.input': fake_input,
200
'bzrlib.relpath': 'foo/bar',
202
iterable = wsgi_app(environ, self.start_response)
203
response = self.read_response(iterable)
204
self.assertEqual('200 OK', self.status)
205
self.assertEqual('error\x01incomplete request\n', response)
208
class FakeRequest(object):
210
def __init__(self, transport, write_func):
211
self.transport = transport
212
self.write_func = write_func
213
self.accepted_bytes = ''
215
def accept_bytes(self, bytes):
216
self.accepted_bytes = bytes
217
self.write_func('got bytes: ' + bytes)
219
def next_read_size(self):
223
class FakeTransport(object):
227
self.base = 'fake:///'
229
def abspath(self, relpath):
230
return 'fake:///' + relpath
232
def clone(self, relpath):
233
self.calls.append(('clone', relpath))
237
class IncompleteRequest(FakeRequest):
238
"""A request-like object that always expects to read more bytes."""
240
def next_read_size(self):
241
# this request always asks for more
added
removed
bzrlib/tests/test_wsgi.py
