← 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: Aaron Bentley
  • Date: 2007-02-06 14:52:16 UTC
  • mfrom: (2266 +trunk)
  • mto: This revision was merged to the branch mainline in revision 2268.
  • Revision ID: abentley@panoramicfeedback.com-20070206145216-fcpi8o3ufvuzwbp9
Merge bzr.dev

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_wsgi.py
19
19
from cStringIO import StringIO
20
20
 
21
21
from bzrlib import tests
22
 
from bzrlib.smart import protocol
23
22
from bzrlib.transport.http import wsgi
24
23
from bzrlib.transport import chroot, memory
25
24
 
72
71
    def test_construct(self):
73
72
        app = wsgi.SmartWSGIApp(FakeTransport())
74
73
        self.assertIsInstance(
75
 
            app.backing_transport, chroot.ChrootTransport)
 
74
            app.backing_transport, chroot.ChrootTransportDecorator)
76
75
 
77
76
    def test_http_get_rejected(self):
78
77
        # GET requests are rejected.
83
82
        self.assertEqual('405 Method not allowed', self.status)
84
83
        self.assertTrue(('Allow', 'POST') in self.headers)
85
84
        
86
 
    def _fake_make_request(self, transport, write_func, bytes, rcp):
87
 
        request = FakeRequest(transport, write_func)
88
 
        request.accept_bytes(bytes)
89
 
        self.request = request
90
 
        return request
91
 
    
92
85
    def test_smart_wsgi_app_uses_given_relpath(self):
93
86
        # The SmartWSGIApp should use the "bzrlib.relpath" field from the
94
 
        # WSGI environ to clone from its backing transport to get a specific
95
 
        # transport for this request.
 
87
        # WSGI environ to construct the transport for this request, by cloning
 
88
        # its base transport with the given relpath.
96
89
        transport = FakeTransport()
97
90
        wsgi_app = wsgi.SmartWSGIApp(transport)
98
 
        wsgi_app.backing_transport = transport
99
 
        wsgi_app.make_request = self._fake_make_request
 
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
100
96
        fake_input = StringIO('fake request')
101
97
        environ = self.build_environ({
102
98
            'REQUEST_METHOD': 'POST',
106
102
        })
107
103
        iterable = wsgi_app(environ, self.start_response)
108
104
        response = self.read_response(iterable)
109
 
        self.assertEqual([('clone', 'foo/bar/')] , transport.calls)
 
105
        self.assertEqual([('clone', 'foo/bar')] , transport.calls)
110
106
 
111
107
    def test_smart_wsgi_app_request_and_response(self):
112
108
        # SmartWSGIApp reads the smart request from the 'wsgi.input' file-like
115
111
        transport = memory.MemoryTransport()
116
112
        transport.put_bytes('foo', 'some bytes')
117
113
        wsgi_app = wsgi.SmartWSGIApp(transport)
118
 
        wsgi_app.make_request = self._fake_make_request
 
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
119
        fake_input = StringIO('fake request')
120
120
        environ = self.build_environ({
121
121
            'REQUEST_METHOD': 'POST',
175
175
            path_var='a path_var')
176
176
        self.assertIsInstance(app, wsgi.RelpathSetter)
177
177
        self.assertIsInstance(app.app, wsgi.SmartWSGIApp)
178
 
        self.assertStartsWith(app.app.backing_transport.base, 'chroot-')
179
 
        backing_transport = app.app.backing_transport
180
 
        chroot_backing_transport = backing_transport.server.backing_transport
181
 
        self.assertEndsWith(chroot_backing_transport.base, 'a%20root/')
182
 
        self.assertEqual(app.app.root_client_path, 'a prefix')
 
178
        self.assertEndsWith(app.app.backing_transport.base, 'a%20root/')
 
179
        self.assertEqual(app.prefix, 'a prefix')
183
180
        self.assertEqual(app.path_var, 'a path_var')
184
181
 
185
182
    def test_incomplete_request(self):
186
183
        transport = FakeTransport()
187
184
        wsgi_app = wsgi.SmartWSGIApp(transport)
188
 
        def make_request(transport, write_func, bytes, root_client_path):
 
185
        def make_request(transport, write_func):
189
186
            request = IncompleteRequest(transport, write_func)
190
 
            request.accept_bytes(bytes)
191
187
            self.request = request
192
188
            return request
193
189
        wsgi_app.make_request = make_request
204
200
        self.assertEqual('200 OK', self.status)
205
201
        self.assertEqual('error\x01incomplete request\n', response)
206
202
 
207
 
    def test_protocol_version_detection_one(self):
208
 
        # SmartWSGIApp detects requests that don't start with
209
 
        # REQUEST_VERSION_TWO as version one.
 
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.
210
206
        transport = memory.MemoryTransport()
211
 
        wsgi_app = wsgi.SmartWSGIApp(transport)
212
 
        fake_input = StringIO('hello\n')
213
 
        environ = self.build_environ({
214
 
            'REQUEST_METHOD': 'POST',
215
 
            'CONTENT_LENGTH': len(fake_input.getvalue()),
216
 
            'wsgi.input': fake_input,
217
 
            'bzrlib.relpath': 'foo',
218
 
        })
219
 
        iterable = wsgi_app(environ, self.start_response)
220
 
        response = self.read_response(iterable)
221
 
        self.assertEqual('200 OK', self.status)
222
 
        # Expect a version 1-encoded response.
223
 
        self.assertEqual('ok\x012\n', response)
 
207
        transport.mkdir('foo')
 
208
        transport.put_bytes('foo/bar', 'this is foo/bar')
 
209
        wsgi_app = wsgi.SmartWSGIApp(transport.clone('foo'))
224
210
 
225
 
    def test_protocol_version_detection_two(self):
226
 
        # SmartWSGIApp detects requests that start with REQUEST_VERSION_TWO
227
 
        # as version two.
228
 
        transport = memory.MemoryTransport()
229
 
        wsgi_app = wsgi.SmartWSGIApp(transport)
230
 
        fake_input = StringIO(protocol.REQUEST_VERSION_TWO + 'hello\n')
 
211
        smart_request = StringIO('mkdir\x01/bad file\x01\n0\ndone\n')
231
212
        environ = self.build_environ({
232
213
            'REQUEST_METHOD': 'POST',
233
 
            'CONTENT_LENGTH': len(fake_input.getvalue()),
234
 
            'wsgi.input': fake_input,
235
 
            'bzrlib.relpath': 'foo',
 
214
            'CONTENT_LENGTH': len(smart_request.getvalue()),
 
215
            'wsgi.input': smart_request,
 
216
            'bzrlib.relpath': '.',
236
217
        })
237
218
        iterable = wsgi_app(environ, self.start_response)
238
219
        response = self.read_response(iterable)
239
220
        self.assertEqual('200 OK', self.status)
240
 
        # Expect a version 2-encoded response.
241
221
        self.assertEqual(
242
 
            protocol.RESPONSE_VERSION_TWO + 'success\nok\x012\n', response)
 
222
            "error\x01Path '/bad file' is not a child of "
 
223
            "path 'memory:///foo/'\n",
 
224
            response)
243
225
 
244
226
 
245
227
class FakeRequest(object):