openstack/keystonemiddleware
Code Issues Proposed changes

Return and use an app wherever possible

Browse Source 
Audit middleware does a lot of faking up environments and calling
methods instead of just running the middleware. It's a whole bunch
easier to just run the middleware wherever possible.

This means we don't have to stub context from tests as these tests
actually pass through the wsgi layer correctly. Ideally we would do more
of this.

Change-Id: I95377f030b07ffae18698ecc3c82cc6aa1dddbc7
This commit is contained in:
Jamie Lennox 2016-07-04 12:28:28 +10:00 committed by Steve Martinelli
parent 6a5ef484f2
commit b8024ff8c6
7 changed files with 44 additions and 61 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
keystonemiddleware/tests/unit/audit/base.py
View File

@ -67,7 +67,7 @@ class BaseAuditMiddlewareTest(utils.MiddlewareTestCase):
return self.audit_map_file_fixture.path
@staticmethod
def get_environ_header(req_type):
def get_environ_header(req_type=None):
env_headers = {'HTTP_X_SERVICE_CATALOG':
'''[{"endpoints_links": [],
"endpoints": [{"adminURL":
@ -85,5 +85,6 @@ class BaseAuditMiddlewareTest(utils.MiddlewareTestCase):
'HTTP_X_AUTH_TOKEN': 'token',
'HTTP_X_PROJECT_ID': 'tenant_id',
'HTTP_X_IDENTITY_STATUS': 'Confirmed'}
env_headers['REQUEST_METHOD'] = req_type
if req_type:
env_headers['REQUEST_METHOD'] = req_type
return env_headers

3
keystonemiddleware/tests/unit/audit/test_audit_api.py
View File

@ -24,10 +24,11 @@ class AuditApiLogicTest(base.BaseAuditMiddlewareTest):
def get_payload(self, method, url,
audit_map=None, body=None, environ=None):
audit_map = audit_map or self.audit_map
environ = environ or self.get_environ_header(method)
environ = environ or self.get_environ_header()
req = webob.Request.blank(url,
body=body,
method=method,
environ=environ,
remote_addr='192.168.0.1')

52
keystonemiddleware/tests/unit/audit/test_audit_middleware.py
View File

@ -32,10 +32,8 @@ class AuditMiddlewareTest(base.BaseAuditMiddlewareTest):
super(AuditMiddlewareTest, self).setUp()
def test_api_request(self):
req = webob.Request.blank('/foo/bar',
environ=self.get_environ_header('GET'))
self.create_simple_middleware()(req)
self.create_simple_app().get('/foo/bar',
extra_environ=self.get_environ_header())
# Check first notification with only 'request'
call_args = self.notifier.notify.call_args_list[0][0]
@ -55,18 +53,18 @@ class AuditMiddlewareTest(base.BaseAuditMiddlewareTest):
def test_api_request_failure(self):
def cb(self, req):
raise Exception('It happens!')
class CustomException(Exception):
pass
middleware = self.create_middleware(cb)
req = webob.Request.blank('/foo/bar',
environ=self.get_environ_header('GET'))
def cb(req):
raise CustomException('It happens!')
try:
middleware(req)
self.create_app(cb).get('/foo/bar',
extra_environ=self.get_environ_header())
self.fail('Application exception has not been re-raised')
except Exception:
except CustomException:
pass
# Check first notification with only 'request'
@ -101,22 +99,17 @@ class AuditMiddlewareTest(base.BaseAuditMiddlewareTest):
self.assertTrue(self.notifier.notify.called)
def test_ignore_req_opt(self):
middleware = self.create_simple_middleware(ignore_req_list='get, PUT')
req = webob.Request.blank('/skip/foo',
environ=self.get_environ_header('GET'))
req1 = webob.Request.blank('/skip/foo',
environ=self.get_environ_header('PUT'))
req2 = webob.Request.blank('/accept/foo',
environ=self.get_environ_header('POST'))
app = self.create_simple_app(ignore_req_list='get, PUT')
# Check GET/PUT request does not send notification
middleware(req)
middleware(req1)
app.get('/skip/foo', extra_environ=self.get_environ_header())
app.put('/skip/foo', extra_environ=self.get_environ_header())
self.assertFalse(self.notifier.notify.called)
# Check non-GET/PUT request does send notification
middleware(req2)
app.post('/accept/foo', extra_environ=self.get_environ_header())
self.assertEqual(2, self.notifier.notify.call_count)
call_args = self.notifier.notify.call_args_list[0][0]
@ -128,10 +121,8 @@ class AuditMiddlewareTest(base.BaseAuditMiddlewareTest):
self.assertEqual('/accept/foo', call_args[2]['requestPath'])
def test_cadf_event_context_scoped(self):
req = webob.Request.blank('/foo/bar',
environ=self.get_environ_header('GET'))
self.create_simple_middleware()(req)
self.create_simple_app().get('/foo/bar',
extra_environ=self.get_environ_header())
self.assertEqual(2, self.notifier.notify.call_count)
first, second = [a[0] for a in self.notifier.notify.call_args_list]
@ -143,10 +134,9 @@ class AuditMiddlewareTest(base.BaseAuditMiddlewareTest):
self.assertIs(first[0], second[0])
def test_cadf_event_scoped_to_request(self):
req = webob.Request.blank('/foo/bar',
environ=self.get_environ_header('GET'))
self.create_simple_middleware()(req)
self.assertIsNotNone(req.environ.get('cadf_event'))
app = self.create_simple_app()
resp = app.get('/foo/bar', extra_environ=self.get_environ_header())
self.assertIsNotNone(resp.request.environ.get('cadf_event'))
# ensure exact same event is used between request and response
self.assertEqual(self.notifier.calls[0].payload['id'],

29
keystonemiddleware/tests/unit/audit/test_audit_oslo_messaging.py
View File

@ -11,7 +11,6 @@
# under the License.
import mock
import webob
from keystonemiddleware.tests.unit.audit import base
@ -20,13 +19,10 @@ class AuditNotifierConfigTest(base.BaseAuditMiddlewareTest):
def test_conf_middleware_log_and_default_as_messaging(self):
self.cfg.config(driver='log', group='audit_middleware_notifications')
middleware = self.create_simple_middleware()
req = webob.Request.blank('/foo/bar',
environ=self.get_environ_header('GET'))
req.context = {}
app = self.create_simple_app()
with mock.patch('oslo_messaging.notify._impl_log.LogDriver.notify',
side_effect=Exception('error')) as driver:
middleware._process_request(req)
app.get('/foo/bar', extra_environ=self.get_environ_header())
# audit middleware conf has 'log' make sure that driver is invoked
# and not the one specified in DEFAULT section
self.assertTrue(driver.called)
@ -37,13 +33,10 @@ class AuditNotifierConfigTest(base.BaseAuditMiddlewareTest):
self.cfg.config(driver='log',
group='audit_middleware_notifications')
middleware = self.create_simple_middleware()
req = webob.Request.blank('/foo/bar',
environ=self.get_environ_header('GET'))
req.context = {}
app = self.create_simple_app()
with mock.patch('oslo_messaging.notify._impl_log.LogDriver.notify',
side_effect=Exception('error')) as driver:
middleware._process_request(req)
app.get('/foo/bar', extra_environ=self.get_environ_header())
# audit middleware conf has 'log' make sure that driver is invoked
# and not the one specified in oslo_messaging_notifications section
self.assertTrue(driver.called)
@ -52,16 +45,13 @@ class AuditNotifierConfigTest(base.BaseAuditMiddlewareTest):
self.cfg.config(driver='log', group='oslo_messaging_notifications')
self.cfg.config(driver='messaging',
group='audit_middleware_notifications')
middleware = self.create_simple_middleware()
req = webob.Request.blank('/foo/bar',
environ=self.get_environ_header('GET'))
req.context = {}
app = self.create_simple_app()
with mock.patch('oslo_messaging.notify.messaging.MessagingDriver'
'.notify',
side_effect=Exception('error')) as driver:
# audit middleware has 'messaging' make sure that driver is invoked
# and not the one specified in oslo_messaging_notifications section
middleware._process_request(req)
app.get('/foo/bar', extra_environ=self.get_environ_header())
self.assertTrue(driver.called)
def test_with_no_middleware_notification_conf(self):
@ -69,16 +59,13 @@ class AuditNotifierConfigTest(base.BaseAuditMiddlewareTest):
group='oslo_messaging_notifications')
self.cfg.config(driver=None, group='audit_middleware_notifications')
middleware = self.create_simple_middleware()
req = webob.Request.blank('/foo/bar',
environ=self.get_environ_header('GET'))
req.context = {}
app = self.create_simple_app()
with mock.patch('oslo_messaging.notify.messaging.MessagingDriver'
'.notify',
side_effect=Exception('error')) as driver:
# audit middleware section is not set. So driver needs to be
# invoked from oslo_messaging_notifications section.
middleware._process_request(req)
app.get('/foo/bar', extra_environ=self.get_environ_header())
self.assertTrue(driver.called)
@mock.patch('oslo_messaging.get_transport')

8
keystonemiddleware/tests/unit/audit/test_logging_notifier.py
View File

@ -12,7 +12,6 @@
import fixtures
import mock
import webob
from keystonemiddleware.tests.unit.audit import base
@ -28,11 +27,8 @@ class TestLoggingNotifier(base.BaseAuditMiddlewareTest):
@mock.patch('keystonemiddleware.audit._LOG.info')
def test_api_request_no_messaging(self, log):
middleware = self.create_simple_middleware()
req = webob.Request.blank('/foo/bar',
environ=self.get_environ_header('GET'))
req.get_response(middleware)
self.create_simple_app().get('/foo/bar',
extra_environ=self.get_environ_header())
# Check first notification with only 'request'
call_args = log.call_args_list[0][0]

7
keystonemiddleware/tests/unit/utils.py
View File

@ -21,6 +21,7 @@ import mock
import oslotest.base as oslotest
import requests
import webob
import webtest
class BaseTestCase(oslotest.BaseTestCase):
@ -95,6 +96,12 @@ class MiddlewareTestCase(BaseTestCase):
return self.create_middleware(cb, **kwargs)
def create_app(self, *args, **kwargs):
return webtest.TestApp(self.create_middleware(*args, **kwargs))
def create_simple_app(self, *args, **kwargs):
return webtest.TestApp(self.create_simple_middleware(*args, **kwargs))
class TestResponse(requests.Response):
"""Utility class to wrap requests.Response.

1
test-requirements.txt
View File

@ -19,6 +19,7 @@ testrepository>=0.0.18 # Apache-2.0/BSD
testresources>=0.2.4 # Apache-2.0/BSD
testtools>=1.4.0 # MIT
python-memcached>=1.56 # PSF
WebTest>=2.0 # MIT
# Bandit security code scanner
bandit>=1.1.0 # Apache-2.0