← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_wsgi.py

  • Committer: John Arbash Meinel
  • Date: 2007-04-12 21:33:07 UTC
  • mfrom: (2413.4.1 api-doc-builders)
  • mto: This revision was merged to the branch mainline in revision 2566.
  • Revision ID: john@arbash-meinel.com-20070412213307-kuh07cnzaud12wx1
[merge] api-doc-builder and remove the pydoctor build code for now.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_wsgi.py
12
12
#
13
13
# You should have received a copy of the GNU General Public License
14
14
# along with this program; if not, write to the Free Software
15
 
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
 
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
16
16
 
17
17
"""Tests for WSGI application"""
18
18
 
19
19
from cStringIO import StringIO
20
20
 
21
21
from bzrlib import tests
22
 
from bzrlib.smart import medium, message, protocol
23
22
from bzrlib.transport.http import wsgi
24
23
from bzrlib.transport import chroot, memory
25
24
 
26
25
 
27
 
class WSGITestMixin(object):
 
26
class TestWSGI(tests.TestCase):
 
27
 
 
28
    def setUp(self):
 
29
        tests.TestCase.setUp(self)
 
30
        self.status = None
 
31
        self.headers = None
28
32
 
29
33
    def build_environ(self, updates=None):
30
34
        """Builds an environ dict with all fields required by PEP 333.
31
 
 
 
35
        
32
36
        :param updates: a dict to that will be incorporated into the returned
33
37
            dict using dict.update(updates).
34
38
        """
53
57
        if updates is not None:
54
58
            environ.update(updates)
55
59
        return environ
56
 
 
 
60
        
57
61
    def read_response(self, iterable):
58
62
        response = ''
59
63
        for string in iterable:
64
68
        self.status = status
65
69
        self.headers = headers
66
70
 
67
 
 
68
 
class TestWSGI(tests.TestCase, WSGITestMixin):
69
 
 
70
 
    def setUp(self):
71
 
        tests.TestCase.setUp(self)
72
 
        self.status = None
73
 
        self.headers = None
74
 
 
75
71
    def test_construct(self):
76
72
        app = wsgi.SmartWSGIApp(FakeTransport())
