~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_wsgi.py

  • Committer: Canonical.com Patch Queue Manager
  • Date: 2009-09-19 00:32:14 UTC
  • mfrom: (4685.2.1 bzr.dev)
  • Revision ID: pqm@pqm.ubuntu.com-20090919003214-2dli9jc4y5xhjj3n
(mbp for garyvdm) Revert rename of
        test_merge_uncommitted_otherbasis_ancestor_of_thisbasis.

Show diffs side-by-side

added added

removed removed

Lines of Context:
19
19
from cStringIO import StringIO
20
20
 
21
21
from bzrlib import tests
22
 
from bzrlib.smart import medium, message, protocol
 
22
from bzrlib.smart import protocol
23
23
from bzrlib.transport.http import wsgi
24
24
from bzrlib.transport import chroot, memory
25
25
 
26
26
 
27
 
class WSGITestMixin(object):
 
27
class TestWSGI(tests.TestCase):
 
28
 
 
29
    def setUp(self):
 
30
        tests.TestCase.setUp(self)
 
31
        self.status = None
 
32
        self.headers = None
28
33
 
29
34
    def build_environ(self, updates=None):
30
35
        """Builds an environ dict with all fields required by PEP 333.
64
69
        self.status = status
65
70
        self.headers = headers
66
71
 
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
72
    def test_construct(self):
76
73
        app = wsgi.SmartWSGIApp(FakeTransport())
77
74
        self.assertIsInstance(
245
242
            protocol.RESPONSE_VERSION_TWO + 'success\nok\x012\n', response)
246
243
 
247
244
 
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
 
 
293
245
class FakeRequest(object):
294
246
 
295
247
    def __init__(self, transport, write_func):