/breezy/unstable

To get this branch, use:
bzr branch https://code.breezy-vcs.org/breezy/unstable

« back to all changes in this revision

Viewing changes to breezy/tests/test_wsgi.py

  • Committer: Jelmer Vernooij
  • Date: 2017-05-24 01:39:33 UTC
  • mfrom: (3815.3776.6)
  • Revision ID: jelmer@jelmer.uk-20170524013933-ir4y4tqtrsiz2ka2
New upstream snapshot.

Show diffs side-by-side

added added

removed removed

Lines of Context:
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, 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 ..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):
44
45
            # Required WSGI variables
45
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,
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)
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
 
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)