← 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: 2009-04-09 20:23:07 UTC
  • mfrom: (4265.1.4 bbc-merge)
  • Revision ID: pqm@pqm.ubuntu.com-20090409202307-n0depb16qepoe21o
(jam) Change _fetch_uses_deltas = False for CHK repos until we can
        write a better fix.

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
23
from bzrlib.transport.http import wsgi
23
24
from bzrlib.transport import chroot, memory
24
25
 
32
33
 
33
34
    def build_environ(self, updates=None):
34
35
        """Builds an environ dict with all fields required by PEP 333.
35
 
        
 
36
 
36
37
        :param updates: a dict to that will be incorporated into the returned
37
38
            dict using dict.update(updates).
38
39
        """
57
58
        if updates is not None:
58
59
            environ.update(updates)
59
60
        return environ
60
 
        
 
61
 
61
62
    def read_response(self, iterable):
62
63
        response = ''
63
64
        for string in iterable:
81
82
        self.read_response(iterable)
82
83
        self.assertEqual('405 Method not allowed', self.status)
83
84
        self.assertTrue(('Allow', 'POST') in self.headers)
84
 
        
 
85
 
 
86
    def _fake_make_request(self, transport, write_func, bytes, rcp):
 
87
        request = FakeRequest(transport, write_func)
 
88
        request.accept_bytes(bytes)
 
89
        self.request = request
 
90
        return request
 
91
 
85
92
    def test_smart_wsgi_app_uses_given_relpath(self):
86
93
        # The SmartWSGIApp should use the "bzrlib.relpath" field from the
87
94
        # WSGI environ to clone from its backing transport to get a specific
89
96
        transport = FakeTransport()
90
97
        wsgi_app = wsgi.SmartWSGIApp(transport)
91
98
        wsgi_app.backing_transport = transport
92
 
        def make_request(transport, write_func):
93
 
            request = FakeRequest(transport, write_func)
94
 
            self.request = request
95
 
            return request
96
 
        wsgi_app.make_request = make_request
 
99
        wsgi_app.make_request = self._fake_make_request
97
100
        fake_input = StringIO('fake request')
98
101
        environ = self.build_environ({
99
102
            'REQUEST_METHOD': 'POST',
103
106
        })
104
107
        iterable = wsgi_app(environ, self.start_response)
105
108
        response = self.read_response(iterable)
106
 
        self.assertEqual([('clone', 'foo/bar')] , transport.calls)
 
109
        self.assertEqual([('clone', 'foo/bar/')] , transport.calls)
107
110
 
108
111
    def test_smart_wsgi_app_request_and_response(self):
109
112
        # SmartWSGIApp reads the smart request from the 'wsgi.input' file-like
112
115
        transport = memory.MemoryTransport()
113
116
        transport.put_bytes('foo', 'some bytes')
114
117
        wsgi_app = wsgi.SmartWSGIApp(transport)
115
 
        def make_request(transport, write_func):
116
 
            request = FakeRequest(transport, write_func)
117
 
            self.request = request
118
 
            return request
119
 
        wsgi_app.make_request = make_request
 
118
        wsgi_app.make_request = self._fake_make_request
120
119
        fake_input = StringIO('fake request')
121
120
        environ = self.build_environ({
122
121
            'REQUEST_METHOD': 'POST',
139
138
            fake_app, prefix='/abc/', path_var='FOO')
140
139
        wrapped_app({'FOO': '/abc/xyz/.bzr/smart'}, None)
141
140
        self.assertEqual(['xyz'], calls)
142
 
       
 
141
 
143
142
    def test_relpath_setter_bad_path_prefix(self):
144
143
        # wsgi.RelpathSetter will reject paths with that don't match the prefix
145
144
        # with a 404.  This is probably a sign of misconfiguration; a server
152
151
            {'FOO': 'AAA/abc/xyz/.bzr/smart'}, self.start_response)
153
152
        self.read_response(iterable)
154
153
        self.assertTrue(self.status.startswith('404'))
