← Back to branch summary

~gandelman-a/ubuntu/precise/nova/UCA_2012.2.1

« back to all changes in this revision

Viewing changes to nova/tests/test_notifications.py

  • Committer: Package Import Robot
  • Author(s): Chuck Short, Chuck Short, Adam Gandelman
  • Date: 2012-08-16 14:04:11 UTC
  • mfrom: (1.1.59)
  • Revision ID: package-import@ubuntu.com-20120816140411-8dvudjblnx1w0mwx
Tags: 2012.2~f3-0ubuntu1
https://launchpad.net/bugs/1002111
[ Chuck Short ]
* New upstream version.
* debian/rules: Re-enable testsuite.
* debian/control:
  - Add python-quantumclient as a build depends.
  - Bump standards to 3.9.3
  - Fix lintian warnings.
  - Recommend python-glanceclient and python-keystoneclient.
  - Add dependency of iptables for nova-network.
* debian/watch: Update
* debian/rules: Do not run pep8 tests since upstream is still using an
  older pep8.
* debian/patches/0001-Update-tools-hacking-for-pep8-1.2-and-
  beyond.patch: Get the testsuite running again.
* debian/nova-volume.install, debian/nova_tgt: Add support for
  persistent volumes.

[ Adam Gandelman ]
* debian/{nova-api.install, nova-api-metadata.install}: Install
  api-metadata.filters. (LP: #1002111)
* debian/control: Added python-glanceclient.

Show diffs side-by-side

added added

removed removed

Lines of Context:
nova/tests/test_notifications.py
27
27
from nova import flags
28
28
import nova.network
29
29
from nova import notifications
30
 
from nova.notifier import test_notifier
31
30
from nova.openstack.common import log as logging
 
31
from nova.openstack.common.notifier import api as notifier_api
 
32
from nova.openstack.common.notifier import test_notifier
32
33
from nova import test
33
34
from nova.tests import fake_network
34
35
 
40
41
class NotificationsTestCase(test.TestCase):
41
42
 
42
43
    def setUp(self):
 
44
        super(NotificationsTestCase, self).setUp()
 
45
 
 
46
        self.net_info = fake_network.fake_get_instance_nw_info(self.stubs, 1,
 
47
                                                        1, spectacular=True)
43
48
 
44
49
        def fake_get_nw_info(cls, ctxt, instance):
45
50
            self.assertTrue(ctxt.is_admin)
46
 
            return fake_network.fake_get_instance_nw_info(self.stubs, 1, 1,
47
 
                spectacular=True)
 
51
            return self.net_info
48
52
 
49
 
        super(NotificationsTestCase, self).setUp()
50
53
        self.stubs.Set(nova.network.API, 'get_instance_nw_info',
51
54
                fake_get_nw_info)
52
55
 
53
56
        self.flags(compute_driver='nova.virt.fake.FakeDriver',
54
57
                   stub_network=True,
55
 
                   notification_driver='nova.notifier.test_notifier',
 
58
          notification_driver=['nova.openstack.common.notifier.test_notifier'],
56
59
                   network_manager='nova.network.manager.FlatManager',
57
60
                   notify_on_state_change="vm_and_task_state",
58
61
                   host='testhost')
64
67
 
65
68
        self.instance = self._wrapped_create()
66
69
 
 
70
    def tearDown(self):
 
71
        notifier_api._reset_drivers()
 
72
        super(NotificationsTestCase, self).tearDown()
 
73
 
67
74
    def _wrapped_create(self, params=None):
68
75
        inst = {}
69
76
        inst['image_ref'] = 1
73
80
        inst['instance_type_id'] = type_id
74
81
        inst['root_gb'] = 0
75
82
        inst['ephemeral_gb'] = 0
 
83
        inst['access_ip_v4'] = '1.2.3.4'
 
84
        inst['access_ip_v6'] = 'feed:5eed'
 
85
        inst['display_name'] = 'test_instance'
76
86
        if params:
77
87
            inst.update(params)
78
88
        return db.instance_create(self.context, inst)
81
91
 
82
92
        # test config disable of the notifcations
83
93
        self.flags(notify_on_state_change=None)
 
94
        self.flags(notify_on_any_change=False)
84
95
 
85
96
        old = copy.copy(self.instance)
86
97
        self.instance["vm_state"] = vm_states.ACTIVE
87
98
 
 
99
        old_vm_state = old['vm_state']
 
100
        new_vm_state = self.instance["vm_state"]
 
101
        old_task_state = old['task_state']
 
102
        new_task_state = self.instance["task_state"]
 
103
 
 
104
        notifications.send_update_with_states(self.context, self.instance,
 
105
                old_vm_state, new_vm_state, old_task_state, new_task_state,
 
106
                verify_states=True)
 
107
 
88
108
        notifications.send_update(self.context, old, self.instance)
89
109
        self.assertEquals(0, len(test_notifier.NOTIFICATIONS))
90
110
 
97
117
        old = copy.copy(self.instance)
98
118
        self.instance["task_state"] = task_states.SPAWNING
99
119
 
100
 
        notifications.send_update(self.context, old, self.instance)
 
120
        old_vm_state = old['vm_state']
 
121
        new_vm_state = self.instance["vm_state"]
 
122
        old_task_state = old['task_state']
 
123
        new_task_state = self.instance["task_state"]
 
124
 
 
125
        notifications.send_update_with_states(self.context, self.instance,
 
126
                old_vm_state, new_vm_state, old_task_state, new_task_state,
 
127
                verify_states=True)
 
128
 
101
129
        self.assertEquals(0, len(test_notifier.NOTIFICATIONS))
102
130
 
103
131
        # ok now enable task state notifcations and re-try
109
137
    def test_send_no_notif(self):
110
138
 
111
139
        # test notification on send no initial vm state:
112
 
        notifications.send_update(self.context, self.instance, self.instance)
 
140
        old_vm_state = self.instance['vm_state']
 
141
        new_vm_state = self.instance['vm_state']
 
142
        old_task_state = self.instance['task_state']
 
143
        new_task_state = self.instance['task_state']
 
144
 
 
145
        notifications.send_update_with_states(self.context, self.instance,
 
146
                old_vm_state, new_vm_state, old_task_state, new_task_state,
 
147
                service="compute", host=None, verify_states=True)
 
148
 
113
149
        self.assertEquals(0, len(test_notifier.NOTIFICATIONS))
114
150
 
115
151
    def test_send_on_vm_change(self):
138
174
 
139
175
        notifications.send_update_with_states(self.context, self.instance,
140
176
                vm_states.BUILDING, vm_states.BUILDING, task_states.SPAWNING,
141
 
                task_states.SPAWNING)
 
177
                task_states.SPAWNING, verify_states=True)