77
73
        self.assertIsInstance(
85
81
        self.read_response(iterable)
86
82
        self.assertEqual('405 Method not allowed', self.status)
87
83
        self.assertTrue(('Allow', 'POST') in self.headers)
88
 
 
89
 
    def _fake_make_request(self, transport, write_func, bytes, rcp):
90
 
        request = FakeRequest(transport, write_func)
91
 
        request.accept_bytes(bytes)
92
 
        self.request = request
93
 
        return request
94
 
 
 
84
        
95
85
    def test_smart_wsgi_app_uses_given_relpath(self):
96
86
        # The SmartWSGIApp should use the "bzrlib.relpath" field from the
97
87
        # WSGI environ to clone from its backing transport to get a specific
99
89
        transport = FakeTransport()
100
90
        wsgi_app = wsgi.SmartWSGIApp(transport)
101
91
        wsgi_app.backing_transport = transport
102
 
        wsgi_app.make_request = self._fake_make_request
 
92
        def make_request(transport, write_func):
 
93
            request = FakeRequest(transport, write_func)
 
94
            self.request = request
 
95
            return request
 
96
        wsgi_app.make_request = make_request
103
97
        fake_input = StringIO('fake request')
104
98
        environ = self.build_environ({
105
99
            'REQUEST_METHOD': 'POST',
109
103
        })
110
104
        iterable = wsgi_app(environ, self.start_response)
111
105
        response = self.read_response(iterable)
112
 
        self.assertEqual([('clone', 'foo/bar/')] , transport.calls)
 
106
        self.assertEqual([('clone', 'foo/bar')] , transport.calls)
113
107
 
114
108
    def test_smart_wsgi_app_request_and_response(self):
115
109
        # SmartWSGIApp reads the smart request from the 'wsgi.input' file-like
118
112
        transport = memory.MemoryTransport()
119
113
        transport.put_bytes('foo', 'some bytes')
120
114
        wsgi_app = wsgi.SmartWSGIApp(transport)
121
 
        wsgi_app.make_request = self._fake_make_request
 
115
        def make_request(transport, write_func):
 
116
            request = FakeRequest(transport, write_func)
 
117
            self.request = request
 
118
            return request
 
119
        wsgi_app.make_request = make_request
122
120
        fake_input = StringIO('fake request')
123
121
        environ = self.build_environ({
124
122
            'REQUEST_METHOD': 'POST',
141
139
            fake_app, prefix='/abc/', path_var='FOO')
142
140
        wrapped_app({'FOO': '/abc/xyz/.bzr/smart'}, None)
143
141
        self.assertEqual(['xyz'], calls)
144
 
 
 
142
       
145
143
    def test_relpath_setter_bad_path_prefix(self):
146
144
        # wsgi.RelpathSetter will reject paths with that don't match the prefix
147
145
        # with a 404.  This is probably a sign of misconfiguration; a server
154
152
            {'FOO': 'AAA/abc/xyz/.bzr/smart'}, self.start_response)
155
153
        self.read_response(iterable)
156
154
        self.assertTrue(self.status.startswith('404'))
157
 
 
 
155
        
158
156
    def test_relpath_setter_bad_path_suffix(self):
159
157
        # Similar to test_relpath_setter_bad_path_prefix: wsgi.RelpathSetter
160
158
        # will reject paths with that don't match the suffix '.bzr/smart' with a
168
166
            {'FOO': '/abc/xyz/.bzr/AAA'}, self.start_response)
169
167
        self.read_response(iterable)
170
168
        self.assertTrue(self.status.startswith('404'))
171
 
 
 
169
        
172
170
    def test_make_app(self):
173
171
        # The make_app helper constructs a SmartWSGIApp wrapped in a
174
172
        # RelpathSetter.
182
180
        backing_transport = app.app.backing_transport
183
181
        chroot_backing_transport = backing_transport.server.backing_transport
184
182
        self.assertEndsWith(chroot_backing_transport.base, 'a%20root/')
185
 
        self.assertEqual(app.app.root_client_path, 'a prefix')
 
183
        self.assertEqual(app.prefix, 'a prefix')
186
184
        self.assertEqual(app.path_var, 'a path_var')
187
185
 
188
186
    def test_incomplete_request(self):
189
187
        transport = FakeTransport()
190
188
        wsgi_app = wsgi.SmartWSGIApp(transport)
191
 
        def make_request(transport, write_func, bytes, root_client_path):
 
189
        def make_request(transport, write_func):
192
190
            request = IncompleteRequest(transport, write_func)
193
 
            request.accept_bytes(bytes)
194
191
            self.request = request
195
192
            return request
196
193
        wsgi_app.make_request = make_request
207
204
        self.assertEqual('200 OK', self.status)
208
205
        self.assertEqual('error\x01incomplete request\n', response)
209
206
 
210
 
    def test_protocol_version_detection_one(self):
211
 
        # SmartWSGIApp detects requests that don't start with
212
 
        # REQUEST_VERSION_TWO as version one.
213
 
        transport = memory.MemoryTransport()
214
 
        wsgi_app = wsgi.SmartWSGIApp(transport)
215
 
        fake_input = StringIO('hello\n')
216
 
        environ = self.build_environ({
217
 
            'REQUEST_METHOD': 'POST',
218
 
            'CONTENT_LENGTH': len(fake_input.getvalue()),
219
 
            'wsgi.input': fake_input,
220
 
            'bzrlib.relpath': 'foo',
221
 
        })
222
 
        iterable = wsgi_app(environ, self.start_response)
223
 
        response = self.read_response(iterable)
