/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: 2020-03-22 01:35:14 UTC
  • mfrom: (7490.7.6 work)
  • mto: This revision was merged to the branch mainline in revision 7499.
  • Revision ID: jelmer@jelmer.uk-20200322013514-7vw1ntwho04rcuj3
merge lp:brz/3.1.

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
 
19
21
from .. import tests
20
 
from ..sixish import (
21
 
    BytesIO,
22
 
    )
23
22
from ..bzr.smart import medium, protocol
24
23
from ..transport.http import wsgi
25
24
from ..transport import chroot, memory
56
55
        return environ
57
56
 
58
57
    def read_response(self, iterable):
59
 
        response = ''
 
58
        response = b''
60
59
        for string in iterable:
61
60
            response += string
62
61
        return response
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', 'some bytes')
 
119
        transport.put_bytes('foo', b'some bytes')
121
120
        wsgi_app = wsgi.SmartWSGIApp(transport)
122
121
        wsgi_app.make_request = self._fake_make_request
123
122
        fake_input = BytesIO(b'fake request')
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('got bytes: fake request', response)
 
132
        self.assertEqual(b'got bytes: fake request', response)
134
133
 
135
134
    def test_relpath_setter(self):
136
135
        # wsgi.RelpathSetter is WSGI "middleware" to set the 'breezy.relpath'
137
136
        # variable.
138
137
        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
 
192
193
        def make_request(transport, write_func, bytes, root_client_path):
193
194
            request = IncompleteRequest(transport, write_func)
194
195
            request.accept_bytes(bytes)
206
207
        iterable = wsgi_app(environ, self.start_response)
207
208
        response = self.read_response(iterable)
208
209
        self.assertEqual('200 OK', self.status)
209
 
        self.assertEqual('error\x01incomplete request\n', response)
 
210
        self.assertEqual(b'error\x01incomplete request\n', response)
210
211
 
211
212
    def test_protocol_version_detection_one(self):
212
213
        # SmartWSGIApp detects requests that don't start with
224
225
        response = self.read_response(iterable)
225
226
        self.assertEqual('200 OK', self.status)
226
227
        # Expect a version 1-encoded response.
227
 
        self.assertEqual('ok\x012\n', response)
 
228
        self.assertEqual(b'ok\x012\n', response)
228
229
 
229
230
    def test_protocol_version_detection_two(self):
230
231
        # SmartWSGIApp detects requests that start with REQUEST_VERSION_TWO
231
232
        # as version two.
232
233
        transport = memory.MemoryTransport()
233
234
        wsgi_app = wsgi.SmartWSGIApp(transport)
234
 
        fake_input = BytesIO(protocol.REQUEST_VERSION_TWO + 'hello\n')
 
235
        fake_input = BytesIO(protocol.REQUEST_VERSION_TWO + b'hello\n')
235
236
        environ = self.build_environ({
236
237
            'REQUEST_METHOD': 'POST',
237
238
            'CONTENT_LENGTH': len(fake_input.getvalue()),
243
244
        self.assertEqual('200 OK', self.status)
244
245
        # Expect a version 2-encoded response.
245
246
        self.assertEqual(
246
 
            protocol.RESPONSE_VERSION_TWO + 'success\nok\x012\n', response)
 
247
            protocol.RESPONSE_VERSION_TWO + b'success\nok\x012\n', response)
247
248
 
248
249
 
249
250
class TestWSGIJail(tests.TestCaseWithMemoryTransport, WSGITestMixin):
276
277
        wsgi_app = wsgi.SmartWSGIApp(self.get_transport())
277
278
        # send a request to /repo/branch that will have to access /repo.
278
279
        environ = self.make_hpss_wsgi_request(
279
 
            '/repo/branch', 'BzrDir.open_branchV2', '.')
 
280
            '/repo/branch', b'BzrDir.open_branchV2', b'.')
280
281
        iterable = wsgi_app(environ, self.start_response)
281
282
        response_bytes = self.read_response(iterable)
282
283
        self.assertEqual('200 OK', self.status)
287
288
            message_handler, expect_version_marker=True)
288
289
        decoder.accept_bytes(response_bytes)
289
290
        self.assertTrue(
290
 
            ('structure', ('branch', branch._format.network_name()))
 
291
            ('structure', (b'branch', branch._format.network_name()))
291
292
            in message_handler.event_log)
292
293
 
293
294
 
296
297
    def __init__(self, transport, write_func):
297
298
        self.transport = transport
298
299
        self.write_func = write_func
299
 
        self.accepted_bytes = ''
 
300
        self.accepted_bytes = b''
300
301
 
301
302
    def accept_bytes(self, bytes):
302
303
        self.accepted_bytes = bytes
303
 
        self.write_func('got bytes: ' + bytes)
 
304
        self.write_func(b'got bytes: ' + bytes)
304
305
 
305
306
    def next_read_size(self):
306
307
        return 0
326
327
    def next_read_size(self):
327
328
        # this request always asks for more
328
329
        return 1
329