/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-07-08 14:45:27 UTC
  • mto: This revision was merged to the branch mainline in revision 7036.
  • Revision ID: jelmer@jelmer.uk-20180708144527-codhlvdcdg9y0nji
Fix a bunch of merge tests.

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,
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
119
120
        transport.put_bytes('foo', '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)
132
133
        self.assertEqual('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)
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)
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 + '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)
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)