← Back to branch summary

~ubuntu-branches/ubuntu/quantal/nova/quantal-proposed

« back to all changes in this revision

Viewing changes to nova/tests/test_notifier.py

  • Committer: Package Import Robot
  • Author(s): Chuck Short
  • Date: 2012-08-16 14:04:11 UTC
  • mto: This revision was merged to the branch mainline in revision 84.
  • Revision ID: package-import@ubuntu.com-20120816140411-0mr4n241wmk30t9l
Tags: upstream-2012.2~f3
Import upstream version 2012.2~f3

Show diffs side-by-side

added added

removed removed

Lines of Context:
nova/tests/test_notifier.py
1
 
# Copyright 2011 OpenStack LLC.
2
 
# All Rights Reserved.
3
 
#
4
 
#    Licensed under the Apache License, Version 2.0 (the "License"); you may
5
 
#    not use this file except in compliance with the License. You may obtain
6
 
#    a copy of the License at
7
 
#
8
 
#         http://www.apache.org/licenses/LICENSE-2.0
9
 
#
10
 
#    Unless required by applicable law or agreed to in writing, software
11
 
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
12
 
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
13
 
#    License for the specific language governing permissions and limitations
14
 
#    under the License.
15
 
 
16
 
import nova
17
 
from nova import context
18
 
from nova import flags
19
 
from nova.notifier import api as notifier_api
20
 
import nova.notifier.no_op_notifier
21
 
from nova.openstack.common import log
22
 
from nova import test
23
 
 
24
 
 
25
 
ctxt = context.get_admin_context()
26
 
ctxt2 = context.get_admin_context()
27
 
 
28
 
 
29
 
class NotifierTestCase(test.TestCase):
30
 
    """Test case for notifications"""
31
 
    def setUp(self):
32
 
        super(NotifierTestCase, self).setUp()
33
 
        self.flags(notification_driver='nova.notifier.no_op_notifier')
34
 
 
35
 
    def test_send_notification(self):
36
 
        self.notify_called = False
37
 
 
38
 
        def mock_notify(cls, *args):
39
 
            self.notify_called = True
40
 
 
41
 
        self.stubs.Set(nova.notifier.no_op_notifier, 'notify',
42
 
                mock_notify)
43
 
 
44
 
        notifier_api.notify(ctxt, 'publisher_id', 'event_type',
45
 
                            nova.notifier.api.WARN, dict(a=3))
46
 
        self.assertEqual(self.notify_called, True)
47
 
 
48
 
    def test_verify_message_format(self):
49
 
        """A test to ensure changing the message format is prohibitively
50
 
        annoying"""
51
 
 
52
 
        def message_assert(context, message):
53
 
            fields = [('publisher_id', 'publisher_id'),
54
 
                      ('event_type', 'event_type'),
55
 
                      ('priority', 'WARN'),
56
 
                      ('payload', dict(a=3))]
57
 
            for k, v in fields:
58
 
                self.assertEqual(message[k], v)
59
 
            self.assertTrue(len(message['message_id']) > 0)
60
 
            self.assertTrue(len(message['timestamp']) > 0)
61
 
            self.assertEqual(context, ctxt)
62
 
 
63
 
        self.stubs.Set(nova.notifier.no_op_notifier, 'notify',
64
 
                message_assert)
65
 
        notifier_api.notify(ctxt, 'publisher_id', 'event_type',
66
 
                            nova.notifier.api.WARN, dict(a=3))
67
 
 
68
 
    def test_send_rabbit_notification(self):
69
 
        self.stubs.Set(nova.flags.FLAGS, 'notification_driver',
70
 
                'nova.notifier.rabbit_notifier')
71
 
        self.mock_notify = False
72
 
 
73
 
        def mock_notify(cls, *args):
74
 
            self.mock_notify = True
75
 
 
76
 
        self.stubs.Set(nova.openstack.common.rpc, 'notify', mock_notify)
77
 
        notifier_api.notify(ctxt, 'publisher_id', 'event_type',
78
 
                            nova.notifier.api.WARN, dict(a=3))
79
 
 
80
 
        self.assertEqual(self.mock_notify, True)
81
 
 
82
 
    def test_invalid_priority(self):
83
 
        self.assertRaises(nova.notifier.api.BadPriorityException,
84
 
                notifier_api.notify, ctxt, 'publisher_id',
85
 
                'event_type', 'not a priority', dict(a=3))