142
178
        self.assertEquals(0, len(test_notifier.NOTIFICATIONS))
143
179
 
144
180
    def test_vm_update_with_states(self):
145
181
 
146
182
        notifications.send_update_with_states(self.context, self.instance,
147
183
                vm_states.BUILDING, vm_states.ACTIVE, task_states.SPAWNING,
148
 
                task_states.SPAWNING)
 
184
                task_states.SPAWNING, verify_states=True)
149
185
        self.assertEquals(1, len(test_notifier.NOTIFICATIONS))
150
186
        notif = test_notifier.NOTIFICATIONS[0]
151
187
        payload = notif["payload"]
 
188
        access_ip_v4 = self.instance["access_ip_v4"]
 
189
        access_ip_v6 = self.instance["access_ip_v6"]
 
190
        display_name = self.instance["display_name"]
152
191
 
153
192
        self.assertEquals(vm_states.BUILDING, payload["old_state"])
154
193
        self.assertEquals(vm_states.ACTIVE, payload["state"])
155
194
        self.assertEquals(task_states.SPAWNING, payload["old_task_state"])
156
195
        self.assertEquals(task_states.SPAWNING, payload["new_task_state"])
 
196
        self.assertEquals(payload["access_ip_v4"], access_ip_v4)
 
197
        self.assertEquals(payload["access_ip_v6"], access_ip_v6)
 
