1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
|
# 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
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')
app = self.create_simple_app()
with mock.patch('oslo_messaging.notify._impl_log.LogDriver.notify',
side_effect=Exception('error')) as driver:
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)
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')
app = self.create_simple_app()
with mock.patch('oslo_messaging.notify._impl_log.LogDriver.notify',
side_effect=Exception('error')) as driver:
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)
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')
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
app.get('/foo/bar', extra_environ=self.get_environ_header())
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')
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.
app.get('/foo/bar', extra_environ=self.get_environ_header())
self.assertTrue(driver.called)
@mock.patch('oslo_messaging.get_notification_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')
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'])
def test_do_not_use_oslo_messaging(self):
self.cfg.config(use_oslo_messaging=False,
group='audit_middleware_notifications')
audit_middleware = self.create_simple_middleware()
# make sure it is using a local notifier instead of oslo_messaging
self.assertTrue(
isinstance(audit_middleware._notifier,
audit._notifier._LogNotifier))
|