155
 
        
 
154
 
156
155
    def test_relpath_setter_bad_path_suffix(self):
157
156
        # Similar to test_relpath_setter_bad_path_prefix: wsgi.RelpathSetter
158
157
        # will reject paths with that don't match the suffix '.bzr/smart' with a
166
165
            {'FOO': '/abc/xyz/.bzr/AAA'}, self.start_response)
167
166
        self.read_response(iterable)
168
167
        self.assertTrue(self.status.startswith('404'))
169
 
        
 
168
 
170
169
    def test_make_app(self):
171
170
        # The make_app helper constructs a SmartWSGIApp wrapped in a
172
171
        # RelpathSetter.
180
179
        backing_transport = app.app.backing_transport
181
180
        chroot_backing_transport = backing_transport.server.backing_transport
182
181
        self.assertEndsWith(chroot_backing_transport.base, 'a%20root/')
183
 
        self.assertEqual(app.prefix, 'a prefix')
 
182
        self.assertEqual(app.app.root_client_path, 'a prefix')
184
183
        self.assertEqual(app.path_var, 'a path_var')
185
184
 
186
185
    def test_incomplete_request(self):
187
186
        transport = FakeTransport()
188
187
        wsgi_app = wsgi.SmartWSGIApp(transport)
189
 
        def make_request(transport, write_func):
 
188
        def make_request(transport, write_func, bytes, root_client_path):
190
189
            request = IncompleteRequest(transport, write_func)
 
190
            request.accept_bytes(bytes)
191
191
            self.request = request
192
192
            return request
193
193
        wsgi_app.make_request = make_request
204
204
        self.assertEqual('200 OK', self.status)
205
205
        self.assertEqual('error\x01incomplete request\n', response)
206
206
 
 
207
    def test_protocol_version_detection_one(self):
 
208
        # SmartWSGIApp detects requests that don't start with
 
209
        # REQUEST_VERSION_TWO as version one.
 
210
        transport = memory.MemoryTransport()
 
211
        wsgi_app = wsgi.SmartWSGIApp(transport)
 
212
        fake_input = StringIO('hello\n')
 
213
        environ = self.build_environ({
 
214
            'REQUEST_METHOD': 'POST',
 
215
            'CONTENT_LENGTH': len(fake_input.getvalue()),
 
216
            'wsgi.input': fake_input,
 
217
            'bzrlib.relpath': 'foo',
 
218
        })
 
219
        iterable = wsgi_app(environ, self.start_response)
 
220
        response = self.read_response(iterable)
 
221
        self.assertEqual('200 OK', self.status)
 
222
        # Expect a version 1-encoded response.
 
223
        self.assertEqual('ok\x012\n', response)
 
224
 
 
225
    def test_protocol_version_detection_two(self):
 
226
        # SmartWSGIApp detects requests that start with REQUEST_VERSION_TWO
 
227
        # as version two.
 
228
        transport = memory.MemoryTransport()
 
229
        wsgi_app = wsgi.SmartWSGIApp(transport)
 
230
        fake_input = StringIO(protocol.REQUEST_VERSION_TWO + 'hello\n')
 
231
        environ = self.build_environ({
 
232
            'REQUEST_METHOD': 'POST',
 
233
            'CONTENT_LENGTH': len(fake_input.getvalue()),
 
234
            'wsgi.input': fake_input,
 
235
            'bzrlib.relpath': 'foo',
 
236
        })
 
237
        iterable = wsgi_app(environ, self.start_response)
 
238
        response = self.read_response(iterable)
 
239
        self.assertEqual('200 OK', self.status)
 
240
        # Expect a version 2-encoded response.
 
241
        self.assertEqual(
 
242
            protocol.RESPONSE_VERSION_TWO + 'success\nok\x012\n', response)
 
243
 
207
244
 
208
245
class FakeRequest(object):
209
 
    
 
246
 
210
247
    def __init__(self, transport, write_func):
211
248
        self.transport = transport
212
249
        self.write_func = write_func