/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 bzrlib/tests/test_wsgi.py

  • Committer: Richard Wilbur
  • Date: 2016-02-04 19:07:28 UTC
  • mto: This revision was merged to the branch mainline in revision 6618.
  • Revision ID: richard.wilbur@gmail.com-20160204190728-p0zvfii6zase0fw7
Update COPYING.txt from the original http://www.gnu.org/licenses/gpl-2.0.txt  (Only differences were in whitespace.)  Thanks to Petr Stodulka for pointing out the discrepancy.

Show diffs side-by-side

added added

removed removed

Lines of Context:
16
16
 
17
17
"""Tests for WSGI application"""
18
18
 
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
 
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
26
25
 
27
26
 
28
27
class WSGITestMixin(object):
43
42
            'SERVER_PROTOCOL': 'HTTP/1.0',
44
43
 
45
44
            # Required WSGI variables
46
 
            'wsgi.version': (1, 0),
 
45
            'wsgi.version': (1,0),
47
46
            'wsgi.url_scheme': 'http',
48
 
            'wsgi.input': BytesIO(b''),
49
 
            'wsgi.errors': BytesIO(),
 
47
            'wsgi.input': StringIO(''),
 
48
            'wsgi.errors': StringIO(),
50
49
            'wsgi.multithread': False,
51
50
            'wsgi.multiprocess': False,
52
51
            'wsgi.run_once': True,
94
93
        return request
95
94
 
96
95
    def test_smart_wsgi_app_uses_given_relpath(self):
97
 
        # The SmartWSGIApp should use the "breezy.relpath" field from the
 
96
        # The SmartWSGIApp should use the "bzrlib.relpath" field from the
98
97
        # WSGI environ to clone from its backing transport to get a specific
99
98
        # transport for this request.
100
99
        transport = FakeTransport()
101
100
        wsgi_app = wsgi.SmartWSGIApp(transport)
102
101
        wsgi_app.backing_transport = transport
103
102
        wsgi_app.make_request = self._fake_make_request
104
 
        fake_input = BytesIO(b'fake request')
 
103
        fake_input = StringIO('fake request')
105
104
        environ = self.build_environ({
106
105
            'REQUEST_METHOD': 'POST',
107
106
            'CONTENT_LENGTH': len(fake_input.getvalue()),
108
107
            'wsgi.input': fake_input,
109
 
            'breezy.relpath': 'foo/bar',
 
108
            'bzrlib.relpath': 'foo/bar',
110
109
        })
111
110
        iterable = wsgi_app(environ, self.start_response)
112
111
        response = self.read_response(iterable)
113
 
        self.assertEqual([('clone', 'foo/bar/')], transport.calls)
 
112
        self.assertEqual([('clone', 'foo/bar/')] , transport.calls)
114
113
 
115
114
    def test_smart_wsgi_app_request_and_response(self):
116
115
        # SmartWSGIApp reads the smart request from the 'wsgi.input' file-like
120
119
        transport.put_bytes('foo', 'some bytes')
121
120
        wsgi_app = wsgi.SmartWSGIApp(transport)
122
121
        wsgi_app.make_request = self._fake_make_request
123
 
        fake_input = BytesIO(b'fake request')
 
122
        fake_input = StringIO('fake request')
124
123
        environ = self.build_environ({
125
124
            'REQUEST_METHOD': 'POST',
126
125
            'CONTENT_LENGTH': len(fake_input.getvalue()),
127
126
            'wsgi.input': fake_input,
128
 
            'breezy.relpath': 'foo',
 
127
            'bzrlib.relpath': 'foo',
129
128
        })
130
129
        iterable = wsgi_app(environ, self.start_response)
131
130
        response = self.read_response(iterable)
133
132
        self.assertEqual('got bytes: fake request', response)
134
133
 
135
134
    def test_relpath_setter(self):
136
 
        # wsgi.RelpathSetter is WSGI "middleware" to set the 'breezy.relpath'
 
135
        # wsgi.RelpathSetter is WSGI "middleware" to set the 'bzrlib.relpath'
137
136
        # variable.
138
137
        calls = []
139
138
        def fake_app(environ, start_response):
140
 
            calls.append(environ['breezy.relpath'])
 
139
            calls.append(environ['bzrlib.relpath'])
