← 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: Canonical.com Patch Queue Manager
  • Date: 2010-01-15 04:49:48 UTC
  • mfrom: (3984.5.22 switch-r-183559)
  • Revision ID: pqm@pqm.ubuntu.com-20100115044948-yxz5m3vchxapbq22
(andrew) Add --revision option to 'bzr switch'. (#184559)

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., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 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 protocol
 
22
from bzrlib.smart import medium, message, protocol
23
23
from bzrlib.transport.http import wsgi
24
24
from bzrlib.transport import chroot, memory
25
25
 
26
26
 
27
 
class TestWSGI(tests.TestCase):
28
 
 
29
 
    def setUp(self):
30
 
        tests.TestCase.setUp(self)
31
 
        self.status = None
32
 
        self.headers = None
 
27
class WSGITestMixin(object):
33
28
 
34
29
    def build_environ(self, updates=None):
35
30
        """Builds an environ dict with all fields required by PEP 333.
36
 
        
 
31
 
37
32
        :param updates: a dict to that will be incorporated into the returned
38
33
            dict using dict.update(updates).
39
34
        """
58
53
        if updates is not None:
59
54
            environ.update(updates)
60
55
        return environ
61
 
        
 
56
 
62
57
    def read_response(self, iterable):
63
58
        response = ''
64
59
        for string in iterable:
69
64
        self.status = status
70
65
        self.headers = headers
71
66
 
 
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
 
72
75
    def test_construct(self):
73
76
        app = wsgi.SmartWSGIApp(FakeTransport())
74
77
        self.assertIsInstance(
82
85
        self.read_response(iterable)
83
86
        self.assertEqual('405 Method not allowed', self.status)
84
87
        self.assertTrue(('Allow', 'POST') in self.headers)
85
 
        
 
88
 
86
89
    def _fake_make_request(self, transport, write_func, bytes, rcp):
87
90
        request = FakeRequest(transport, write_func)
88
91
        request.accept_bytes(bytes)
89
92
        self.request = request
90
93
        return request
91
 
    
 
94
 
92
95
    def test_smart_wsgi_app_uses_given_relpath(self):
93
96
        # The SmartWSGIApp should use the "bzrlib.relpath" field from the
94
97
        # WSGI environ to clone from its backing transport to get a specific
138
141
            fake_app, prefix='/abc/', path_var='FOO')
139
142
        wrapped_app({'FOO': '/abc/xyz/.bzr/smart'}, None)
140
143
        self.assertEqual(['xyz'], calls)
141
 
       
 
144
 
142
145
    def test_relpath_setter_bad_path_prefix(self):
143
146
        # wsgi.RelpathSetter will reject paths with that don't match the prefix
144
147
        # with a 404.  This is probably a sign of misconfiguration; a server
151
154
            {'FOO': 'AAA/abc/xyz/.bzr/smart'}, self.start_response)
152
155
        self.read_response(iterable)
153
156
        self.assertTrue(self.status.startswith('404'))
154
 
        
 
157
 
155
158
    def test_relpath_setter_bad_path_suffix(self):
156
159
        # Similar to test_relpath_setter_bad_path_prefix: wsgi.RelpathSetter
157
160
        # will reject paths with that don't match the suffix '.bzr/smart' with a
165
168
            {'FOO': '/abc/xyz/.bzr/AAA'}, self.start_response)
166
169
        self.read_response(iterable)
167
170
        self.assertTrue(self.status.startswith('404'))
168
 
        
 
171
 
169
172
    def test_make_app(self):
170
173
        # The make_app helper constructs a SmartWSGIApp wrapped in a
171
174
        # RelpathSetter.
242
245
            protocol.RESPONSE_VERSION_TWO + 'success\nok\x012\n', response)
243
246
 
244
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
 
245
293
class FakeRequest(object):
246
 
    
 
294
 
247
295
    def __init__(self, transport, write_func):
248
296
        self.transport = transport
249
297
        self.write_func = write_func