/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: Breezy landing bot
  • Author(s): Colin Watson
  • Date: 2020-11-16 21:47:08 UTC
  • mfrom: (7521.1.1 remove-lp-workaround)
  • Revision ID: breezy.the.bot@gmail.com-20201116214708-jos209mgxi41oy15
Remove breezy.git workaround for bazaar.launchpad.net.

Merged from https://code.launchpad.net/~cjwatson/brz/remove-lp-workaround/+merge/393710

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