/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): Jelmer Vernooij
  • Date: 2019-10-13 17:31:55 UTC
  • mfrom: (7397.4.9 remove-unused)
  • Revision ID: breezy.the.bot@gmail.com-20191013173155-yoiokny4mknxb3um
Remove Tree.has_id.

Merged from https://code.launchpad.net/~jelmer/brz/remove-unused/+merge/373320

Show diffs side-by-side

added added

removed removed

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