/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 breezy/tests/test_wsgi.py

  • Committer: Jelmer Vernooij
  • Date: 2017-06-08 23:30:31 UTC
  • mto: This revision was merged to the branch mainline in revision 6690.
  • Revision ID: jelmer@jelmer.uk-20170608233031-3qavls2o7a1pqllj
Update imports.

Show diffs side-by-side

added added

removed removed

Lines of Context:
breezy/tests/test_wsgi.py
16
16
 
17
17
"""Tests for WSGI application"""
18
18
 
19
 
from io import BytesIO
20
 
 
21
19
from .. import tests
22
 
from ..bzr.smart import medium, protocol
 
20
from ..sixish import (
 
21
    BytesIO,
 
22
    )
 
23
from ..smart import medium, protocol
23
24
from ..transport.http import wsgi
24
25
from ..transport import chroot, memory
25
26
 
42
43
            'SERVER_PROTOCOL': 'HTTP/1.0',
43
44
 
44
45
            # Required WSGI variables
45
 
            'wsgi.version': (1, 0),
 
46
            'wsgi.version': (1,0),
46
47
            'wsgi.url_scheme': 'http',
47
48
            'wsgi.input': BytesIO(b''),
48
49
            'wsgi.errors': BytesIO(),
55
56
        return environ
56
57
 
57
58
    def read_response(self, iterable):
58
 
        response = b''
 
59
        response = ''
59
60
        for string in iterable:
60
61
            response += string
61
62
        return response
109
110
        })
110
111
        iterable = wsgi_app(environ, self.start_response)
111
112
        response = self.read_response(iterable)
112
 
        self.assertEqual([('clone', 'foo/bar/')], transport.calls)
 
113
        self.assertEqual([('clone', 'foo/bar/')] , transport.calls)
113
114
 
114
115
    def test_smart_wsgi_app_request_and_response(self):
115
116
        # SmartWSGIApp reads the smart request from the 'wsgi.input' file-like
116
117
        # object in the environ dict, and returns the response via the iterable
117
118
        # returned to the WSGI handler.
118
119
        transport = memory.MemoryTransport()
119
 
        transport.put_bytes('foo', b'some bytes')
 
120
        transport.put_bytes('foo', 'some bytes')
120
121
        wsgi_app = wsgi.SmartWSGIApp(transport)
121
122
        wsgi_app.make_request = self._fake_make_request
122
123
        fake_input = BytesIO(b'fake request')
129
130
        iterable = wsgi_app(environ, self.start_response)
130
131
        response = self.read_response(iterable)
131
132
        self.assertEqual('200 OK', self.status)
132
 
        self.assertEqual(b'got bytes: fake request', response)
 
133
        self.assertEqual('got bytes: fake request', response)
133
134
 
134
135
    def test_relpath_setter(self):
135
136
        # wsgi.RelpathSetter is WSGI "middleware" to set the 'breezy.relpath'
136
137
        # variable.
137
138
        calls = []
138
 
 
139
139
        def fake_app(environ, start_response):
140
140
            calls.append(environ['breezy.relpath'])
141
141
        wrapped_app = wsgi.RelpathSetter(
189
189
    def test_incomplete_request(self):
190
190
        transport = FakeTransport()
191
191
        wsgi_app = wsgi.SmartWSGIApp(transport)
192
 
 
193
192
        def make_request(transport, write_func, bytes, root_client_path):
194
193
            request = IncompleteRequest(transport, write_func)
195
194
            request.accept_bytes(bytes)
207
206
        iterable = wsgi_app(environ, self.start_response)
208
207
        response = self.read_response(iterable)
209
208
        self.assertEqual('200 OK', self.status)
210
 
        self.assertEqual(b'error\x01incomplete request\n', response)
 
209
        self.assertEqual('error\x01incomplete request\n', response)
211
210
 
212
211
    def test_protocol_version_detection_one(self):
213
212
        # SmartWSGIApp detects requests that don't start with
225
224
        response = self.read_response(iterable)
226
225
        self.assertEqual('200 OK', self.status)
227
226
        # Expect a version 1-encoded response.
228
 
        self.assertEqual(b'ok\x012\n', response)
 
227
        self.assertEqual('ok\x012\n', response)
229
228
 
230
229
    def test_protocol_version_detection_two(self):
231
230
        # SmartWSGIApp detects requests that start with REQUEST_VERSION_TWO
232
231
        # as version two.
233
232
        transport = memory.MemoryTransport()
234
233
        wsgi_app = wsgi.SmartWSGIApp(transport)
235
 
        fake_input = BytesIO(protocol.REQUEST_VERSION_TWO + b'hello\n')
 
234
        fake_input = BytesIO(protocol.REQUEST_VERSION_TWO + 'hello\n')
236
235
        environ = self.build_environ({
237
236
            'REQUEST_METHOD': 'POST',
238
237
            'CONTENT_LENGTH': len(fake_input.getvalue()),
244
243
        self.assertEqual('200 OK', self.status)
245
244
        # Expect a version 2-encoded response.
246
245
        self.assertEqual(
247
 
            protocol.RESPONSE_VERSION_TWO + b'success\nok\x012\n', response)
 
246
            protocol.RESPONSE_VERSION_TWO + 'success\nok\x012\n', response)
248
247
 
249
248
 
250
249
class TestWSGIJail(tests.TestCaseWithMemoryTransport, WSGITestMixin):
272
271
        """
273
272
        # make a branch in a shared repo
274
273
        self.make_repository('repo', shared=True)
275
 
        branch = self.make_controldir('repo/branch').create_branch()
 
274
        branch = self.make_bzrdir('repo/branch').create_branch()
276
275
        # serve the repo via bzr+http WSGI
277
276
        wsgi_app = wsgi.SmartWSGIApp(self.get_transport())
278
277
        # send a request to /repo/branch that will have to access /repo.
279
278
        environ = self.make_hpss_wsgi_request(
280
 
            '/repo/branch', b'BzrDir.open_branchV2', b'.')
 
279
            '/repo/branch', 'BzrDir.open_branchV2', '.')
281
280
        iterable = wsgi_app(environ, self.start_response)
282
281
        response_bytes = self.read_response(iterable)
283
282
        self.assertEqual('200 OK', self.status)
284
283
        # expect a successful response, rather than a jail break error
285
 
        from breezy.bzr.tests.test_smart_transport import LoggingMessageHandler
 
284
        from breezy.tests.test_smart_transport import LoggingMessageHandler
286
285
        message_handler = LoggingMessageHandler()
287
286
        decoder = protocol.ProtocolThreeDecoder(
288
287
            message_handler, expect_version_marker=True)
289
288
        decoder.accept_bytes(response_bytes)
290
289
        self.assertTrue(
291
 
            ('structure', (b'branch', branch._format.network_name()))
 
290
            ('structure', ('branch', branch._format.network_name()))
292
291
            in message_handler.event_log)
293
292
 
294
293
 
297
296
    def __init__(self, transport, write_func):
298
297
        self.transport = transport
299
298
        self.write_func = write_func
300
 
        self.accepted_bytes = b''
 
299
        self.accepted_bytes = ''
301
300
 
302
301
    def accept_bytes(self, bytes):
303
302
        self.accepted_bytes = bytes
304
 
        self.write_func(b'got bytes: ' + bytes)
 
303
        self.write_func('got bytes: ' + bytes)
305
304
 
306
305
    def next_read_size(self):
307
306
        return 0
327
326
    def next_read_size(self):
328
327
        # this request always asks for more
329
328
        return 1
 
329