86
 
 
87
 
    def test_rabbit_priority_queue(self):
88
 
        flags.DECLARE('notification_topics', 'nova.notifier.rabbit_notifier')
89
 
        self.stubs.Set(nova.flags.FLAGS, 'notification_driver',
90
 
                'nova.notifier.rabbit_notifier')
91
 
        self.stubs.Set(nova.flags.FLAGS, 'notification_topics',
92
 
                       ['testnotify', ])
93
 
 
94
 
        self.test_topic = None
95
 
 
96
 
        def mock_notify(context, topic, msg):
97
 
            self.test_topic = topic
98
 
 
99
 
        self.stubs.Set(nova.openstack.common.rpc, 'notify', mock_notify)
100
 
        notifier_api.notify(ctxt, 'publisher_id',
101
 
                            'event_type', 'DEBUG', dict(a=3))
102
 
        self.assertEqual(self.test_topic, 'testnotify.debug')
103
 
 
104
 
    def test_error_notification(self):
105
 
        self.stubs.Set(nova.flags.FLAGS, 'notification_driver',
106
 
            'nova.notifier.rabbit_notifier')
107
 
        self.stubs.Set(nova.flags.FLAGS, 'publish_errors', True)
108
 
        LOG = log.getLogger('nova')
109
 
        log.setup('nova')
110
 
        msgs = []
111
 
 
112
 
        def mock_notify(context, topic, data):
113
 
            msgs.append(data)
114
 
 
115
 
        self.stubs.Set(nova.openstack.common.rpc, 'notify', mock_notify)
116
 
        LOG.error('foo')
117
 
        self.assertEqual(1, len(msgs))
118
 
        msg = msgs[0]
119
 
        self.assertEqual(msg['event_type'], 'error_notification')
120
 
        self.assertEqual(msg['priority'], 'ERROR')
121
 
        self.assertEqual(msg['payload']['error'], 'foo')
122
 
 
123
 
    def test_send_notification_by_decorator(self):
124
 
        self.notify_called = False
125
 
 
126
 
        def example_api(arg1, arg2):
127
 
            return arg1 + arg2
128
 
 
129
 
        example_api = nova.notifier.api.notify_decorator(
130
 
                            'example_api',
131
 
                             example_api)
132
 
 
133
 
        def mock_notify(cls, *args):
134
 
            self.notify_called = True
135
 
 
136
 
        self.stubs.Set(nova.notifier.no_op_notifier, 'notify',
137
 
                mock_notify)
138
 
 
139
 
        self.assertEqual(3, example_api(1, 2))
140
 
        self.assertEqual(self.notify_called, True)
141
 
 
142
 
    def test_decorator_context(self):
143
 
        """Verify that the notify decorator can extract the 'context' arg."""
144
 
        self.notify_called = False
145
 
        self.context_arg = None
146
 
 
147
 
        def example_api(arg1, arg2, context):
148
 
            return arg1 + arg2
149
 
 
150
 
        def example_api2(arg1, arg2, **kw):
151
 
            return arg1 + arg2
152
 
 
153
 
        example_api = nova.notifier.api.notify_decorator(
154
 
                            'example_api',
155
 
                             example_api)
156
 
 
157
 
        example_api2 = nova.notifier.api.notify_decorator(
158
 
                             'example_api2',
159
 
                              example_api2)
160
 
 
161
 
        def mock_notify(context, cls, _type, _priority, _payload):
162
 
            self.notify_called = True
163
 
            self.context_arg = context
164
 
 
165
 
        self.stubs.Set(nova.notifier.api, 'notify',
166
 
                mock_notify)
167
 
 
168
 
        # Test positional context
169
 
        self.assertEqual(3, example_api(1, 2, ctxt))
170
 
        self.assertEqual(self.notify_called, True)
171
 
        self.assertEqual(self.context_arg, ctxt)
172
 
 
173
 
        self.notify_called = False
174
 
        self.context_arg = None
175
 
 
176
 
        # Test named context
177
 
        self.assertEqual(3, example_api2(1, 2, context=ctxt2))
178
 
        self.assertEqual(self.notify_called, True)
179
 
        self.assertEqual(self.context_arg, ctxt2)
180
 
 
181
 
        # Test missing context
182
 
        self.assertEqual(3, example_api2(1, 2, bananas="delicious"))
183
 
        self.assertEqual(self.notify_called, True)
184
 
        self.assertEqual(self.context_arg, None)