← 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
  • Author(s): Mark Hammond
  • Date: 2008-09-09 17:02:21 UTC
  • mto: This revision was merged to the branch mainline in revision 3697.
  • Revision ID: john@arbash-meinel.com-20080909170221-svim3jw2mrz0amp3
An updated transparent icon for bzr.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_wsgi.py
1
 
# Copyright (C) 2006-2009, 2011 Canonical Ltd
 
1
# Copyright (C) 2006 Canonical Ltd
2
2
#
3
3
# This program is free software; you can redistribute it and/or modify
4
4
# it under the terms of the GNU General Public License as published by
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, protocol
 
22
from bzrlib.smart import protocol
23
23
from bzrlib.transport.http import wsgi
24
24
from bzrlib.transport import chroot, memory
25
25
 
26
26
 
27
 
class WSGITestMixin(object):
 
27
class TestWSGI(tests.TestCase):
 
28
 
 
29
    def setUp(self):
 
30
        tests.TestCase.setUp(self)
 
31
        self.status = None
 
32
        self.headers = None
28
33
 
29
34
    def build_environ(self, updates=None):
30
35
        """Builds an environ dict with all fields required by PEP 333.
31
 
 
 
36
        
32
37
        :param updates: a dict to that will be incorporated into the returned
33
38
            dict using dict.update(updates).
34
39
        """
53
58
        if updates is not None:
54
59
            environ.update(updates)
55
60
        return environ
56
 
 
 
61
        
57
62
    def read_response(self, iterable):
58
63
        response = ''
59
64
        for string in iterable:
64
69
        self.status = status
65
70
        self.headers = headers
66
71
 
67
 
 
68
 
class TestWSGI(tests.TestCaseInTempDir, WSGITestMixin):
69
 
 
70
 
    def setUp(self):
71
 
        super(TestWSGI, self).setUp()
72
 
        self.status = None
73
 
        self.headers = None
74
 
 
75
72
    def test_construct(self):
76
73
        app = wsgi.SmartWSGIApp(FakeTransport())
77
74
        self.assertIsInstance(
85
82
        self.read_response(iterable)
86
83
        self.assertEqual('405 Method not allowed', self.status)
87
84
        self.assertTrue(('Allow', 'POST') in self.headers)
88
 
 
 
85
        
89
86
    def _fake_make_request(self, transport, write_func, bytes, rcp):
90
87
        request = FakeRequest(transport, write_func)
91
88
        request.accept_bytes(bytes)
92
89
        self.request = request
93
90
        return request
94
 
 
 
91
    
95
92
    def test_smart_wsgi_app_uses_given_relpath(self):
96
93
        # The SmartWSGIApp should use the "bzrlib.relpath" field from the
97
94
        # WSGI environ to clone from its backing transport to get a specific
141
138
            fake_app, prefix='/abc/', path_var='FOO')
142
139
        wrapped_app({'FOO': '/abc/xyz/.bzr/smart'}, None)
143
140
        self.assertEqual(['xyz'], calls)
144
 
 
 
141
       
145
142
    def test_relpath_setter_bad_path_prefix(self):
146
143
        # wsgi.RelpathSetter will reject paths with that don't match the prefix
147
144
        # with a 404.  This is probably a sign of misconfiguration; a server
154
151
            {'FOO': 'AAA/abc/xyz/.bzr/smart'}, self.start_response)
155
152
        self.read_response(iterable)
156
153
        self.assertTrue(self.status.startswith('404'))
157
 
 
 
154
        
158
155
    def test_relpath_setter_bad_path_suffix(self):
159
156
        # Similar to test_relpath_setter_bad_path_prefix: wsgi.RelpathSetter
160
157
        # will reject paths with that don't match the suffix '.bzr/smart' with a
168
165
            {'FOO': '/abc/xyz/.bzr/AAA'}, self.start_response)
169
166
        self.read_response(iterable)
170
167
        self.assertTrue(self.status.startswith('404'))
171
 
 
 
168
        
172
169
    def test_make_app(self):
173
170
        # The make_app helper constructs a SmartWSGIApp wrapped in a
174
171
        # RelpathSetter.
245
242
            protocol.RESPONSE_VERSION_TWO + 'success\nok\x012\n', response)
246
243
 
247
244
 
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
 
 
293
245
class FakeRequest(object):
294
 
 
 
246
    
295
247
    def __init__(self, transport, write_func):
296
248
        self.transport = transport
297
249
        self.write_func = write_func