224
 
        self.assertEqual('200 OK', self.status)
225
 
        # Expect a version 1-encoded response.
226
 
        self.assertEqual('ok\x012\n', response)
227
 
 
228
 
    def test_protocol_version_detection_two(self):
229
 
        # SmartWSGIApp detects requests that start with REQUEST_VERSION_TWO
230
 
        # as version two.
231
 
        transport = memory.MemoryTransport()
232
 
        wsgi_app = wsgi.SmartWSGIApp(transport)
233
 
        fake_input = StringIO(protocol.REQUEST_VERSION_TWO + 'hello\n')
234
 
        environ = self.build_environ({
235
 
            'REQUEST_METHOD': 'POST',
236
 
            'CONTENT_LENGTH': len(fake_input.getvalue()),
237
 
            'wsgi.input': fake_input,
238
 
            'bzrlib.relpath': 'foo',
239
 
        })
240
 
        iterable = wsgi_app(environ, self.start_response)
241
 
        response = self.read_response(iterable)
242
 
        self.assertEqual('200 OK', self.status)
243
 
        # Expect a version 2-encoded response.
244
 
        self.assertEqual(
245
 
            protocol.RESPONSE_VERSION_TWO + 'success\nok\x012\n', response)
246
 
 
247
 
 
248
 
class TestWSGIJail(tests.TestCaseWithMemoryTransport, WSGITestMixin):
249
 
 
250
 
    def make_hpss_wsgi_request(self, wsgi_relpath, *args):
251
 
        write_buf = StringIO()
252
 
        request_medium = medium.SmartSimplePipesClientMedium(
253
 
            None, write_buf, 'fake:' + wsgi_relpath)
254
 
        request_encoder = protocol.ProtocolThreeRequester(
255
 
            request_medium.get_request())
256
 
        request_encoder.call(*args)
257
 
        write_buf.seek(0)
258
 
        environ = self.build_environ({
259
 
            'REQUEST_METHOD': 'POST',
260
 
            'CONTENT_LENGTH': len(write_buf.getvalue()),
261
 
            'wsgi.input': write_buf,
262
 
            'bzrlib.relpath': wsgi_relpath,
263
 
        })
264
 
        return environ
265
 
 
266
 
    def test_jail_root(self):
267
 
        """The WSGI HPSS glue allows access to the whole WSGI backing
268
 
        transport, regardless of which HTTP path the request was delivered
269
 
        to.
270
 
        """
271
 
        # make a branch in a shared repo
272
 
        self.make_repository('repo', shared=True)
273
 
        branch = self.make_bzrdir('repo/branch').create_branch()
274
 
        # serve the repo via bzr+http WSGI
275
 
        wsgi_app = wsgi.SmartWSGIApp(self.get_transport())
276
 
        # send a request to /repo/branch that will have to access /repo.
277
 
        environ = self.make_hpss_wsgi_request(
278
 
            '/repo/branch', 'BzrDir.open_branchV2', '.')
279
 
        iterable = wsgi_app(environ, self.start_response)
280
 
        response_bytes = self.read_response(iterable)
281
 
        self.assertEqual('200 OK', self.status)
282
 
        # expect a successful response, rather than a jail break error
283
 
        from bzrlib.tests.test_smart_transport import LoggingMessageHandler
284
 
        message_handler = LoggingMessageHandler()
285
 
        decoder = protocol.ProtocolThreeDecoder(
286
 
            message_handler, expect_version_marker=True)
287
 
        decoder.accept_bytes(response_bytes)
288
 
        self.assertTrue(
289
 
            ('structure', ('branch', branch._format.network_name()))
290
 
            in message_handler.event_log)
291
 
 
292
207
 
293
208
class FakeRequest(object):
294
 
 
 
209
    
295
210
    def __init__(self, transport, write_func):
296
211
        self.transport = transport
297
212
        self.write_func = write_func