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.ChrootTransportDecorator)
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 construct the transport for this request, by cloning
88
# its base transport with the given relpath.
89
transport = FakeTransport()
90
wsgi_app = wsgi.SmartWSGIApp(transport)
91
def make_request(transport, write_func):
92
request = FakeRequest(transport, write_func)
93
self.request = request
95
wsgi_app.make_request = make_request
96
fake_input = StringIO('fake request')
97
environ = self.build_environ({
98
'REQUEST_METHOD': 'POST',
99
'CONTENT_LENGTH': len(fake_input.getvalue()),
100
'wsgi.input': fake_input,
101
'bzrlib.relpath': 'foo/bar',
103
iterable = wsgi_app(environ, self.start_response)
104
response = self.read_response(iterable)
105
self.assertEqual([('clone', 'foo/bar')] , transport.calls)
107
def test_smart_wsgi_app_request_and_response(self):
108
# SmartWSGIApp reads the smart request from the 'wsgi.input' file-like
109
# object in the environ dict, and returns the response via the iterable
110
# returned to the WSGI handler.
111
transport = memory.MemoryTransport()
112
transport.put_bytes('foo', 'some bytes')
113
wsgi_app = wsgi.SmartWSGIApp(transport)
114
def make_request(transport, write_func):
115
request = FakeRequest(transport, write_func)
116
self.request = request
118
wsgi_app.make_request = make_request
119
fake_input = StringIO('fake request')
120
environ = self.build_environ({
121
'REQUEST_METHOD': 'POST',
122
'CONTENT_LENGTH': len(fake_input.getvalue()),
123
'wsgi.input': fake_input,
124
'bzrlib.relpath': 'foo',
126
iterable = wsgi_app(environ, self.start_response)
127
response = self.read_response(iterable)
128
self.assertEqual('200 OK', self.status)
129
self.assertEqual('got bytes: fake request', response)
131
def test_relpath_setter(self):
132
# wsgi.RelpathSetter is WSGI "middleware" to set the 'bzrlib.relpath'
135
def fake_app(environ, start_response):
136
calls.append(environ['bzrlib.relpath'])
137
wrapped_app = wsgi.RelpathSetter(
138
fake_app, prefix='/abc/', path_var='FOO')
139
wrapped_app({'FOO': '/abc/xyz/.bzr/smart'}, None)
140
self.assertEqual(['xyz'], calls)
142
def test_relpath_setter_bad_path_prefix(self):
143
# wsgi.RelpathSetter will reject paths with that don't match the prefix
144
# with a 404. This is probably a sign of misconfiguration; a server
145
# shouldn't ever be invoking our WSGI application with bad paths.
146
def fake_app(environ, start_response):
147
self.fail('The app should never be called when the path is wrong')
148
wrapped_app = wsgi.RelpathSetter(
149
fake_app, prefix='/abc/', path_var='FOO')
150
iterable = wrapped_app(
151
{'FOO': 'AAA/abc/xyz/.bzr/smart'}, self.start_response)
152
self.read_response(iterable)
153
self.assertTrue(self.status.startswith('404'))
155
def test_relpath_setter_bad_path_suffix(self):
156
# Similar to test_relpath_setter_bad_path_prefix: wsgi.RelpathSetter
157
# will reject paths with that don't match the suffix '.bzr/smart' with a
158
# 404 as well. Again, this shouldn't be seen by our WSGI application if
159
# the server is configured correctly.
160
def fake_app(environ, start_response):
161
self.fail('The app should never be called when the path is wrong')
162
wrapped_app = wsgi.RelpathSetter(
163
fake_app, prefix='/abc/', path_var='FOO')
164
iterable = wrapped_app(
165
{'FOO': '/abc/xyz/.bzr/AAA'}, self.start_response)
166
self.read_response(iterable)
167
self.assertTrue(self.status.startswith('404'))
169
def test_make_app(self):
170
# The make_app helper constructs a SmartWSGIApp wrapped in a
175
path_var='a path_var')
176
self.assertIsInstance(app, wsgi.RelpathSetter)
177
self.assertIsInstance(app.app, wsgi.SmartWSGIApp)
178
self.assertEndsWith(app.app.backing_transport.base, 'a%20root/')
179
self.assertEqual(app.prefix, 'a prefix')
180
self.assertEqual(app.path_var, 'a path_var')
182
def test_incomplete_request(self):
183
transport = FakeTransport()
184
wsgi_app = wsgi.SmartWSGIApp(transport)
185
def make_request(transport, write_func):
186
request = IncompleteRequest(transport, write_func)
187
self.request = request
189
wsgi_app.make_request = make_request
191
fake_input = StringIO('incomplete request')
192
environ = self.build_environ({
193
'REQUEST_METHOD': 'POST',
194
'CONTENT_LENGTH': len(fake_input.getvalue()),
195
'wsgi.input': fake_input,
196
'bzrlib.relpath': 'foo/bar',
198
iterable = wsgi_app(environ, self.start_response)
199
response = self.read_response(iterable)
200
self.assertEqual('200 OK', self.status)
201
self.assertEqual('error\x01incomplete request\n', response)
203
def test_chrooting(self):
204
# Show that requests that try to access things outside of the base
205
# really will get intercepted by the ChrootTransportDecorator.
206
transport = memory.MemoryTransport()
207
transport.mkdir('foo')
208
transport.put_bytes('foo/bar', 'this is foo/bar')
209
wsgi_app = wsgi.SmartWSGIApp(transport.clone('foo'))
211
smart_request = StringIO('mkdir\x01/bad file\x01\n0\ndone\n')
212
environ = self.build_environ({
213
'REQUEST_METHOD': 'POST',
214
'CONTENT_LENGTH': len(smart_request.getvalue()),
215
'wsgi.input': smart_request,
216
'bzrlib.relpath': '.',
218
iterable = wsgi_app(environ, self.start_response)
219
response = self.read_response(iterable)
220
self.assertEqual('200 OK', self.status)
222
"error\x01Path '/bad file' is not a child of "
223
"path 'memory:///foo/'\n",
227
class FakeRequest(object):
229
def __init__(self, transport, write_func):
230
self.transport = transport
231
self.write_func = write_func
232
self.accepted_bytes = ''
234
def accept_bytes(self, bytes):
235
self.accepted_bytes = bytes
236
self.write_func('got bytes: ' + bytes)
238
def next_read_size(self):
242
class FakeTransport(object):
246
self.base = 'fake:///'
248
def abspath(self, relpath):
249
return 'fake:///' + relpath
251
def clone(self, relpath):
252
self.calls.append(('clone', relpath))
256
class IncompleteRequest(FakeRequest):
257
"""A request-like object that always expects to read more bytes."""
259
def next_read_size(self):
260
# this request always asks for more
added
removed
bzrlib/tests/test_wsgi.py
