~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_wsgi.py

  • Committer: Alexander Belchenko
  • Date: 2007-01-24 13:03:32 UTC
  • mto: This revision was merged to the branch mainline in revision 2242.
  • Revision ID: bialix@ukr.net-20070124130332-ane2eqz3eqrtm9u1
Use new API for testing

Show diffs side-by-side

added added

removed removed

Lines of Context:
 
1
# Copyright (C) 2006 Canonical Ltd
 
2
#
 
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.
 
7
#
 
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.
 
12
#
 
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
 
16
 
 
17
"""Tests for WSGI application"""
 
18
 
 
19
from cStringIO import StringIO
 
20
 
 
21
from bzrlib import tests
 
22
from bzrlib.transport.http import wsgi
 
23
from bzrlib.transport import chroot, memory
 
24
 
 
25
 
 
26
class TestWSGI(tests.TestCase):
 
27
 
 
28
    def setUp(self):
 
29
        tests.TestCase.setUp(self)
 
30
        self.status = None
 
31
        self.headers = None
 
32
 
 
33
    def build_environ(self, updates=None):
 
34
        """Builds an environ dict with all fields required by PEP 333.
 
35
        
 
36
        :param updates: a dict to that will be incorporated into the returned
 
37
            dict using dict.update(updates).
 
38
        """
 
39
        environ = {
 
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',
 
47
 
 
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,
 
56
        }
 
57
        if updates is not None:
 
58
            environ.update(updates)
 
59
        return environ
 
60
        
 
61
    def read_response(self, iterable):
 
62
        response = ''
 
63
        for string in iterable:
 
64
            response += string
 
65
        return response
 
66
 
 
67
    def start_response(self, status, headers):
 
68
        self.status = status
 
69
        self.headers = headers
 
70
 
 
71
    def test_construct(self):
 
72
        app = wsgi.SmartWSGIApp(FakeTransport())
 
73
        self.assertIsInstance(
 
74
            app.backing_transport, chroot.ChrootTransportDecorator)
 
75
 
 
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)
 
84
        
 
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
 
94
            return 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',
 
102
        })
 
103
        iterable = wsgi_app(environ, self.start_response)
 
104
        response = self.read_response(iterable)
 
105
        self.assertEqual([('clone', 'foo/bar')] , transport.calls)
 
106
 
 
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
 
117
            return 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',
 
125
        })
 
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)
 
130
 
 
131
    def test_relpath_setter(self):
 
132
        # wsgi.RelpathSetter is WSGI "middleware" to set the 'bzrlib.relpath'
 
133
        # variable.
 
134
        calls = []
 
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)
 
141
       
 
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'))
 
154
        
 
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'))
 
168
        
 
169
    def test_make_app(self):
 
170
        # The make_app helper constructs a SmartWSGIApp wrapped in a
 
171
        # RelpathSetter.
 
172
        app = wsgi.make_app(
 
173
            root='a root',
 
174
            prefix='a prefix',
 
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')
 
181
 
 
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
 
188
            return request
 
189
        wsgi_app.make_request = make_request
 
190
 
 
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',
 
197
        })
 
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)
 
202
 
 
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'))
 
210
 
 
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': '.',
 
217
        })
 
218
        iterable = wsgi_app(environ, self.start_response)
 
219
        response = self.read_response(iterable)
 
220
        self.assertEqual('200 OK', self.status)
 
221
        self.assertEqual(
 
222
            "error\x01Path '/bad file' is not a child of "
 
223
            "path 'memory:///foo/'\n",
 
224
            response)
 
225
 
 
226
 
 
227
class FakeRequest(object):
 
228
    
 
229
    def __init__(self, transport, write_func):
 
230
        self.transport = transport
 
231
        self.write_func = write_func
 
232
        self.accepted_bytes = ''
 
233
 
 
234
    def accept_bytes(self, bytes):
 
235
        self.accepted_bytes = bytes
 
236
        self.write_func('got bytes: ' + bytes)
 
237
 
 
238
    def next_read_size(self):
 
239
        return 0
 
240
 
 
241
 
 
242
class FakeTransport(object):
 
243
 
 
244
    def __init__(self):
 
245
        self.calls = []
 
246
        self.base = 'fake:///'
 
247
 
 
248
    def abspath(self, relpath):
 
249
        return 'fake:///' + relpath
 
250
 
 
251
    def clone(self, relpath):
 
252
        self.calls.append(('clone', relpath))
 
253
        return self
 
254
 
 
255
 
 
256
class IncompleteRequest(FakeRequest):
 
257
    """A request-like object that always expects to read more bytes."""
 
258
 
 
259
    def next_read_size(self):
 
260
        # this request always asks for more
 
261
        return 1
 
262