/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_wsgi.py

  • Committer: Robert Collins
  • Date: 2010-05-06 23:41:35 UTC
  • mto: This revision was merged to the branch mainline in revision 5223.
  • Revision ID: robertc@robertcollins.net-20100506234135-yivbzczw1sejxnxc
Lock methods on ``Tree``, ``Branch`` and ``Repository`` are now
expected to return an object which can be used to unlock them. This reduces
duplicate code when using cleanups. The previous 'tokens's returned by
``Branch.lock_write`` and ``Repository.lock_write`` are now attributes
on the result of the lock_write. ``repository.RepositoryWriteLockResult``
and ``branch.BranchWriteLockResult`` document this. (Robert Collins)

``log._get_info_for_log_files`` now takes an add_cleanup callable.
(Robert Collins)

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
16
16
 
17
17
"""Tests for WSGI application"""
18
18
 
19
 
from .. import tests
20
 
from ..sixish import (
21
 
    BytesIO,
22
 
    )
23
 
from ..bzr.smart import medium, protocol
24
 
from ..transport.http import wsgi
25
 
from ..transport import chroot, memory
 
19
from cStringIO import StringIO
 
20
 
 
21
from bzrlib import tests
 
22
from bzrlib.smart import medium, message, protocol
 
23
from bzrlib.transport.http import wsgi
 
24
from bzrlib.transport import chroot, memory
26
25
 
27
26
 
28
27
class WSGITestMixin(object):
43
42
            'SERVER_PROTOCOL': 'HTTP/1.0',
44
43
 
45
44
            # Required WSGI variables
46
 
            'wsgi.version': (1, 0),
 
45
            'wsgi.version': (1,0),
47
46
            'wsgi.url_scheme': 'http',
48
 
            'wsgi.input': BytesIO(b''),
49
 
            'wsgi.errors': BytesIO(),
 
47
            'wsgi.input': StringIO(''),
 
48
            'wsgi.errors': StringIO(),
50
49
            'wsgi.multithread': False,
51
50
            'wsgi.multiprocess': False,
52
51
            'wsgi.run_once': True,
56
55
        return environ
57
56
 
58
57
    def read_response(self, iterable):
59
 
        response = b''
 
58
        response = ''
60
59
        for string in iterable:
61
60
            response += string
62
61
        return response
66
65
        self.headers = headers
67
66
 
68
67
 
69
 
class TestWSGI(tests.TestCaseInTempDir, WSGITestMixin):
 
68
class TestWSGI(tests.TestCase, WSGITestMixin):
70
69
 
71
70
    def setUp(self):
72
 
        super(TestWSGI, self).setUp()
 
71
        tests.TestCase.setUp(self)
73
72
        self.status = None
74
73
        self.headers = None
75
74
 
94
93
        return request
95
94
 
96
95
    def test_smart_wsgi_app_uses_given_relpath(self):
97
 
        # The SmartWSGIApp should use the "breezy.relpath" field from the
 
96
        # The SmartWSGIApp should use the "bzrlib.relpath" field from the
98
97
        # WSGI environ to clone from its backing transport to get a specific
99
98
        # transport for this request.
100
99
        transport = FakeTransport()
101
100
        wsgi_app = wsgi.SmartWSGIApp(transport)
102
101
        wsgi_app.backing_transport = transport
103
102
        wsgi_app.make_request = self._fake_make_request
104
 
        fake_input = BytesIO(b'fake request')
 
103
        fake_input = StringIO('fake request')
105
104
        environ = self.build_environ({
106
105
            'REQUEST_METHOD': 'POST',
107
106
            'CONTENT_LENGTH': len(fake_input.getvalue()),
108
107
            'wsgi.input': fake_input,
109
 
            'breezy.relpath': 'foo/bar',
 
108
            'bzrlib.relpath': 'foo/bar',
110
109
        })
111
110
        iterable = wsgi_app(environ, self.start_response)
112
111
        response = self.read_response(iterable)
113
 
        self.assertEqual([('clone', 'foo/bar/')], transport.calls)
 
112
        self.assertEqual([('clone', 'foo/bar/')] , transport.calls)
114
113
 
115
114
    def test_smart_wsgi_app_request_and_response(self):
116
115
        # SmartWSGIApp reads the smart request from the 'wsgi.input' file-like
117
116
        # object in the environ dict, and returns the response via the iterable
