~ubuntu-branches/ubuntu/gutsy/bzr/gutsy

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_wsgi.py

  • Committer: Bazaar Package Importer
  • Author(s): Etienne Goyer
  • Date: 2007-04-27 17:53:49 UTC
  • mfrom: (1.1.23 upstream)
  • Revision ID: james.westby@ubuntu.com-20070427175349-rvowqx994rfuikuu
Tags: 0.16~rc1-0ubuntu1
New upstream development release 

Show diffs side-by-side

added added

removed removed

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