19
19
from cStringIO import StringIO
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
83
82
self.assertEqual('405 Method not allowed', self.status)
84
83
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
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
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
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
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)
245
208
class FakeRequest(object):