← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_wsgi.py

  • Committer: Martin Pool
  • Date: 2007-03-22 05:27:08 UTC
  • mto: (2323.5.2 0.15)
  • mto: This revision was merged to the branch mainline in revision 2390.
  • Revision ID: mbp@sourcefrog.net-20070322052708-t4g4iv46n2dcjjaa
(broken) Give a message when opening old workingtree formats suggesting upgrade

This breaks some blackbox tests of old formats that don't expect to see this
message.

Add WorkingTree._after_opening callback

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_wsgi.py
1
 
# Copyright (C) 2006-2009, 2011 Canonical Ltd
 
1
# Copyright (C) 2006 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
12
12
#
13
13
# You should have received a copy of the GNU General Public License
14
14
# along with this program; if not, write to the Free Software
15
 
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
 
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
16
16
 
17
17
"""Tests for WSGI application"""
18
18
 
19
19
from cStringIO import StringIO
20
20
 
21
21
from bzrlib import tests
22
 
from bzrlib.smart import medium, protocol
23
22
from bzrlib.transport.http import wsgi
24
23
from bzrlib.transport import chroot, memory
25
24
 
26
25
 
27
 
class WSGITestMixin(object):
 
26
class TestWSGI(tests.TestCase):
 
27
 
 
28
    def setUp(self):
 
29
        tests.TestCase.setUp(self)
 
30
        self.status = None
 
31
        self.headers = None
28
32
 
29
33
    def build_environ(self, updates=None):
30
34
        """Builds an environ dict with all fields required by PEP 333.
31
 
 
 
35
        
32
36
        :param updates: a dict to that will be incorporated into the returned
33
37
            dict using dict.update(updates).
34
38
        """
53
57
        if updates is not None:
54
58
            environ.update(updates)
55
59
        return environ
56
 
 
 
60
        
57
61
    def read_response(self, iterable):
58
62
        response = ''
59
63
        for string in iterable:
64
68
        self.status = status
65
69
        self.headers = headers
66
70
 
67
 
 
68
 
class TestWSGI(tests.TestCaseInTempDir, WSGITestMixin):
69
 
 
70
 
    def setUp(self):
71
 
        super(TestWSGI, self).setUp()
72
 
        self.status = None
73
 
        self.headers = None
74
 
 
75
71
    def test_construct(self):
76
72
        app = wsgi.SmartWSGIApp(FakeTransport())
77
73
        self.assertIsInstance(
78
 
            app.backing_transport, chroot.ChrootTransport)
 
74
            app.backing_transport, chroot.ChrootTransportDecorator)
79
75
 
80
76
    def test_http_get_rejected(self):
81
77
        # GET requests are rejected.
85
81
        self.read_response(iterable)
86
82
        self.assertEqual('405 Method not allowed', self.status)
87
83
        self.assertTrue(('Allow', 'POST') in self.headers)
88
 
 
89
 
    def _fake_make_request(self, transport, write_func, bytes, rcp):
90
 
        request = FakeRequest(transport, write_func)
91
 
        request.accept_bytes(bytes)
92
 
        self.request = request
93
 
        return request
94
 
 
 
84
        
95
85
    def test_smart_wsgi_app_uses_given_relpath(self):
96
86
        # The SmartWSGIApp should use the "bzrlib.relpath" field from the
97
 
        # WSGI environ to clone from its backing transport to get a specific
98
 
        # transport for this request.
 
87
        # WSGI environ to construct the transport for this request, by cloning
 
88
        # its base transport with the given relpath.
99
89
        transport = FakeTransport()
100
90
        wsgi_app = wsgi.SmartWSGIApp(transport)
101
 
        wsgi_app.backing_transport = transport
102
 
        wsgi_app.make_request = self._fake_make_request
 
91
        def make_request(transport, write_func):
 
92
            request = FakeRequest(transport, write_func)
 
93
            self.request = request
 
94
            return request
 
95
        wsgi_app.make_request = make_request
103
96
        fake_input = StringIO('fake request')
104
97
        environ = self.build_environ({
105
98
            'REQUEST_METHOD': 'POST',
109
102
        })
110
103
        iterable = wsgi_app(environ, self.start_response)
111
104
        response = self.read_response(iterable)
112
 
        self.assertEqual([('clone', 'foo/bar/')] , transport.calls)
 
105
        self.assertEqual([('clone', 'foo/bar')] , transport.calls)
113
106
 
114
107
    def test_smart_wsgi_app_request_and_response(self):
