/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: Robert Collins
  • Date: 2007-03-29 06:03:18 UTC
  • mto: (2018.18.6 hpss-faster-copy)
  • mto: This revision was merged to the branch mainline in revision 2435.
  • Revision ID: robertc@robertcollins.net-20070329060318-uqqabpbnzsnaldx7
Make the InterRepository tests less abusive of the API for InterRepository. (Robert Collins, Andrew Bennetts)

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_wsgi.py
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, message, 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.TestCase, WSGITestMixin):
69
 
 
70
 
    def setUp(self):
71
 
        tests.TestCase.setUp(self)
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(
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):
 
86
        # XXX XXX XXX update comment
96
87
        # The SmartWSGIApp should use the "bzrlib.relpath" field from the
97
88
        # WSGI environ to clone from its backing transport to get a specific
98
89
        # transport for this request.
99
90
        transport = FakeTransport()
100
91
        wsgi_app = wsgi.SmartWSGIApp(transport)
101
92
        wsgi_app.backing_transport = transport
102
 
        wsgi_app.make_request = self._fake_make_request
 
93
        def make_request(transport, write_func):
 
94
            request = FakeRequest(transport, write_func)
 
95
            self.request = request
 
96
            return request
 
97
        wsgi_app.make_request = make_request
103
98
        fake_input = StringIO('fake request')
104
99
        environ = self.build_environ({
105
100
            'REQUEST_METHOD': 'POST',
109
104
        })
110
105
        iterable = wsgi_app(environ, self.start_response)
111
106
        response = self.read_response(iterable)
112
 
        self.assertEqual([('clone', 'foo/bar/')] , transport.calls)
 
107
        self.assertEqual([('clone', 'foo/bar')] , transport.calls)
113
108
 
114
109
    def test_smart_wsgi_app_request_and_response(self):
115
110
        # SmartWSGIApp reads the smart request from the 'wsgi.input' file-like
118
113
        transport = memory.MemoryTransport()
119
114
        transport.put_bytes('foo', 'some bytes')
120
115
        wsgi_app = wsgi.SmartWSGIApp(transport)
121
 
        wsgi_app.make_request = self._fake_make_request
 
116
        def make_request(transport, write_func):
 
117
            request = FakeRequest(transport, write_func)
 
118
            self.request = request
 
119
            return request
 
120
        wsgi_app.make_request = make_request
122
121
        fake_input = StringIO('fake request')
123
122
        environ = self.build_environ({
124
123
            'REQUEST_METHOD': 'POST',
141
140
            fake_app, prefix='/abc/', path_var='FOO')
142
141
        wrapped_app({'FOO': '/abc/xyz/.bzr/smart'}, None)
143
142
        self.assertEqual(['xyz'], calls)
144
 
 
 
143
       
145
144
    def test_relpath_setter_bad_path_prefix(self):
146
145
        # wsgi.RelpathSetter will reject paths with that don't match the prefix
147
146
        # with a 404.  This is probably a sign of misconfiguration; a server
154
153
            {'FOO': 'AAA/abc/xyz/.bzr/smart'}, self.start_response)
155
154
        self.read_response(iterable)
156
155
        self.assertTrue(self.status.startswith('404'))
157
 
 
 
156
        
158
157
    def test_relpath_setter_bad_path_suffix(self):
159
158
        # Similar to test_relpath_setter_bad_path_prefix: wsgi.RelpathSetter
160
159
        # will reject paths with that don't match the suffix '.bzr/smart' with a
168
167
            {'FOO': '/abc/xyz/.bzr/AAA'}, self.start_response)
169
168
        self.read_response(iterable)
170
169
        self.assertTrue(self.status.startswith('404'))
171
 
 
 
170
        
172
171
    def test_make_app(self):
173
172
        # The make_app helper constructs a SmartWSGIApp wrapped in a
174
173
        # RelpathSetter.
182
181
        backing_transport = app.app.backing_transport
183
182
        chroot_backing_transport = backing_transport.server.backing_transport
184
183
        self.assertEndsWith(chroot_backing_transport.base, 'a%20root/')
185
 
        self.assertEqual(app.app.root_client_path, 'a prefix')
 
184
        self.assertEqual(app.prefix, 'a prefix')
186
185
        self.assertEqual(app.path_var, 'a path_var')
187
186
 
188
187
    def test_incomplete_request(self):
189
188
        transport = FakeTransport()
190
189
        wsgi_app = wsgi.SmartWSGIApp(transport)
191
 
        def make_request(transport, write_func, bytes, root_client_path):
 
190
        def make_request(transport, write_func):
192
191
            request = IncompleteRequest(transport, write_func)
193
 
            request.accept_bytes(bytes)
194
192
            self.request = request
195
193
            return request
196
194
        wsgi_app.make_request = make_request
207
205
        self.assertEqual('200 OK', self.status)
208
206
        self.assertEqual('error\x01incomplete request\n', response)
209
207
 
210
 
    def test_protocol_version_detection_one(self):
211
 
        # SmartWSGIApp detects requests that don't start with
212
 
        # REQUEST_VERSION_TWO as version one.
213
 
        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)
227
 
 
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')
234
 
        environ = self.build_environ({
235
 
            'REQUEST_METHOD': 'POST',
236
 
            'CONTENT_LENGTH': len(fake_input.getvalue()),
237
 
            'wsgi.input': fake_input,
238
 
            'bzrlib.relpath': 'foo',
239
 
        })
240
 
        iterable = wsgi_app(environ, self.start_response)
241
 
        response = self.read_response(iterable)
242
 
        self.assertEqual('200 OK', self.status)
243
 
        # Expect a version 2-encoded response.
244
 
        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)
291
 
 
292
208
 
293
209
class FakeRequest(object):
294
 
 
 
210
    
295
211
    def __init__(self, transport, write_func):
296
212
        self.transport = transport
297
213
        self.write_func = write_func