e744501c47
Currently application that doesn't use the global configuration object have to rely on hack to setup the global oslo config object for each middleware it want to use. For example, gnocchi have its own middleware loader and add crap to load keystonemiddleware: https://github.com/openstack/gnocchi/blob/master/gnocchi/rest/app.py#L140 And it can't use oslo.middleware that relies on the global conf object. Also aodh (use 'paste' for middleware) have to hack the global configuration object for each middlewares it want to use by code... https://review.openstack.org/#/c/208632/1/aodh/service.py But middleware are optional deployer stuffs, we should not write any code for them... This change allows application to use paste-deploy (or any middleware loader) without enforcing the application to use the global oslo.config object. If the middleware want to use oslo.config it should load the configuration file himself (and fallback to the global one if any) The proposed paste configuration to allow this is: [filter:cors] paste.filter_factory = oslo.middleware:cors oslo_config_project = aodh So the cors middleware can find and load the aodh config and what is it interested in. Also, some of them use oslo.config local, some other the global object. Some can be loaded by an middleware loader like paste, some other not. This change make consistent the way we bootstrap all middlewares. Closes-bug: #1482086 Change-Id: Iad197d1f3a386683d818b59718df34e14e15ca5c
109 lines
3.7 KiB
Python
109 lines
3.7 KiB
Python
# Copyright (c) 2012 Red Hat, Inc.
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License"); you may
|
|
# not use this file except in compliance with the License. You may obtain
|
|
# a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
|
|
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
|
|
# License for the specific language governing permissions and limitations
|
|
# under the License.
|
|
|
|
from oslotest import base as test_base
|
|
import six
|
|
import webob
|
|
|
|
from oslo.middleware import sizelimit
|
|
from oslo_config import fixture as config
|
|
|
|
|
|
class TestLimitingReader(test_base.BaseTestCase):
|
|
|
|
def test_limiting_reader(self):
|
|
BYTES = 1024
|
|
bytes_read = 0
|
|
data = six.StringIO("*" * BYTES)
|
|
for chunk in sizelimit.LimitingReader(data, BYTES):
|
|
bytes_read += len(chunk)
|
|
|
|
self.assertEqual(bytes_read, BYTES)
|
|
|
|
bytes_read = 0
|
|
data = six.StringIO("*" * BYTES)
|
|
reader = sizelimit.LimitingReader(data, BYTES)
|
|
byte = reader.read(1)
|
|
while len(byte) != 0:
|
|
bytes_read += 1
|
|
byte = reader.read(1)
|
|
|
|
self.assertEqual(bytes_read, BYTES)
|
|
|
|
def test_read_default_value(self):
|
|
BYTES = 1024
|
|
data_str = "*" * BYTES
|
|
data = six.StringIO(data_str)
|
|
reader = sizelimit.LimitingReader(data, BYTES)
|
|
res = reader.read()
|
|
self.assertEqual(data_str, res)
|
|
|
|
def test_limiting_reader_fails(self):
|
|
BYTES = 1024
|
|
|
|
def _consume_all_iter():
|
|
bytes_read = 0
|
|
data = six.StringIO("*" * BYTES)
|
|
for chunk in sizelimit.LimitingReader(data, BYTES - 1):
|
|
bytes_read += len(chunk)
|
|
|
|
self.assertRaises(webob.exc.HTTPRequestEntityTooLarge,
|
|
_consume_all_iter)
|
|
|
|
def _consume_all_read():
|
|
bytes_read = 0
|
|
data = six.StringIO("*" * BYTES)
|
|
reader = sizelimit.LimitingReader(data, BYTES - 1)
|
|
byte = reader.read(1)
|
|
while len(byte) != 0:
|
|
bytes_read += 1
|
|
byte = reader.read(1)
|
|
|
|
self.assertRaises(webob.exc.HTTPRequestEntityTooLarge,
|
|
_consume_all_read)
|
|
|
|
|
|
class TestRequestBodySizeLimiter(test_base.BaseTestCase):
|
|
|
|
def setUp(self):
|
|
super(TestRequestBodySizeLimiter, self).setUp()
|
|
self.useFixture(config.Config())
|
|
|
|
@webob.dec.wsgify()
|
|
def fake_app(req):
|
|
return webob.Response(req.body)
|
|
|
|
self.middleware = sizelimit.RequestBodySizeLimiter(fake_app)
|
|
self.MAX_REQUEST_BODY_SIZE = (
|
|
self.middleware.oslo_conf.oslo_middleware.max_request_body_size)
|
|
self.request = webob.Request.blank('/', method='POST')
|
|
|
|
def test_content_length_acceptable(self):
|
|
self.request.headers['Content-Length'] = self.MAX_REQUEST_BODY_SIZE
|
|
self.request.body = b"0" * self.MAX_REQUEST_BODY_SIZE
|
|
response = self.request.get_response(self.middleware)
|
|
self.assertEqual(response.status_int, 200)
|
|
|
|
def test_content_length_too_large(self):
|
|
self.request.headers['Content-Length'] = self.MAX_REQUEST_BODY_SIZE + 1
|
|
self.request.body = b"0" * (self.MAX_REQUEST_BODY_SIZE + 1)
|
|
response = self.request.get_response(self.middleware)
|
|
self.assertEqual(response.status_int, 413)
|
|
|
|
def test_request_too_large_no_content_length(self):
|
|
self.request.body = b"0" * (self.MAX_REQUEST_BODY_SIZE + 1)
|
|
self.request.headers['Content-Length'] = None
|
|
response = self.request.get_response(self.middleware)
|
|
self.assertEqual(response.status_int, 413)
|