198
        self.assertEquals(payload["display_name"], display_name)
157
199
 
158
200
    def test_task_update_with_states(self):
 
201
        self.flags(notify_on_state_change="vm_and_task_state")
159
202
 
160
203
        notifications.send_update_with_states(self.context, self.instance,
161
204
                vm_states.BUILDING, vm_states.BUILDING, task_states.SPAWNING,
162
 
                None)
 
205
                None, verify_states=True)
163
206
        self.assertEquals(1, len(test_notifier.NOTIFICATIONS))
164
207
        notif = test_notifier.NOTIFICATIONS[0]
165
208
        payload = notif["payload"]
 
209
        access_ip_v4 = self.instance["access_ip_v4"]
 
210
        access_ip_v6 = self.instance["access_ip_v6"]
 
211
        display_name = self.instance["display_name"]
166
212
 
167
213
        self.assertEquals(vm_states.BUILDING, payload["old_state"])
168
214
        self.assertEquals(vm_states.BUILDING, payload["state"])
169
215
        self.assertEquals(task_states.SPAWNING, payload["old_task_state"])
170
216
        self.assertEquals(None, payload["new_task_state"])
 
217
        self.assertEquals(payload["access_ip_v4"], access_ip_v4)
 
218
        self.assertEquals(payload["access_ip_v6"], access_ip_v6)
 
219
        self.assertEquals(payload["display_name"], display_name)
171
220
 
172
221
    def test_update_no_service_name(self):
173
222
        notifications.send_update_with_states(self.context, self.instance,
198
247
        # service name should default to 'compute'
199
248
        notif = test_notifier.NOTIFICATIONS[0]
200
249
        self.assertEquals('compute.someotherhost', notif['publisher_id'])
 
250
 
 
251
    def test_payload_has_fixed_ip_labels(self):
 
252
        info = notifications.info_from_instance(self.context, self.instance,
 
253
                                                  self.net_info, None)
 
254
        self.assertTrue("fixed_ips" in info)
 
255
        self.assertEquals(info["fixed_ips"][0]["label"], "test1")
 
256
 
 
257
    def test_send_access_ip_update(self):
 
258
        notifications.send_update(self.context, self.instance, self.instance)
 
259
        self.assertEquals(1, len(test_notifier.NOTIFICATIONS))
 
260
        notif = test_notifier.NOTIFICATIONS[0]
 
261
        payload = notif["payload"]
 
262
        access_ip_v4 = self.instance["access_ip_v4"]
 
263
        access_ip_v6 = self.instance["access_ip_v6"]
 
264
 
 
265
        self.assertEquals(payload["access_ip_v4"], access_ip_v4)
 
266
        self.assertEquals(payload["access_ip_v6"], access_ip_v6)
 
267
 
 
268
    def test_send_name_update(self):
 
269
        notifications.send_update(self.context, self.instance, self.instance)
 
270
        self.assertEquals(1, len(test_notifier.NOTIFICATIONS))
 
271
        notif = test_notifier.NOTIFICATIONS[0]
 
272
        payload = notif["payload"]
 
273
        display_name = self.instance["display_name"]
 
274
 
 
275
        self.assertEquals(payload["display_name"], display_name)
 
276
 
 
277
    def test_send_no_state_change(self):
 
278
        called = [False]
 
279
 
 
280
        def sending_no_state_change(context, instance, **kwargs):
 
281
            called[0] = True
 
282
        self.stubs.Set(notifications, '_send_instance_update_notification',
 
283
                       sending_no_state_change)
 
284
        notifications.send_update(self.context, self.instance, self.instance)
 
285
        self.assertTrue(called[0])
 
286
 
 
287
    def test_fail_sending_update(self):
 
288
        def fail_sending(context, instance, **kwargs):
 
289
            raise Exception('failed to notify')
 
290
        self.stubs.Set(notifications, '_send_instance_update_notification',
 
291
                       fail_sending)
 
292
 
 
293
        notifications.send_update(self.context, self.instance, self.instance)
 
294
        self.assertEquals(0, len(test_notifier.NOTIFICATIONS))