115
108
        # SmartWSGIApp reads the smart request from the 'wsgi.input' file-like
118
111
        transport = memory.MemoryTransport()
119
112
        transport.put_bytes('foo', 'some bytes')
120
113
        wsgi_app = wsgi.SmartWSGIApp(transport)
121
 
        wsgi_app.make_request = self._fake_make_request
 
114
        def make_request(transport, write_func):
 
115
            request = FakeRequest(transport, write_func)
 
116
            self.request = request
 
117
            return request
 
118
        wsgi_app.make_request = make_request
122
119
        fake_input = StringIO('fake request')
123
120
        environ = self.build_environ({
124
121
            'REQUEST_METHOD': 'POST',
141
138
            fake_app, prefix='/abc/', path_var='FOO')
142
139
        wrapped_app({'FOO': '/abc/xyz/.bzr/smart'}, None)
143
140
        self.assertEqual(['xyz'], calls)
144
 
 
 
141
       
145
142
    def test_relpath_setter_bad_path_prefix(self):
146
143
        # wsgi.RelpathSetter will reject paths with that don't match the prefix
147
144
        # with a 404.  This is probably a sign of misconfiguration; a server
154
151
            {'FOO': 'AAA/abc/xyz/.bzr/smart'}, self.start_response)
155
152
        self.read_response(iterable)
156
153
        self.assertTrue(self.status.startswith('404'))
157
 
 
 
154
        
158
155
    def test_relpath_setter_bad_path_suffix(self):
159
156
        # Similar to test_relpath_setter_bad_path_prefix: wsgi.RelpathSetter
160
157
        # will reject paths with that don't match the suffix '.bzr/smart' with a
168
165
            {'FOO': '/abc/xyz/.bzr/AAA'}, self.start_response)
169
166
        self.read_response(iterable)
170
167
        self.assertTrue(self.status.startswith('404'))
171
 
 
 
168
        
172
169
    def test_make_app(self):
173
170
        # The make_app helper constructs a SmartWSGIApp wrapped in a
174
171
        # RelpathSetter.
178
175
            path_var='a path_var')
179
176
        self.assertIsInstance(app, wsgi.RelpathSetter)
180
177
        self.assertIsInstance(app.app, wsgi.SmartWSGIApp)
181
 
        self.assertStartsWith(app.app.backing_transport.base, 'chroot-')
182
 
        backing_transport = app.app.backing_transport
183
 
        chroot_backing_transport = backing_transport.server.backing_transport
184
 
        self.assertEndsWith(chroot_backing_transport.base, 'a%20root/')
185
 
        self.assertEqual(app.app.root_client_path, 'a prefix')
 
178
        self.assertEndsWith(app.app.backing_transport.base, 'a%20root/')
 
179
        self.assertEqual(app.prefix, 'a prefix')
186
180
        self.assertEqual(app.path_var, 'a path_var')
187
181
 
188
182
    def test_incomplete_request(self):
189
183
        transport = FakeTransport()
190
184
        wsgi_app = wsgi.SmartWSGIApp(transport)
191
 
        def make_request(transport, write_func, bytes, root_client_path):
 
185
        def make_request(transport, write_func):
192
186
            request = IncompleteRequest(transport, write_func)
193
 
            request.accept_bytes(bytes)
194
187
            self.request = request
195
188
            return request
196
189
        wsgi_app.make_request = make_request
207
200
        self.assertEqual('200 OK', self.status)
208
201
        self.assertEqual('error\x01incomplete request\n', response)
209
202
 
210
 
    def test_protocol_version_detection_one(self):
211
 
        # SmartWSGIApp detects requests that don't start with
212
 
        # REQUEST_VERSION_TWO as version one.
 
203
    def test_chrooting(self):
 
204
        # Show that requests that try to access things outside of the base
 
205
        # really will get intercepted by the ChrootTransportDecorator.
213
206
        transport = memory.MemoryTransport()
214
 
        wsgi_app = wsgi.SmartWSGIApp(transport)
215
 
        fake_input = StringIO('hello\n')
216
 
        environ = self.build_environ({
217
 
            'REQUEST_METHOD': 'POST',
218
 
            'CONTENT_LENGTH': len(fake_input.getvalue()),
219
 
            'wsgi.input': fake_input,
220
 
            'bzrlib.relpath': 'foo',
221
 
        })
222
 
        iterable = wsgi_app(environ, self.start_response)
223
 
        response = self.read_response(iterable)
