/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: 2018-05-06 11:48:54 UTC
  • mto: This revision was merged to the branch mainline in revision 6960.
  • Revision ID: jelmer@jelmer.uk-20180506114854-h4qd9ojaqy8wxjsd
Move .mailmap to root.

Show diffs side-by-side

added added

removed removed

Lines of Context:
breezy/tests/test_wsgi.py
56
56
        return environ
57
57
 
58
58
    def read_response(self, iterable):
59
 
        response = b''
 
59
        response = ''
60
60
        for string in iterable:
61
61
            response += string
62
62
        return response
117
117
        # object in the environ dict, and returns the response via the iterable
118
118
        # returned to the WSGI handler.
119
119
        transport = memory.MemoryTransport()
120
 
        transport.put_bytes('foo', b'some bytes')
 
120
        transport.put_bytes('foo', 'some bytes')
121
121
        wsgi_app = wsgi.SmartWSGIApp(transport)
122
122
        wsgi_app.make_request = self._fake_make_request
123
123
        fake_input = BytesIO(b'fake request')
130
130
        iterable = wsgi_app(environ, self.start_response)
131
131
        response = self.read_response(iterable)
132
132
        self.assertEqual('200 OK', self.status)
133
 
        self.assertEqual(b'got bytes: fake request', response)
 
133
        self.assertEqual('got bytes: fake request', response)
134
134
 
135
135
    def test_relpath_setter(self):
136
136
        # wsgi.RelpathSetter is WSGI "middleware" to set the 'breezy.relpath'
137
137
        # variable.
138
138
        calls = []
139
 
 
140
139
        def fake_app(environ, start_response):
141
140
            calls.append(environ['breezy.relpath'])
142
141
        wrapped_app = wsgi.RelpathSetter(
190
189
    def test_incomplete_request(self):
191
190
        transport = FakeTransport()
192
191
        wsgi_app = wsgi.SmartWSGIApp(transport)
193
 
 
194
192
        def make_request(transport, write_func, bytes, root_client_path):
195
193
            request = IncompleteRequest(transport, write_func)
196
194
            request.accept_bytes(bytes)
208
206
        iterable = wsgi_app(environ, self.start_response)
209
207
        response = self.read_response(iterable)
210
208
        self.assertEqual('200 OK', self.status)
211
 
        self.assertEqual(b'error\x01incomplete request\n', response)
 
209
        self.assertEqual('error\x01incomplete request\n', response)
212
210
 
213
211
    def test_protocol_version_detection_one(self):
214
212
        # SmartWSGIApp detects requests that don't start with
226
224
        response = self.read_response(iterable)
227
225
        self.assertEqual('200 OK', self.status)
228
226
        # Expect a version 1-encoded response.
229
 
        self.assertEqual(b'ok\x012\n', response)
 
227
        self.assertEqual('ok\x012\n', response)
230
228
 
231
229
    def test_protocol_version_detection_two(self):
232
230
        # SmartWSGIApp detects requests that start with REQUEST_VERSION_TWO
233
231
        # as version two.
234
232
        transport = memory.MemoryTransport()
235
233
        wsgi_app = wsgi.SmartWSGIApp(transport)
236
 
        fake_input = BytesIO(protocol.REQUEST_VERSION_TWO + b'hello\n')
 
234
        fake_input = BytesIO(protocol.REQUEST_VERSION_TWO + 'hello\n')
237
235
        environ = self.build_environ({
238
236
            'REQUEST_METHOD': 'POST',
239
237
            'CONTENT_LENGTH': len(fake_input.getvalue()),
245
243
        self.assertEqual('200 OK', self.status)
246
244
        # Expect a version 2-encoded response.
247
245
        self.assertEqual(
248
 
            protocol.RESPONSE_VERSION_TWO + b'success\nok\x012\n', response)
 
246
            protocol.RESPONSE_VERSION_TWO + 'success\nok\x012\n', response)
249
247
 
250
248
 
251
249
class TestWSGIJail(tests.TestCaseWithMemoryTransport, WSGITestMixin):
278
276
        wsgi_app = wsgi.SmartWSGIApp(self.get_transport())
279
277
        # send a request to /repo/branch that will have to access /repo.
280
278
        environ = self.make_hpss_wsgi_request(
281
 
            '/repo/branch', b'BzrDir.open_branchV2', b'.')
 
279
            '/repo/branch', 'BzrDir.open_branchV2', '.')
282
280
        iterable = wsgi_app(environ, self.start_response)
283
281
        response_bytes = self.read_response(iterable)
284
282
        self.assertEqual('200 OK', self.status)
289
287
            message_handler, expect_version_marker=True)
290
288
        decoder.accept_bytes(response_bytes)
291
289
        self.assertTrue(
292
 
            ('structure', (b'branch', branch._format.network_name()))
 
290
            ('structure', ('branch', branch._format.network_name()))
293
291
            in message_handler.event_log)
294
292
 
295
293
 
298
296
    def __init__(self, transport, write_func):
299
297
        self.transport = transport
300
298
        self.write_func = write_func
301
 
        self.accepted_bytes = b''
 
299
        self.accepted_bytes = ''
302
300
 
303
301
    def accept_bytes(self, bytes):
304
302
        self.accepted_bytes = bytes
305
 
        self.write_func(b'got bytes: ' + bytes)
 
303
        self.write_func('got bytes: ' + bytes)
306
304
 
307
305
    def next_read_size(self):
308
306
        return 0
328
326
    def next_read_size(self):
329
327
        # this request always asks for more
330
328
        return 1
 
329