openstack/aodh
Code Issues Proposed changes

Merge "simplify crud notification test"

Browse Source
This commit is contained in:
Jenkins 2017-06-26 18:01:30 +00:00 committed by Gerrit Code Review
commit ba461b4f03
1 changed files with 14 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

35
aodh/tests/functional/api/v2/test_alarm_scenarios.py
View File

@ -18,7 +18,6 @@ import datetime
import os import os
import mock import mock
import oslo_messaging.conffixture
from oslo_serialization import jsonutils from oslo_serialization import jsonutils
from oslo_utils import uuidutils from oslo_utils import uuidutils
import six import six
@ -1667,30 +1666,22 @@ class TestAlarms(TestAlarmsBase):
} }
} }
endpoint = mock.MagicMock() with mock.patch.object(messaging, 'get_notifier') as get_notifier:
target = oslo_messaging.Target(topic="notifications") notifier = get_notifier.return_value
listener = messaging.get_batch_notification_listener(
self.transport, [target], [endpoint])
listener.start()
endpoint.info.side_effect = lambda *args: listener.stop()
self.post_json('/alarms', params=json, headers=self.auth_headers) self.post_json('/alarms', params=json, headers=self.auth_headers)
listener.wait() get_notifier.assert_called_once_with(mock.ANY,
publisher_id='aodh.api')
class NotificationsMatcher(object): calls = notifier.info.call_args_list
def __eq__(self, notifications): self.assertEqual(1, len(calls))
payload = notifications[0]['payload'] args, _ = calls[0]
return (payload['detail']['name'] == 'sent_notification' and context, event_type, payload = args
payload['type'] == 'creation' and self.assertEqual('alarm.creation', event_type)
payload['detail']['rule']['meter_name'] == 'ameter' and self.assertEqual('sent_notification', payload['detail']['name'])
set(['alarm_id', 'detail', 'event_id', 'on_behalf_of', self.assertEqual('ameter', payload['detail']['rule']['meter_name'])
'project_id', 'timestamp', self.assertTrue(set(['alarm_id', 'detail', 'event_id', 'on_behalf_of',
'project_id', 'timestamp', 'type',
'user_id']).issubset(payload.keys())) 'user_id']).issubset(payload.keys()))
def __ne__(self, other):
return not self.__eq__(other)
endpoint.info.assert_called_once_with(NotificationsMatcher())
def test_alarm_sends_notification(self): def test_alarm_sends_notification(self):
with mock.patch.object(messaging, 'get_notifier') as get_notifier: with mock.patch.object(messaging, 'get_notifier') as get_notifier:
notifier = get_notifier.return_value notifier = get_notifier.return_value