141
140
        wrapped_app = wsgi.RelpathSetter(
142
141
            fake_app, prefix='/abc/', path_var='FOO')
143
142
        wrapped_app({'FOO': '/abc/xyz/.bzr/smart'}, None)
196
195
            return request
197
196
        wsgi_app.make_request = make_request
198
197
 
199
 
        fake_input = BytesIO(b'incomplete request')
 
198
        fake_input = StringIO('incomplete request')
200
199
        environ = self.build_environ({
201
200
            'REQUEST_METHOD': 'POST',
202
201
            'CONTENT_LENGTH': len(fake_input.getvalue()),
203
202
            'wsgi.input': fake_input,
204
 
            'breezy.relpath': 'foo/bar',
 
203
            'bzrlib.relpath': 'foo/bar',
205
204
        })
206
205
        iterable = wsgi_app(environ, self.start_response)
207
206
        response = self.read_response(iterable)
213
212
        # REQUEST_VERSION_TWO as version one.
214
213
        transport = memory.MemoryTransport()
215
214
        wsgi_app = wsgi.SmartWSGIApp(transport)
216
 
        fake_input = BytesIO(b'hello\n')
 
215
        fake_input = StringIO('hello\n')
217
216
        environ = self.build_environ({
218
217
            'REQUEST_METHOD': 'POST',
219
218
            'CONTENT_LENGTH': len(fake_input.getvalue()),
220
219
            'wsgi.input': fake_input,
221
 
            'breezy.relpath': 'foo',
 
220
            'bzrlib.relpath': 'foo',
222
221
        })
223
222
        iterable = wsgi_app(environ, self.start_response)
224
223
        response = self.read_response(iterable)
231
230
        # as version two.
232
231
        transport = memory.MemoryTransport()
233
232
        wsgi_app = wsgi.SmartWSGIApp(transport)
234
 
        fake_input = BytesIO(protocol.REQUEST_VERSION_TWO + 'hello\n')
 
233
        fake_input = StringIO(protocol.REQUEST_VERSION_TWO + 'hello\n')
235
234
        environ = self.build_environ({
236
235
            'REQUEST_METHOD': 'POST',
237
236
            'CONTENT_LENGTH': len(fake_input.getvalue()),
238
237
            'wsgi.input': fake_input,
239
 
            'breezy.relpath': 'foo',
 
238
            'bzrlib.relpath': 'foo',
240
239
        })
241
240
        iterable = wsgi_app(environ, self.start_response)
242
241
        response = self.read_response(iterable)
249
248
class TestWSGIJail(tests.TestCaseWithMemoryTransport, WSGITestMixin):
250
249
 
251
250
    def make_hpss_wsgi_request(self, wsgi_relpath, *args):
252
 
        write_buf = BytesIO()
 
251
        write_buf = StringIO()
253
252
        request_medium = medium.SmartSimplePipesClientMedium(
254
253
            None, write_buf, 'fake:' + wsgi_relpath)
255
254
        request_encoder = protocol.ProtocolThreeRequester(
260
259
            'REQUEST_METHOD': 'POST',
261
260
            'CONTENT_LENGTH': len(write_buf.getvalue()),
262
261
            'wsgi.input': write_buf,
263
 
            'breezy.relpath': wsgi_relpath,
 
262
            'bzrlib.relpath': wsgi_relpath,
264
263
        })
265
264
        return environ
266
265
 
271
270
        """
272
271
        # make a branch in a shared repo
273
272
        self.make_repository('repo', shared=True)
274
 
        branch = self.make_controldir('repo/branch').create_branch()
 
273
        branch = self.make_bzrdir('repo/branch').create_branch()
275
274
        # serve the repo via bzr+http WSGI
276
275
        wsgi_app = wsgi.SmartWSGIApp(self.get_transport())
277
276
        # send a request to /repo/branch that will have to access /repo.
281
280
        response_bytes = self.read_response(iterable)
282
281
        self.assertEqual('200 OK', self.status)
283
282
        # expect a successful response, rather than a jail break error
284
 
        from breezy.tests.test_smart_transport import LoggingMessageHandler
 
283
        from bzrlib.tests.test_smart_transport import LoggingMessageHandler
285
284
        message_handler = LoggingMessageHandler()
286
285
        decoder = protocol.ProtocolThreeDecoder(
287
286
            message_handler, expect_version_marker=True)