118
117
        # returned to the WSGI handler.
119
118
        transport = memory.MemoryTransport()
120
 
        transport.put_bytes('foo', b'some bytes')
 
119
        transport.put_bytes('foo', 'some bytes')
121
120
        wsgi_app = wsgi.SmartWSGIApp(transport)
122
121
        wsgi_app.make_request = self._fake_make_request
123
 
        fake_input = BytesIO(b'fake request')
 
122
        fake_input = StringIO('fake request')
124
123
        environ = self.build_environ({
125
124
            'REQUEST_METHOD': 'POST',
126
125
            'CONTENT_LENGTH': len(fake_input.getvalue()),
127
126
            'wsgi.input': fake_input,
128
 
            'breezy.relpath': 'foo',
 
127
            'bzrlib.relpath': 'foo',
129
128
        })
130
129
        iterable = wsgi_app(environ, self.start_response)
131
130
        response = self.read_response(iterable)
132
131
        self.assertEqual('200 OK', self.status)
133
 
        self.assertEqual(b'got bytes: fake request', response)
 
132
        self.assertEqual('got bytes: fake request', response)
134
133
 
135
134
    def test_relpath_setter(self):
136
 
        # wsgi.RelpathSetter is WSGI "middleware" to set the 'breezy.relpath'
 
135
        # wsgi.RelpathSetter is WSGI "middleware" to set the 'bzrlib.relpath'
137
136
        # variable.
138
137
        calls = []
139
 
 
140
138
        def fake_app(environ, start_response):
141
 
            calls.append(environ['breezy.relpath'])
 
139
            calls.append(environ['bzrlib.relpath'])
142
140
        wrapped_app = wsgi.RelpathSetter(
143
141
            fake_app, prefix='/abc/', path_var='FOO')
144
142
        wrapped_app({'FOO': '/abc/xyz/.bzr/smart'}, None)
190
188
    def test_incomplete_request(self):
191
189
        transport = FakeTransport()
192
190
        wsgi_app = wsgi.SmartWSGIApp(transport)
193
 
 
194
191
        def make_request(transport, write_func, bytes, root_client_path):
195
192
            request = IncompleteRequest(transport, write_func)
196
193
            request.accept_bytes(bytes)
198
195
            return request
199
196
        wsgi_app.make_request = make_request
200
197
 
201
 
        fake_input = BytesIO(b'incomplete request')
 
198
        fake_input = StringIO('incomplete request')
202
199
        environ = self.build_environ({
203
200
            'REQUEST_METHOD': 'POST',
204
201
            'CONTENT_LENGTH': len(fake_input.getvalue()),
205
202
            'wsgi.input': fake_input,
206
 
            'breezy.relpath': 'foo/bar',
 
203
            'bzrlib.relpath': 'foo/bar',
207
204
        })
208
205
        iterable = wsgi_app(environ, self.start_response)
209
206
        response = self.read_response(iterable)
210
207
        self.assertEqual('200 OK', self.status)
211
 
        self.assertEqual(b'error\x01incomplete request\n', response)
 
208
        self.assertEqual('error\x01incomplete request\n', response)
212
209
 
213
210
    def test_protocol_version_detection_one(self):
214
211
        # SmartWSGIApp detects requests that don't start with
215
212
        # REQUEST_VERSION_TWO as version one.
216
213
        transport = memory.MemoryTransport()
217
214
        wsgi_app = wsgi.SmartWSGIApp(transport)
218
 
        fake_input = BytesIO(b'hello\n')
 
215
        fake_input = StringIO('hello\n')
219
216
        environ = self.build_environ({
220
217
            'REQUEST_METHOD': 'POST',
221
218
            'CONTENT_LENGTH': len(fake_input.getvalue()),
222
219
            'wsgi.input': fake_input,
223
 
            'breezy.relpath': 'foo',
 
220
            'bzrlib.relpath': 'foo',
224
221
        })
225
222
        iterable = wsgi_app(environ, self.start_response)
226
223
        response = self.read_response(iterable)
227
224
        self.assertEqual('200 OK', self.status)
228
225
        # Expect a version 1-encoded response.
229
 
        self.assertEqual(b'ok\x012\n', response)
 
226
        self.assertEqual('ok\x012\n', response)
230
227
 
231
228
    def test_protocol_version_detection_two(self):