224
 
        self.assertEqual('200 OK', self.status)
225
 
        # Expect a version 1-encoded response.
226
 
        self.assertEqual('ok\x012\n', response)
 
207
        transport.mkdir('foo')
 
208
        transport.put_bytes('foo/bar', 'this is foo/bar')
 
209
        wsgi_app = wsgi.SmartWSGIApp(transport.clone('foo'))
227
210
 
228
 
    def test_protocol_version_detection_two(self):
229
 
        # SmartWSGIApp detects requests that start with REQUEST_VERSION_TWO
230
 
        # as version two.
231
 
        transport = memory.MemoryTransport()
232
 
        wsgi_app = wsgi.SmartWSGIApp(transport)
233
 
        fake_input = StringIO(protocol.REQUEST_VERSION_TWO + 'hello\n')
 
211
        smart_request = StringIO('mkdir\x01/bad file\x01\n0\ndone\n')
234
212
        environ = self.build_environ({
235
213
            'REQUEST_METHOD': 'POST',
236
 
            'CONTENT_LENGTH': len(fake_input.getvalue()),
237
 
            'wsgi.input': fake_input,
238
 
            'bzrlib.relpath': 'foo',
 
214
            'CONTENT_LENGTH': len(smart_request.getvalue()),
 
215
            'wsgi.input': smart_request,
 
216
            'bzrlib.relpath': '.',
239
217
        })
240
218
        iterable = wsgi_app(environ, self.start_response)
241
219
        response = self.read_response(iterable)
242
220
        self.assertEqual('200 OK', self.status)
243
 
        # Expect a version 2-encoded response.
244
221
        self.assertEqual(
245
 
            protocol.RESPONSE_VERSION_TWO + 'success\nok\x012\n', response)
246
 
 
247
 
 
248
 
class TestWSGIJail(tests.TestCaseWithMemoryTransport, WSGITestMixin):
249
 
 
250
 
    def make_hpss_wsgi_request(self, wsgi_relpath, *args):
251
 
        write_buf = StringIO()
252
 
        request_medium = medium.SmartSimplePipesClientMedium(
253
 
            None, write_buf, 'fake:' + wsgi_relpath)
254
 
        request_encoder = protocol.ProtocolThreeRequester(
255
 
            request_medium.get_request())
256
 
        request_encoder.call(*args)
257
 
        write_buf.seek(0)
258
 
        environ = self.build_environ({
259
 
            'REQUEST_METHOD': 'POST',
260
 
            'CONTENT_LENGTH': len(write_buf.getvalue()),
261
 
            'wsgi.input': write_buf,
262
 
            'bzrlib.relpath': wsgi_relpath,
263
 
        })
264
 
        return environ
265
 
 
266
 
    def test_jail_root(self):
267
 
        """The WSGI HPSS glue allows access to the whole WSGI backing
268
 
        transport, regardless of which HTTP path the request was delivered
269
 
        to.
270
 
        """
271
 
        # make a branch in a shared repo
272
 
        self.make_repository('repo', shared=True)
273
 
        branch = self.make_bzrdir('repo/branch').create_branch()
274
 
        # serve the repo via bzr+http WSGI
275
 
        wsgi_app = wsgi.SmartWSGIApp(self.get_transport())
276
 
        # send a request to /repo/branch that will have to access /repo.
277
 
        environ = self.make_hpss_wsgi_request(
278
 
            '/repo/branch', 'BzrDir.open_branchV2', '.')
279
 
        iterable = wsgi_app(environ, self.start_response)
280
 
        response_bytes = self.read_response(iterable)
281
 
        self.assertEqual('200 OK', self.status)
282
 
        # expect a successful response, rather than a jail break error
283
 
        from bzrlib.tests.test_smart_transport import LoggingMessageHandler
284
 
        message_handler = LoggingMessageHandler()
285
 
        decoder = protocol.ProtocolThreeDecoder(
286
 
            message_handler, expect_version_marker=True)
287
 
        decoder.accept_bytes(response_bytes)
288
 
        self.assertTrue(
289
 
            ('structure', ('branch', branch._format.network_name()))
290
 
            in message_handler.event_log)
 
222
            "error\x01Path '/bad file' is not a child of "
 
223
            "path 'memory:///foo/'\n",
 
224
            response)
291
225
 
292
226
 
293
227
class FakeRequest(object):
294
 
 
 
228
    
295
229
    def __init__(self, transport, write_func):
296
230
        self.transport = transport
297
231
        self.write_func = write_func