/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: 2018-09-13 12:50:28 UTC
  • mfrom: (7096.2.2 empty-port)
  • Revision ID: breezy.the.bot@gmail.com-20180913125028-mja5gz8xsams9iey
Allow port to be empty when parsing URLs.

Merged from https://code.launchpad.net/~jelmer/brz/empty-port/+merge/354640

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 = []
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)
195
196
            return request
196
197
        wsgi_app.make_request = make_request
197
198
 
198
 
        fake_input = StringIO('incomplete request')
 
199
        fake_input = BytesIO(b'incomplete request')
199
200
        environ = self.build_environ({
200
201
            'REQUEST_METHOD': 'POST',
201
202
            'CONTENT_LENGTH': len(fake_input.getvalue()),
202
203
            'wsgi.input': fake_input,
203
 
            'bzrlib.relpath': 'foo/bar',
 
204
            'breezy.relpath': 'foo/bar',
204
205
        })
205
206
        iterable = wsgi_app(environ, self.start_response)
206
207
        response = self.read_response(iterable)
207
208
        self.assertEqual('200 OK', self.status)
208
 
        self.assertEqual('error\x01incomplete request\n', response)
 
209
        self.assertEqual(b'error\x01incomplete request\n', response)
209
210
 
210
211
    def test_protocol_version_detection_one(self):
211
212
        # SmartWSGIApp detects requests that don't start with
212
213
        # REQUEST_VERSION_TWO as version one.
213
214
        transport = memory.MemoryTransport()
214
215
        wsgi_app = wsgi.SmartWSGIApp(transport)
215
 
        fake_input = StringIO('hello\n')
 
216
        fake_input = BytesIO(b'hello\n')
216
217
        environ = self.build_environ({
217
218
            'REQUEST_METHOD': 'POST',
218
219
            'CONTENT_LENGTH': len(fake_input.getvalue()),
219
220
            'wsgi.input': fake_input,
220
 
            'bzrlib.relpath': 'foo',
 
221
            'breezy.relpath': 'foo',
221
222
        })
222
223
        iterable = wsgi_app(environ, self.start_response)
223
224
        response = self.read_response(iterable)
224
225
        self.assertEqual('200 OK', self.status)
225
226
        # Expect a version 1-encoded response.
226
 
        self.assertEqual('ok\x012\n', response)
 
227
        self.assertEqual(b'ok\x012\n', response)
227
228
 
228
229
    def test_protocol_version_detection_two(self):
229
230
        # SmartWSGIApp detects requests that start with REQUEST_VERSION_TWO
230
231
        # as version two.
231
232
        transport = memory.MemoryTransport()
232
233
        wsgi_app = wsgi.SmartWSGIApp(transport)
233
 
        fake_input = StringIO(protocol.REQUEST_VERSION_TWO + 'hello\n')
 
234
        fake_input = BytesIO(protocol.REQUEST_VERSION_TWO + b'hello\n')
234
235
        environ = self.build_environ({
235
236
            'REQUEST_METHOD': 'POST',
236
237
            'CONTENT_LENGTH': len(fake_input.getvalue()),
237
238
            'wsgi.input': fake_input,
238
 
            'bzrlib.relpath': 'foo',
 
239
            'breezy.relpath': 'foo',
239
240
        })
240
241
        iterable = wsgi_app(environ, self.start_response)
241
242
        response = self.read_response(iterable)
242
243
        self.assertEqual('200 OK', self.status)
243
244
        # Expect a version 2-encoded response.
244
245
        self.assertEqual(
245
 
            protocol.RESPONSE_VERSION_TWO + 'success\nok\x012\n', response)
 
246
            protocol.RESPONSE_VERSION_TWO + b'success\nok\x012\n', response)
246
247
 
247
248
 
248
249
class TestWSGIJail(tests.TestCaseWithMemoryTransport, WSGITestMixin):
249
250
 
250
251
    def make_hpss_wsgi_request(self, wsgi_relpath, *args):
251
 
        write_buf = StringIO()
 
252
        write_buf = BytesIO()
252
253
        request_medium = medium.SmartSimplePipesClientMedium(
253
254
            None, write_buf, 'fake:' + wsgi_relpath)
254
255
        request_encoder = protocol.ProtocolThreeRequester(
259
260
            'REQUEST_METHOD': 'POST',
260
261
            'CONTENT_LENGTH': len(write_buf.getvalue()),
261
262
            'wsgi.input': write_buf,
262
 
            'bzrlib.relpath': wsgi_relpath,
 
263
            'breezy.relpath': wsgi_relpath,
263
264
        })
264
265
        return environ
265
266
 
270
271
        """
271
272
        # make a branch in a shared repo
272
273
        self.make_repository('repo', shared=True)
273
 
        branch = self.make_bzrdir('repo/branch').create_branch()
 
274
        branch = self.make_controldir('repo/branch').create_branch()
274
275
        # serve the repo via bzr+http WSGI
275
276
        wsgi_app = wsgi.SmartWSGIApp(self.get_transport())
276
277
        # send a request to /repo/branch that will have to access /repo.
277
278
        environ = self.make_hpss_wsgi_request(
278
 
            '/repo/branch', 'BzrDir.open_branchV2', '.')
 
279
            '/repo/branch', b'BzrDir.open_branchV2', b'.')
279
280
        iterable = wsgi_app(environ, self.start_response)
280
281
        response_bytes = self.read_response(iterable)
281
282
        self.assertEqual('200 OK', self.status)
282
283
        # expect a successful response, rather than a jail break error
283
 
        from bzrlib.tests.test_smart_transport import LoggingMessageHandler
 
284
        from breezy.tests.test_smart_transport import LoggingMessageHandler
284
285
        message_handler = LoggingMessageHandler()
285
286
        decoder = protocol.ProtocolThreeDecoder(
286
287
            message_handler, expect_version_marker=True)
287
288
        decoder.accept_bytes(response_bytes)
288
289
        self.assertTrue(
289
 
            ('structure', ('branch', branch._format.network_name()))
 
290
            ('structure', (b'branch', branch._format.network_name()))
290
291
            in message_handler.event_log)
291
292
 
292
293
 
295
296
    def __init__(self, transport, write_func):
296
297
        self.transport = transport
297
298
        self.write_func = write_func
298
 
        self.accepted_bytes = ''
 
299
        self.accepted_bytes = b''
299
300
 
300
301
    def accept_bytes(self, bytes):
301
302
        self.accepted_bytes = bytes
302
 
        self.write_func('got bytes: ' + bytes)
 
303
        self.write_func(b'got bytes: ' + bytes)
303
304
 
304
305
    def next_read_size(self):
305
306
        return 0