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