19
19
from cStringIO import StringIO
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
82
83
self.assertEqual('405 Method not allowed', self.status)
83
84
self.assertTrue(('Allow', 'POST') in self.headers)
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
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
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',
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
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',
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):
190
189
request = IncompleteRequest(transport, write_func)
190
request.accept_bytes(bytes)
191
191
self.request = 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)
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',
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)
225
def test_protocol_version_detection_two(self):
226
# SmartWSGIApp detects requests that start with REQUEST_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',
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.
242
protocol.RESPONSE_VERSION_TWO + 'success\nok\x012\n', response)
208
245
class FakeRequest(object):