232
229
        # SmartWSGIApp detects requests that start with REQUEST_VERSION_TWO
233
230
        # as version two.
234
231
        transport = memory.MemoryTransport()
235
232
        wsgi_app = wsgi.SmartWSGIApp(transport)
236
 
        fake_input = BytesIO(protocol.REQUEST_VERSION_TWO + b'hello\n')
 
233
        fake_input = StringIO(protocol.REQUEST_VERSION_TWO + 'hello\n')
237
234
        environ = self.build_environ({
238
235
            'REQUEST_METHOD': 'POST',
239
236
            'CONTENT_LENGTH': len(fake_input.getvalue()),
240
237
            'wsgi.input': fake_input,
241
 
            'breezy.relpath': 'foo',
 
238
            'bzrlib.relpath': 'foo',
242
239
        })
243
240
        iterable = wsgi_app(environ, self.start_response)
244
241
        response = self.read_response(iterable)
245
242
        self.assertEqual('200 OK', self.status)
246
243
        # Expect a version 2-encoded response.
247
244
        self.assertEqual(
248
 
            protocol.RESPONSE_VERSION_TWO + b'success\nok\x012\n', response)
 
245
            protocol.RESPONSE_VERSION_TWO + 'success\nok\x012\n', response)
249
246
 
250
247
 
251
248
class TestWSGIJail(tests.TestCaseWithMemoryTransport, WSGITestMixin):
252
249
 
253
250
    def make_hpss_wsgi_request(self, wsgi_relpath, *args):
254
 
        write_buf = BytesIO()
 
251
        write_buf = StringIO()
255
252
        request_medium = medium.SmartSimplePipesClientMedium(
256
253
            None, write_buf, 'fake:' + wsgi_relpath)
257
254
        request_encoder = protocol.ProtocolThreeRequester(
262
259
            'REQUEST_METHOD': 'POST',
263
260
            'CONTENT_LENGTH': len(write_buf.getvalue()),
264
261
            'wsgi.input': write_buf,
265
 
            'breezy.relpath': wsgi_relpath,
 
262
            'bzrlib.relpath': wsgi_relpath,
266
263
        })
267
264
        return environ
268
265
 
273
270
        """
274
271
        # make a branch in a shared repo
275
272
        self.make_repository('repo', shared=True)
276
 
        branch = self.make_controldir('repo/branch').create_branch()
 
273
        branch = self.make_bzrdir('repo/branch').create_branch()
277
274
        # serve the repo via bzr+http WSGI
278
275
        wsgi_app = wsgi.SmartWSGIApp(self.get_transport())
279
276
        # send a request to /repo/branch that will have to access /repo.
280
277
        environ = self.make_hpss_wsgi_request(
281
 
            '/repo/branch', b'BzrDir.open_branchV2', b'.')
 
278
            '/repo/branch', 'BzrDir.open_branchV2', '.')
282
279
        iterable = wsgi_app(environ, self.start_response)
283
280
        response_bytes = self.read_response(iterable)
284
281
        self.assertEqual('200 OK', self.status)
285
282
        # expect a successful response, rather than a jail break error
286
 
        from breezy.tests.test_smart_transport import LoggingMessageHandler
 
283
        from bzrlib.tests.test_smart_transport import LoggingMessageHandler
287
284
        message_handler = LoggingMessageHandler()
288
285
        decoder = protocol.ProtocolThreeDecoder(
289
286
            message_handler, expect_version_marker=True)
290
287
        decoder.accept_bytes(response_bytes)
291
288
        self.assertTrue(
292
 
            ('structure', (b'branch', branch._format.network_name()))
 
289
            ('structure', ('branch', branch._format.network_name()))
293
290
            in message_handler.event_log)
294
291
 
295
292
 
298
295
    def __init__(self, transport, write_func):
299
296
        self.transport = transport
300
297
        self.write_func = write_func
301
 
        self.accepted_bytes = b''
 
298
        self.accepted_bytes = ''
302
299
 
303
300
    def accept_bytes(self, bytes):
304
301
        self.accepted_bytes = bytes
305
 
        self.write_func(b'got bytes: ' + bytes)
 
302
        self.write_func('got bytes: ' + bytes)
306
303
 
307
304
    def next_read_size(self):
308
305
        return 0
328
325
    def next_read_size(self):
329
326
        # this request always asks for more
330
327
        return 1
 
328