openstack/keystonemiddleware
Code Issues Proposed changes

Refactor audit tests to use create_middleware

Browse Source 
The auth_token tests have a good layout for using create_middleware and
create_simple_middleware that makes it easy for tests to create
middleware for testing as required rather than having a self.middleware
object that gets overriden a lot.

Extract this create_middleware into a base class and implement it in
audit so the patterns are similar.

Change-Id: I2f050eef1684c8046f94dc2b88b4c97a56ea9cd8
This commit is contained in:
Jamie Lennox 2016-07-04 11:31:28 +10:00 committed by Steve Martinelli
parent 98b639aeb7
commit 6a5ef484f2
7 changed files with 72 additions and 98 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -14,6 +14,7 @@
from oslo_config import fixture as cfg_fixture
from oslo_messaging import conffixture as msg_fixture
from oslotest import createfile
import webob.dec
from keystonemiddleware import audit
from keystonemiddleware.tests.unit import utils
@ -36,22 +37,7 @@ compute = service/compute
"""
class FakeApp(object):
def __call__(self, env, start_response):
body = 'Some response'
start_response('200 OK', [
('Content-Type', 'text/plain'),
('Content-Length', str(sum(map(len, body))))
])
return [body]
class FakeFailingApp(object):
def __call__(self, env, start_response):
raise Exception('It happens!')
class BaseAuditMiddlewareTest(utils.BaseTestCase):
class BaseAuditMiddlewareTest(utils.MiddlewareTestCase):
PROJECT_NAME = 'keystonemiddleware'
def setUp(self):
@ -65,9 +51,16 @@ class BaseAuditMiddlewareTest(utils.BaseTestCase):
self.cfg.conf([], project=self.PROJECT_NAME)
self.middleware = audit.AuditMiddleware(
FakeApp(), audit_map_file=self.audit_map,
service_name='pycadf')
def create_middleware(self, cb, **kwargs):
@webob.dec.wsgify
def _do_cb(req):
return cb(req)
kwargs.setdefault('audit_map_file', self.audit_map)
kwargs.setdefault('service_name', 'pycadf')
return audit.AuditMiddleware(_do_cb, **kwargs)
@property
def audit_map(self):

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

@ -201,9 +201,10 @@ class AuditApiLogicTest(base.BaseAuditMiddlewareTest):
environ=self.get_environ_header('GET'),
remote_addr='192.168.0.1')
req.context = {}
self.middleware._process_request(req)
middleware = self.create_simple_middleware()
middleware._process_request(req)
payload = req.environ['cadf_event'].as_dict()
self.middleware._process_response(req, webob.Response())
middleware._process_response(req, webob.Response())
payload2 = req.environ['cadf_event'].as_dict()
self.assertEqual(payload['id'], payload2['id'])
self.assertEqual(payload['tags'], payload2['tags'])

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

@ -17,7 +17,6 @@ import fixtures
import mock
import webob
from keystonemiddleware import audit
from keystonemiddleware.tests.unit.audit import base
@ -36,7 +35,7 @@ class AuditMiddlewareTest(base.BaseAuditMiddlewareTest):
req = webob.Request.blank('/foo/bar',
environ=self.get_environ_header('GET'))
self.middleware(req)
self.create_simple_middleware()(req)
# Check first notification with only 'request'
call_args = self.notifier.notify.call_args_list[0][0]
@ -55,16 +54,17 @@ class AuditMiddlewareTest(base.BaseAuditMiddlewareTest):
self.assertIn('reporterchain', call_args[2])
def test_api_request_failure(self):
self.middleware = audit.AuditMiddleware(
base.FakeFailingApp(),
audit_map_file=self.audit_map,
service_name='pycadf')
def cb(self, req):
raise Exception('It happens!')
middleware = self.create_middleware(cb)
req = webob.Request.blank('/foo/bar',
environ=self.get_environ_header('GET'))
try:
self.middleware(req)
middleware(req)
self.fail('Application exception has not been re-raised')
except Exception:
pass
@ -88,7 +88,7 @@ class AuditMiddlewareTest(base.BaseAuditMiddlewareTest):
environ=self.get_environ_header('GET'))
req.context = {}
self.middleware._process_request(req)
self.create_simple_middleware()._process_request(req)
self.assertTrue(self.notifier.notify.called)
def test_process_response_fail(self):
@ -96,13 +96,13 @@ class AuditMiddlewareTest(base.BaseAuditMiddlewareTest):
environ=self.get_environ_header('GET'))
req.context = {}
self.middleware._process_response(req, webob.response.Response())
middleware = self.create_simple_middleware()
middleware._process_response(req, webob.response.Response())
self.assertTrue(self.notifier.notify.called)
def test_ignore_req_opt(self):
self.middleware = audit.AuditMiddleware(base.FakeApp(),
audit_map_file=self.audit_map,
ignore_req_list='get, PUT')
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',
@ -111,12 +111,12 @@ class AuditMiddlewareTest(base.BaseAuditMiddlewareTest):
environ=self.get_environ_header('POST'))
# Check GET/PUT request does not send notification
self.middleware(req)
self.middleware(req1)
middleware(req)
middleware(req1)
self.assertFalse(self.notifier.notify.called)
# Check non-GET/PUT request does send notification
self.middleware(req2)
middleware(req2)
self.assertEqual(2, self.notifier.notify.call_count)
call_args = self.notifier.notify.call_args_list[0][0]
@ -128,14 +128,10 @@ class AuditMiddlewareTest(base.BaseAuditMiddlewareTest):
self.assertEqual('/accept/foo', call_args[2]['requestPath'])
def test_cadf_event_context_scoped(self):
middleware = audit.AuditMiddleware(
base.FakeApp(),
audit_map_file=self.audit_map,
service_name='pycadf')
req = webob.Request.blank('/foo/bar',
environ=self.get_environ_header('GET'))
middleware(req)
self.create_simple_middleware()(req)
self.assertEqual(2, self.notifier.notify.call_count)
first, second = [a[0] for a in self.notifier.notify.call_args_list]
@ -147,13 +143,9 @@ class AuditMiddlewareTest(base.BaseAuditMiddlewareTest):
self.assertIs(first[0], second[0])
def test_cadf_event_scoped_to_request(self):
middleware = audit.AuditMiddleware(
base.FakeApp(),
audit_map_file=self.audit_map,
service_name='pycadf')
req = webob.Request.blank('/foo/bar',
environ=self.get_environ_header('GET'))
middleware(req)
self.create_simple_middleware()(req)
self.assertIsNotNone(req.environ.get('cadf_event'))
# ensure exact same event is used between request and response
@ -161,17 +153,14 @@ class AuditMiddlewareTest(base.BaseAuditMiddlewareTest):
self.notifier.calls[1].payload['id'])
def test_cadf_event_scoped_to_request_on_error(self):
middleware = audit.AuditMiddleware(
base.FakeApp(),
audit_map_file=self.audit_map,
service_name='pycadf')
middleware = self.create_simple_middleware()
req = webob.Request.blank('/foo/bar',
environ=self.get_environ_header('GET'))
req.context = {}
self.notifier.notify.side_effect = Exception('error')
middleware._process_request(req)
middleware(req)
self.assertTrue(self.notifier.notify.called)
req2 = webob.Request.blank('/foo/bar',
@ -187,24 +176,23 @@ class AuditMiddlewareTest(base.BaseAuditMiddlewareTest):
def test_project_name_from_oslo_config(self):
self.assertEqual(self.PROJECT_NAME,
self.middleware._conf.project)
self.create_simple_middleware()._conf.project)
def test_project_name_from_local_config(self):
project_name = uuid.uuid4().hex
self.middleware = audit.AuditMiddleware(
base.FakeApp(), audit_map_file=self.audit_map,
service_name='pycadf', project=project_name)
self.assertEqual(project_name, self.middleware._conf.project)
middleware = self.create_simple_middleware(project=project_name)
self.assertEqual(project_name, middleware._conf.project)
def test_no_response(self):
middleware = self.create_simple_middleware()
url = 'http://admin_host:8774/v2/' + str(uuid.uuid4()) + '/servers'
req = webob.Request.blank(url,
environ=self.get_environ_header('GET'),
remote_addr='192.168.0.1')
req.context = {}
self.middleware._process_request(req)
middleware._process_request(req)
payload = req.environ['cadf_event'].as_dict()
self.middleware._process_response(req, None)
middleware._process_response(req, None)
payload2 = req.environ['cadf_event'].as_dict()
self.assertEqual(payload['id'], payload2['id'])
self.assertEqual(payload['tags'], payload2['tags'])
@ -221,7 +209,9 @@ class AuditMiddlewareTest(base.BaseAuditMiddlewareTest):
environ=self.get_environ_header('GET'))
req.context = {}
self.assertNotIn('cadf_event', req.environ)
self.middleware._process_response(req, webob.Response())
self.create_simple_middleware()._process_response(req,
webob.Response())
self.assertIn('cadf_event', req.environ)
payload = req.environ['cadf_event'].as_dict()
self.assertEqual(payload['outcome'], 'success')

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

@ -13,7 +13,6 @@
import mock
import webob
from keystonemiddleware import audit
from keystonemiddleware.tests.unit.audit import base
@ -21,10 +20,7 @@ class AuditNotifierConfigTest(base.BaseAuditMiddlewareTest):
def test_conf_middleware_log_and_default_as_messaging(self):
self.cfg.config(driver='log', group='audit_middleware_notifications')
middleware = audit.AuditMiddleware(
base.FakeApp(),
audit_map_file=self.audit_map,
service_name='pycadf')
middleware = self.create_simple_middleware()
req = webob.Request.blank('/foo/bar',
environ=self.get_environ_header('GET'))
req.context = {}
@ -41,10 +37,7 @@ class AuditNotifierConfigTest(base.BaseAuditMiddlewareTest):
self.cfg.config(driver='log',
group='audit_middleware_notifications')
middleware = audit.AuditMiddleware(
base.FakeApp(),
audit_map_file=self.audit_map,
service_name='pycadf')
middleware = self.create_simple_middleware()
req = webob.Request.blank('/foo/bar',
environ=self.get_environ_header('GET'))
req.context = {}
@ -59,10 +52,7 @@ class AuditNotifierConfigTest(base.BaseAuditMiddlewareTest):
self.cfg.config(driver='log', group='oslo_messaging_notifications')
self.cfg.config(driver='messaging',
group='audit_middleware_notifications')
middleware = audit.AuditMiddleware(
base.FakeApp(),
audit_map_file=self.audit_map,
service_name='pycadf')
middleware = self.create_simple_middleware()
req = webob.Request.blank('/foo/bar',
environ=self.get_environ_header('GET'))
req.context = {}
@ -79,10 +69,7 @@ class AuditNotifierConfigTest(base.BaseAuditMiddlewareTest):
group='oslo_messaging_notifications')
self.cfg.config(driver=None, group='audit_middleware_notifications')
middleware = audit.AuditMiddleware(
base.FakeApp(),
audit_map_file=self.audit_map,
service_name='pycadf')
middleware = self.create_simple_middleware()
req = webob.Request.blank('/foo/bar',
environ=self.get_environ_header('GET'))
req.context = {}
@ -101,10 +88,7 @@ class AuditNotifierConfigTest(base.BaseAuditMiddlewareTest):
transport_url=transport_url,
group='audit_middleware_notifications')
audit.AuditMiddleware(
base.FakeApp(),
audit_map_file=self.audit_map,
service_name='pycadf')
self.create_simple_middleware()
self.assertTrue(m.called)
# make sure first call kwarg 'url' is same as provided transport_url
self.assertEqual(transport_url, m.call_args_list[0][1]['url'])

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

@ -14,7 +14,6 @@ import fixtures
import mock
import webob
from keystonemiddleware import audit
from keystonemiddleware.tests.unit.audit import base
@ -29,8 +28,7 @@ class TestLoggingNotifier(base.BaseAuditMiddlewareTest):
@mock.patch('keystonemiddleware.audit._LOG.info')
def test_api_request_no_messaging(self, log):
middleware = audit.AuditMiddleware(base.FakeApp(),
audit_map_file=self.audit_map)
middleware = self.create_simple_middleware()
req = webob.Request.blank('/foo/bar',
environ=self.get_environ_header('GET'))

14
keystonemiddleware/tests/unit/auth_token/base.py
View File

@ -24,7 +24,7 @@ from keystonemiddleware import auth_token
from keystonemiddleware.tests.unit import utils
class BaseAuthTokenTestCase(utils.BaseTestCase):
class BaseAuthTokenTestCase(utils.MiddlewareTestCase):
def setUp(self):
super(BaseAuthTokenTestCase, self).setUp()
@ -49,18 +49,6 @@ class BaseAuthTokenTestCase(utils.BaseTestCase):
return auth_token.AuthProtocol(_do_cb, opts)
def create_simple_middleware(self,
status='200 OK',
body='',
headers=None,
**kwargs):
def cb(req):
resp = webob.Response(body, status)
resp.headers.update(headers or {})
return resp
return self.create_middleware(cb, **kwargs)
def call(self, middleware, method='GET', path='/', headers=None,
expected_status=http_client.OK):
req = webob.Request.blank(path)

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

@ -13,13 +13,14 @@
import logging
import sys
import time
import uuid
import warnings
import fixtures
import mock
import oslotest.base as oslotest
import requests
import uuid
import webob
class BaseTestCase(oslotest.BaseTestCase):
@ -32,6 +33,8 @@ class BaseTestCase(oslotest.BaseTestCase):
module='^keystonemiddleware\\.')
self.addCleanup(warnings.resetwarnings)
self.logger = self.useFixture(fixtures.FakeLogger(level=logging.DEBUG))
class TestCase(BaseTestCase):
TEST_DOMAIN_ID = '1'
@ -49,7 +52,6 @@ class TestCase(BaseTestCase):
def setUp(self):
super(TestCase, self).setUp()
self.logger = self.useFixture(fixtures.FakeLogger(level=logging.DEBUG))
self.time_patcher = mock.patch.object(time, 'time', lambda: 1234)
self.time_patcher.start()
@ -76,6 +78,24 @@ if tuple(sys.version_info)[0:2] < (2, 7):
TestCase.assertDictEqual = assertDictEqual
class MiddlewareTestCase(BaseTestCase):
def create_middleware(self, cb, **kwargs):
raise NotImplemented("implement this in your tests")
def create_simple_middleware(self,
status='200 OK',
body='',
headers=None,
**kwargs):
def cb(req):
resp = webob.Response(body, status)
resp.headers.update(headers or {})
return resp
return self.create_middleware(cb, **kwargs)
class TestResponse(requests.Response):
"""Utility class to wrap requests.Response.