71
72
def test_construct(self):
72
73
app = wsgi.SmartWSGIApp(FakeTransport())
73
74
self.assertIsInstance(
74
app.backing_transport, chroot.ChrootTransportDecorator)
75
app.backing_transport, chroot.ChrootTransport)
76
77
def test_http_get_rejected(self):
77
78
# GET requests are rejected.
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
# WSGI environ to construct the transport for this request, by cloning
88
# its base transport with the given relpath.
94
# WSGI environ to clone from its backing transport to get a specific
95
# transport for this request.
89
96
transport = FakeTransport()
90
97
wsgi_app = wsgi.SmartWSGIApp(transport)
91
def make_request(transport, write_func):
92
request = FakeRequest(transport, write_func)
93
self.request = request
95
wsgi_app.make_request = make_request
98
wsgi_app.backing_transport = transport
99
wsgi_app.make_request = self._fake_make_request
96
100
fake_input = StringIO('fake request')
97
101
environ = self.build_environ({
98
102
'REQUEST_METHOD': 'POST',
111
115
transport = memory.MemoryTransport()
112
116
transport.put_bytes('foo', 'some bytes')
113
117
wsgi_app = wsgi.SmartWSGIApp(transport)
114
def make_request(transport, write_func):
115
request = FakeRequest(transport, write_func)
116
self.request = request
118
wsgi_app.make_request = make_request
118
wsgi_app.make_request = self._fake_make_request
119
119
fake_input = StringIO('fake request')
120
120
environ = self.build_environ({
121
121
'REQUEST_METHOD': 'POST',
175
175
path_var='a path_var')
176
176
self.assertIsInstance(app, wsgi.RelpathSetter)
177
177
self.assertIsInstance(app.app, wsgi.SmartWSGIApp)
178
self.assertEndsWith(app.app.backing_transport.base, 'a%20root/')
178
self.assertStartsWith(app.app.backing_transport.base, 'chroot-')
179
backing_transport = app.app.backing_transport
180
chroot_backing_transport = backing_transport.server.backing_transport
181
self.assertEndsWith(chroot_backing_transport.base, 'a%20root/')
179
182
self.assertEqual(app.prefix, 'a prefix')
180
183
self.assertEqual(app.path_var, 'a path_var')
182
185
def test_incomplete_request(self):
183
186
transport = FakeTransport()
184
187
wsgi_app = wsgi.SmartWSGIApp(transport)
185
def make_request(transport, write_func):
188
def make_request(transport, write_func, bytes):
186
189
request = IncompleteRequest(transport, write_func)
190
request.accept_bytes(bytes)
187
191
self.request = request
189
193
wsgi_app.make_request = make_request
200
204
self.assertEqual('200 OK', self.status)
201
205
self.assertEqual('error\x01incomplete request\n', response)
203
def test_chrooting(self):
204
# Show that requests that try to access things outside of the base
205
# really will get intercepted by the ChrootTransportDecorator.
207
def test_protocol_version_detection_one(self):
208
# SmartWSGIApp detects requests that don't start with
209
# REQUEST_VERSION_TWO as version one.
206
210
transport = memory.MemoryTransport()
207
transport.mkdir('foo')
208
transport.put_bytes('foo/bar', 'this is foo/bar')
209
wsgi_app = wsgi.SmartWSGIApp(transport.clone('foo'))
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)
211
smart_request = StringIO('mkdir\x01/bad file\x01\n0\ndone\n')
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')
212
231
environ = self.build_environ({
213
232
'REQUEST_METHOD': 'POST',
214
'CONTENT_LENGTH': len(smart_request.getvalue()),
215
'wsgi.input': smart_request,
216
'bzrlib.relpath': '.',
233
'CONTENT_LENGTH': len(fake_input.getvalue()),
234
'wsgi.input': fake_input,
235
'bzrlib.relpath': 'foo',
218
237
iterable = wsgi_app(environ, self.start_response)
219
238
response = self.read_response(iterable)
220
239
self.assertEqual('200 OK', self.status)
240
# Expect a version 2-encoded response.
221
241
self.assertEqual(
222
"error\x01Path '/bad file' is not a child of "
223
"path 'memory:///foo/'\n",
242
protocol.RESPONSE_VERSION_TWO + 'success\nok\x012\n', response)
227
245
class FakeRequest(object):