~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: 2007-04-17 00:59:30 UTC
  • mfrom: (1551.15.4 Aaron's mergeable stuff)
  • Revision ID: pqm@pqm.ubuntu.com-20070417005930-rofskshyjsfzrahh
Fix ftp transport with servers that don't support atomic rename

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 protocol
23
22
from bzrlib.transport.http import wsgi
24
23
from bzrlib.transport import chroot, memory
25
24
 
83
82
        self.assertEqual('405 Method not allowed', self.status)
84
83
        self.assertTrue(('Allow', 'POST') in self.headers)
85
84
        
86
 
    def _fake_make_request(self, transport, write_func, bytes):
87
 
        request = FakeRequest(transport, write_func)
88
 
        request.accept_bytes(bytes)
89
 
        self.request = request
90
 
        return request
91
 
    
92
85
    def test_smart_wsgi_app_uses_given_relpath(self):
93
86
        # The SmartWSGIApp should use the "bzrlib.relpath" field from the
94
87
        # WSGI environ to clone from its backing transport to get a specific
96
89
        transport = FakeTransport()
97
90
        wsgi_app = wsgi.SmartWSGIApp(transport)
98
91
        wsgi_app.backing_transport = transport
99
 
        wsgi_app.make_request = self._fake_make_request
 
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
100
97
        fake_input = StringIO('fake request')
101
98
        environ = self.build_environ({
102
99
            'REQUEST_METHOD': 'POST',
115
112
        transport = memory.MemoryTransport()
116
113
        transport.put_bytes('foo', 'some bytes')
117
114
        wsgi_app = wsgi.SmartWSGIApp(transport)
118
 
        wsgi_app.make_request = self._fake_make_request
 
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
119
120
        fake_input = StringIO('fake request')
120
121
        environ = self.build_environ({
121
122
            'REQUEST_METHOD': 'POST',
185
186
    def test_incomplete_request(self):
186
187
        transport = FakeTransport()
187
188
        wsgi_app = wsgi.SmartWSGIApp(transport)
188
 
        def make_request(transport, write_func, bytes):
 
189
        def make_request(transport, write_func):
189
190
            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
 
 
244
207
 
245
208
class FakeRequest(object):
246
209