From 23808e1bae2866dc8ffc3be6fbacd94fc25f9878 Mon Sep 17 00:00:00 2001 From: Jamie Lennox Date: Mon, 27 Jun 2016 13:52:02 +1000 Subject: [PATCH] Extract oslo_messaging specific audit tests Extract the whole class of tests that deals with oslo.messaging notifications. This matches what we did with logging notifications. Convert the tests to using mock fixtures rather than mock decorator directly. Change-Id: I805be038c1e4e1d2fe16328c44cfa03d793b0136 --- .../tests/unit/audit/test_audit_middleware.py | 100 -------------- .../unit/audit/test_audit_oslo_messaging.py | 124 ++++++++++++++++++ 2 files changed, 124 insertions(+), 100 deletions(-) create mode 100644 keystonemiddleware/tests/unit/audit/test_audit_oslo_messaging.py diff --git a/keystonemiddleware/tests/unit/audit/test_audit_middleware.py b/keystonemiddleware/tests/unit/audit/test_audit_middleware.py index c3c307da..1bbdc566 100644 --- a/keystonemiddleware/tests/unit/audit/test_audit_middleware.py +++ b/keystonemiddleware/tests/unit/audit/test_audit_middleware.py @@ -228,103 +228,3 @@ class AuditMiddlewareTest(base.BaseAuditMiddlewareTest): self.assertEqual(payload['reason']['reasonType'], 'HTTP') self.assertEqual(payload['reason']['reasonCode'], '200') self.assertEqual(payload['observer']['id'], 'target') - - -def _get_transport(conf, aliases=None, url=None): - transport = mock.MagicMock() - transport.conf = conf - conf.register_opts = mock.MagicMock() - return transport - - -@mock.patch('oslo_messaging.get_transport', side_effect=_get_transport) -class AuditNotifierConfigTest(base.BaseAuditMiddlewareTest): - - def test_conf_middleware_log_and_default_as_messaging(self, t): - self.cfg.config(driver='log', group='audit_middleware_notifications') - 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')) - req.context = {} - with mock.patch('oslo_messaging.notify._impl_log.LogDriver.notify', - side_effect=Exception('error')) as driver: - middleware._process_request(req) - # audit middleware conf has 'log' make sure that driver is invoked - # and not the one specified in DEFAULT section - self.assertTrue(driver.called) - - def test_conf_middleware_log_and_oslo_msg_as_messaging(self, t): - self.cfg.config(driver='messaging', - group='oslo_messaging_notifications') - self.cfg.config(driver='log', - group='audit_middleware_notifications') - 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')) - req.context = {} - with mock.patch('oslo_messaging.notify._impl_log.LogDriver.notify', - side_effect=Exception('error')) as driver: - middleware._process_request(req) - # 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) - - def test_conf_middleware_messaging_and_oslo_msg_as_log(self, t): - 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') - req = webob.Request.blank('/foo/bar', - environ=self.get_environ_header('GET')) - req.context = {} - 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) - self.assertTrue(driver.called) - - def test_with_no_middleware_notification_conf(self, t): - self.cfg.config(driver='messaging', - 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') - req = webob.Request.blank('/foo/bar', - environ=self.get_environ_header('GET')) - req.context = {} - 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) - self.assertTrue(driver.called) - - def test_conf_middleware_messaging_and_transport_set(self, mock_transport): - transport_url = 'rabbit://me:passwd@host:5672/virtual_host' - self.cfg.config(driver='messaging', - transport_url=transport_url, - group='audit_middleware_notifications') - - audit.AuditMiddleware( - base.FakeApp(), - audit_map_file=self.audit_map, - service_name='pycadf') - self.assertTrue(mock_transport.called) - # make sure first call kwarg 'url' is same as provided transport_url - self.assertEqual(transport_url, - mock_transport.call_args_list[0][1]['url']) diff --git a/keystonemiddleware/tests/unit/audit/test_audit_oslo_messaging.py b/keystonemiddleware/tests/unit/audit/test_audit_oslo_messaging.py new file mode 100644 index 00000000..3b3f0796 --- /dev/null +++ b/keystonemiddleware/tests/unit/audit/test_audit_oslo_messaging.py @@ -0,0 +1,124 @@ +# 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. + +import fixtures +import mock +import webob + +from keystonemiddleware import audit +from keystonemiddleware.tests.unit.audit import base + + +class AuditNotifierConfigTest(base.BaseAuditMiddlewareTest): + + def setUp(self): + super(AuditNotifierConfigTest, self).setUp() + + f = fixtures.MockPatch('oslo_messaging.get_transport', + side_effect=self._get_transport) + self.transport_fixture = self.useFixture(f) + + def _get_transport(self, conf, aliases=None, url=None): + transport = mock.MagicMock() + transport.conf = conf + conf.register_opts = mock.MagicMock() + return transport + + 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') + req = webob.Request.blank('/foo/bar', + environ=self.get_environ_header('GET')) + req.context = {} + with mock.patch('oslo_messaging.notify._impl_log.LogDriver.notify', + side_effect=Exception('error')) as driver: + middleware._process_request(req) + # audit middleware conf has 'log' make sure that driver is invoked + # and not the one specified in DEFAULT section + self.assertTrue(driver.called) + + def test_conf_middleware_log_and_oslo_msg_as_messaging(self): + self.cfg.config(driver='messaging', + group='oslo_messaging_notifications') + self.cfg.config(driver='log', + group='audit_middleware_notifications') + + 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')) + req.context = {} + with mock.patch('oslo_messaging.notify._impl_log.LogDriver.notify', + side_effect=Exception('error')) as driver: + middleware._process_request(req) + # 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) + + def test_conf_middleware_messaging_and_oslo_msg_as_log(self): + 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') + req = webob.Request.blank('/foo/bar', + environ=self.get_environ_header('GET')) + req.context = {} + 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) + self.assertTrue(driver.called) + + def test_with_no_middleware_notification_conf(self): + self.cfg.config(driver='messaging', + 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') + req = webob.Request.blank('/foo/bar', + environ=self.get_environ_header('GET')) + req.context = {} + 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) + self.assertTrue(driver.called) + + def test_conf_middleware_messaging_and_transport_set(self): + transport_url = 'rabbit://me:passwd@host:5672/virtual_host' + self.cfg.config(driver='messaging', + transport_url=transport_url, + group='audit_middleware_notifications') + + audit.AuditMiddleware( + base.FakeApp(), + audit_map_file=self.audit_map, + service_name='pycadf') + m = self.transport_fixture.mock + 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'])