← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_wsgi.py

Improvement thanks to John's review.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_wsgi.py
22
22
from bzrlib.transport.http import wsgi
23
23
from bzrlib.transport import memory
24
24
 
 
25
 
25
26
class TestWSGI(tests.TestCase):
26
27
 
27
28
    def setUp(self):
29
30
        self.status = None
30
31
        self.headers = None
31
32
 
32
 
    def build_environ(self, **kw):
 
33
    def build_environ(self, updates=None):
33
34
        """Builds an environ dict with all fields required by PEP 333.
34
35
        
35
 
        The resulting environ dict will be updated with an **kw that are passed.
 
36
        :param updates: a dict to that will be incorporated into the returned
 
37
            dict using dict.update(updates).
36
38
        """
37
39
        environ = {
38
40
            # Required CGI variables
52
54
            'wsgi.multiprocess': False,
53
55
            'wsgi.run_once': True,
54
56
        }
55
 
        environ.update(kw)
 
57
        if updates is not None:
 
58
            environ.update(updates)
56
59
        return environ
57
60
        
58
61
    def read_response(self, iterable):
71
74
    def test_http_get_rejected(self):
72
75
        # GET requests are rejected.
73
76
        app = wsgi.SmartWSGIApp(None)
74
 
        environ = self.build_environ(REQUEST_METHOD='GET')
 
77
        environ = self.build_environ({'REQUEST_METHOD': 'GET'})
75
78
        iterable = app(environ, self.start_response)
76
79
        self.read_response(iterable)
77
80
        self.assertEqual('405 Method not allowed', self.status)
89
92
            return request
90
93
        wsgi_app.make_request = make_request
91
94
        fake_input = StringIO('fake request')
92
 
        environ = self.build_environ()
93
 
        environ.update({
 
95
        environ = self.build_environ({
94
96
            'REQUEST_METHOD': 'POST',
95
97
            'CONTENT_LENGTH': len(fake_input.getvalue()),
96
98
            'wsgi.input': fake_input,
113
115
            return request
114
116
        wsgi_app.make_request = make_request
115
117
        fake_input = StringIO('fake request')
116
 
        environ = self.build_environ()
117
 
        environ.update({
 
118
        environ = self.build_environ({
118
119
            'REQUEST_METHOD': 'POST',
119
120
            'CONTENT_LENGTH': len(fake_input.getvalue()),
120
121
            'wsgi.input': fake_input,
136
137
        wrapped_app({'FOO': '/abc/xyz/.bzr/smart'}, None)
137
138
        self.assertEqual(['xyz'], calls)
138
139
       
139
 
    def test_relpath_setter_bad_path(self):
 
140
    def test_relpath_setter_bad_path_prefix(self):
140
141
        # 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.
 
142
        # with a 404.  This is probably a sign of misconfiguration; a server
 
143
        # shouldn't ever be invoking our WSGI application with bad paths.
143
144
        def fake_app(environ, start_response):
144
145
            self.fail('The app should never be called when the path is wrong')
145
146
        wrapped_app = wsgi.RelpathSetter(
149
150
        self.read_response(iterable)
150
151
        self.assertTrue(self.status.startswith('404'))
151
152
        
 
153
    def test_relpath_setter_bad_path_suffix(self):
 
154
        # Similar to test_relpath_setter_bad_path_prefix: wsgi.RelpathSetter
 
155
        # will reject paths with that don't match the suffix '.bzr/smart' with a
 
156
        # 404 as well.  Again, this shouldn't be seen by our WSGI application if
 
157
        # the server is configured correctly.
 
158
        def fake_app(environ, start_response):
 
159
            self.fail('The app should never be called when the path is wrong')
 
160
        wrapped_app = wsgi.RelpathSetter(
 
161
            fake_app, prefix='/abc/', path_var='FOO')
 
162
        iterable = wrapped_app(
 
163
            {'FOO': '/abc/xyz/.bzr/AAA'}, self.start_response)
 
164
        self.read_response(iterable)
 
165
        self.assertTrue(self.status.startswith('404'))
 
166
        
152
167
    def test_make_app(self):
153
168
        # The make_app helper constructs a SmartWSGIApp wrapped in a
154
169
        # RelpathSetter.
162
177
        self.assertEqual(app.prefix, 'a prefix')
163
178
        self.assertEqual(app.path_var, 'a path_var')
164
179
 
 
180
    def test_incomplete_request(self):
 
181
        transport = FakeTransport()
 
182
        wsgi_app = wsgi.SmartWSGIApp(transport)
 
183
        def make_request(transport, write_func):
 
184
            request = IncompleteRequest(transport, write_func)
 
185
            self.request = request
 
186
            return request
 
187
        wsgi_app.make_request = make_request
 
188
 
 
189
        fake_input = StringIO('incomplete request')
 
190
        environ = self.build_environ({
 
191
            'REQUEST_METHOD': 'POST',
 
192
            'CONTENT_LENGTH': len(fake_input.getvalue()),
 
193
            'wsgi.input': fake_input,
 
194
            'bzrlib.relpath': 'foo/bar',
 
195
        })
 
196
        iterable = wsgi_app(environ, self.start_response)
 
197
        response = self.read_response(iterable)
 
198
        self.assertEqual('200 OK', self.status)
 
199
        self.assertEqual('error\x01incomplete request\n', response)
 
200
 
165
201
 
166
202
class FakeRequest(object):
167
203
    
187
223
        self.calls.append(('clone', relpath))
188
224
        return self
189
225
 
 
226
 
 
227
class IncompleteRequest(FakeRequest):
 
228
    """A request-like object that always expects to read more bytes."""
 
229
 
 
230
    def next_read_size(self):
 
231
        # this request always asks for more
 
232
        return 1
 
233