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
This commit is contained in:
Jamie Lennox
Steve Martinelli
parent
2892744d5b
commit
23808e1bae
@ -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'])
|
||||
|
||||
124
keystonemiddleware/tests/unit/audit/test_audit_oslo_messaging.py
Normal file
124
keystonemiddleware/tests/unit/audit/test_audit_oslo_messaging.py
Normal file
@ -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'])
|
||||
Loading…
x
Reference in New Issue
Block a user