← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_wsgi.py

Add WSGI smart server.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_wsgi.py
 
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 memory
 
24
 
 
25
class TestWSGI(tests.TestCase):
 
26
 
 
27
    def setUp(self):
 
28
        tests.TestCase.setUp(self)
 
29
        self.status = None
 
30
        self.headers = None
 
31
 
 
32
    def build_environ(self, **kw):
 
33
        """Builds an environ dict with all fields required by PEP 333.
 
34
        
 
35
        The resulting environ dict will be updated with an **kw that are passed.
 
36
        """
 
37
        environ = {
 
38
            # Required CGI variables
 
39
            'REQUEST_METHOD': 'GET',
 
40
            'SCRIPT_NAME': '/script/name/',
 
41
            'PATH_INFO': 'path/info',
 
42
            'SERVER_NAME': 'test',
 
43
            'SERVER_PORT': '9999',
 
44
            'SERVER_PROTOCOL': 'HTTP/1.0',
 
45
 
 
46
            # Required WSGI variables
 
47
            'wsgi.version': (1,0),
 
48
            'wsgi.url_scheme': 'http',
 
49
            'wsgi.input': StringIO(''),
 
50
            'wsgi.errors': StringIO(),
 
51
            'wsgi.multithread': False,
 
52
            'wsgi.multiprocess': False,
 
53
            'wsgi.run_once': True,
 
54
        }
 
55
        environ.update(kw)
 
56
        return environ
 
57
        
 
58
    def read_response(self, iterable):
 
59
        response = ''
 
60
        for string in iterable:
 
61
            response += string
 
62
        return response
 
63
 
 
64
    def start_response(self, status, headers):
 
65
        self.status = status
 
66
        self.headers = headers
 
67
 
 
68
    def test_construct(self):
 
69
        wsgi.SmartWSGIApp(None)
 
70
 
 
71
    def test_http_get_rejected(self):
 
72
        # GET requests are rejected.
 
73
        app = wsgi.SmartWSGIApp(None)
 
74
        environ = self.build_environ(REQUEST_METHOD='GET')
 
75
        iterable = app(environ, self.start_response)
 
76
        self.read_response(iterable)
 
77
        self.assertEqual('405 Method not allowed', self.status)
 
78
        self.assertTrue(('Allow', 'POST') in self.headers)
 
79
        
 
80
    def test_smart_wsgi_app_uses_given_relpath(self):
 
81
        # The SmartWSGIApp should use the "bzrlib.relpath" field from the
 
82
        # WSGI environ to construct the transport for this request, by cloning
 
83
        # its base transport with the given relpath.
 
84
        transport = FakeTransport()
 
85
        wsgi_app = wsgi.SmartWSGIApp(transport)
 
86
        def make_request(transport, write_func):
 
87
            request = FakeRequest(transport, write_func)
 
88
            self.request = request
 
89
            return request
 
90
        wsgi_app.make_request = make_request
 
91
        fake_input = StringIO('fake request')
 
92
        environ = self.build_environ()
 
93
        environ.update({
 
94
            'REQUEST_METHOD': 'POST',
 
95
            'CONTENT_LENGTH': len(fake_input.getvalue()),
 
96
            'wsgi.input': fake_input,
 
97
            'bzrlib.relpath': 'foo/bar',
 
98
        })
 
99
        iterable = wsgi_app(environ, self.start_response)
 
100
        response = self.read_response(iterable)
 
101
        self.assertEqual([('clone', 'foo/bar')] , transport.calls)
 
102
 
 
103
    def test_smart_wsgi_app_request_and_response(self):
 
104
        # SmartWSGIApp reads the smart request from the 'wsgi.input' file-like
 
105
        # object in the environ dict, and returns the response via the iterable
 
106
        # returned to the WSGI handler.
 
107
        transport = memory.MemoryTransport()
 
108
        transport.put_bytes('foo', 'some bytes')
 
109
        wsgi_app = wsgi.SmartWSGIApp(transport)
 
110
        def make_request(transport, write_func):
 
111
            request = FakeRequest(transport, write_func)
 
112
            self.request = request
 
113
            return request
 
114
        wsgi_app.make_request = make_request
 
115
        fake_input = StringIO('fake request')
 
116
        environ = self.build_environ()
 
117
        environ.update({
 
118
            'REQUEST_METHOD': 'POST',
 
119
            'CONTENT_LENGTH': len(fake_input.getvalue()),
 
120
            'wsgi.input': fake_input,
 
121
            'bzrlib.relpath': 'foo',
 
122
        })
 
123
        iterable = wsgi_app(environ, self.start_response)
 
124
        response = self.read_response(iterable)
 
125
        self.assertEqual('200 OK', self.status)
 
126
        self.assertEqual('got bytes: fake request', response)
 
127
 
 
128
    def test_relpath_setter(self):
 
129
        # wsgi.RelpathSetter is WSGI "middleware" to set the 'bzrlib.relpath'
 
130
        # variable.
 
131
        calls = []
 
132
        def fake_app(environ, start_response):
 
133
            calls.append(environ['bzrlib.relpath'])
 
134
        wrapped_app = wsgi.RelpathSetter(
 
135
            fake_app, prefix='/abc/', path_var='FOO')
 
136
        wrapped_app({'FOO': '/abc/xyz/.bzr/smart'}, None)
 
137
        self.assertEqual(['xyz'], calls)
 
138
       
 
139
    def test_relpath_setter_bad_path(self):
 
140
        # wsgi.RelpathSetter will reject paths with that don't match the prefix
 
141
        # or suffix with a 404.  This is probably a sign of misconfiguration; a
 
142
        # server shouldn't ever be invoking our WSGI application with bad paths.
 
143
        def fake_app(environ, start_response):
 
144
            self.fail('The app should never be called when the path is wrong')
 
145
        wrapped_app = wsgi.RelpathSetter(
 
146
            fake_app, prefix='/abc/', path_var='FOO')
 
147
        iterable = wrapped_app(
 
148
            {'FOO': 'AAA/abc/xyz/.bzr/smart'}, self.start_response)
 
149
        self.read_response(iterable)
 
150
        self.assertTrue(self.status.startswith('404'))
 
151
        
 
152
    def test_make_app(self):
 
153
        # The make_app helper constructs a SmartWSGIApp wrapped in a
 
154
        # RelpathSetter.
 
155
        app = wsgi.make_app(
 
156
            root='a root',
 
157
            prefix='a prefix',
 
158
            path_var='a path_var')
 
159
        self.assertIsInstance(app, wsgi.RelpathSetter)
 
160
        self.assertIsInstance(app.app, wsgi.SmartWSGIApp)
 
161
        self.assertEndsWith(app.app.backing_transport.base, 'a%20root/')
 
162
        self.assertEqual(app.prefix, 'a prefix')
 
163
        self.assertEqual(app.path_var, 'a path_var')
 
164
 
 
165
 
 
166
class FakeRequest(object):
 
167
    
 
168
    def __init__(self, transport, write_func):
 
169
        self.transport = transport
 
170
        self.write_func = write_func
 
171
        self.accepted_bytes = ''
 
172
 
 
173
    def accept_bytes(self, bytes):
 
174
        self.accepted_bytes = bytes
 
175
        self.write_func('got bytes: ' + bytes)
 
176
 
 
177
    def next_read_size(self):
 
178
        return 0
 
179
 
 
180
 
 
181
class FakeTransport(object):
 
182
 
 
183
    def __init__(self):
 
184
        self.calls = []
 
185
 
 
186
    def clone(self, relpath):
 
187
        self.calls.append(('clone', relpath))
 
188
        return self
 
189