24
24
from nova.compute import vm_states
25
25
from nova import context
26
26
from nova import db
27
from nova import flags
27
from nova.network import api as network_api
29
28
from nova import notifications
30
29
from nova.openstack.common import log as logging
31
30
from nova.openstack.common.notifier import api as notifier_api
49
47
self.assertTrue(ctxt.is_admin)
50
48
return self.net_info
52
self.stubs.Set(nova.network.API, 'get_instance_nw_info',
50
self.stubs.Set(network_api.API, 'get_instance_nw_info',
54
52
fake_network.set_stub_network_methods(self.stubs)
82
80
inst['access_ip_v4'] = '1.2.3.4'
83
81
inst['access_ip_v6'] = 'feed:5eed'
84
82
inst['display_name'] = 'test_instance'
83
inst['hostname'] = 'test_instance_hostname'
86
85
inst.update(params)
87
86
return db.instance_create(self.context, inst)
88
def test_send_api_fault_disabled(self):
89
self.flags(notify_api_faults=False)
90
notifications.send_api_fault("http://example.com/foo", 500, None)
91
self.assertEquals(0, len(test_notifier.NOTIFICATIONS))
93
def test_send_api_fault(self):
94
self.flags(notify_api_faults=True)
97
# Get a real exception with a call stack.
98
raise test.TestingException("junk")
99
except test.TestingException, e:
102
notifications.send_api_fault("http://example.com/foo", 500, exception)
104
self.assertEquals(1, len(test_notifier.NOTIFICATIONS))
105
n = test_notifier.NOTIFICATIONS[0]
106
self.assertEquals(n['priority'], 'ERROR')
107
self.assertEquals(n['event_type'], 'api.fault')
108
self.assertEquals(n['payload']['url'], 'http://example.com/foo')
109
self.assertEquals(n['payload']['status'], 500)
110
self.assertTrue(n['payload']['exception'] is not None)
89
112
def test_notif_disabled(self):
91
114
# test config disable of the notifcations
187
210
access_ip_v4 = self.instance["access_ip_v4"]
188
211
access_ip_v6 = self.instance["access_ip_v6"]
189
212
display_name = self.instance["display_name"]
213
hostname = self.instance["hostname"]
191
215
self.assertEquals(vm_states.BUILDING, payload["old_state"])
192
216
self.assertEquals(vm_states.ACTIVE, payload["state"])
195
219
self.assertEquals(payload["access_ip_v4"], access_ip_v4)
196
220
self.assertEquals(payload["access_ip_v6"], access_ip_v6)
197
221
self.assertEquals(payload["display_name"], display_name)
222
self.assertEquals(payload["hostname"], hostname)
199
224
def test_task_update_with_states(self):
200
225
self.flags(notify_on_state_change="vm_and_task_state")
208
233
access_ip_v4 = self.instance["access_ip_v4"]
209
234
access_ip_v6 = self.instance["access_ip_v6"]
210
235
display_name = self.instance["display_name"]
236
hostname = self.instance["hostname"]
212
238
self.assertEquals(vm_states.BUILDING, payload["old_state"])
213
239
self.assertEquals(vm_states.BUILDING, payload["state"])
216
242
self.assertEquals(payload["access_ip_v4"], access_ip_v4)
217
243
self.assertEquals(payload["access_ip_v6"], access_ip_v6)
218
244
self.assertEquals(payload["display_name"], display_name)
245
self.assertEquals(payload["hostname"], hostname)
220
247
def test_update_no_service_name(self):
221
248
notifications.send_update_with_states(self.context, self.instance,