98b639aeb7
Oslo_messaging mocking in audit tests is a bit odd because middleware creation must be run multiple times so that the oslo_messaging opts are registered before the config fixture tries to set it. If we use the oslo_messaging conf fixture then it handles all the option registration for us correctly and removes the need for us to manually patch get_transport. Change-Id: Iaea2e66e9418a0426d73e1c4554c56be262a7779
111 lines
4.9 KiB
Python
111 lines
4.9 KiB
Python
# 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 mock
|
|
import webob
|
|
|
|
from keystonemiddleware import audit
|
|
from keystonemiddleware.tests.unit.audit import base
|
|
|
|
|
|
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')
|
|
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)
|
|
|
|
@mock.patch('oslo_messaging.get_transport')
|
|
def test_conf_middleware_messaging_and_transport_set(self, m):
|
|
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(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'])
|