"Fossies" - the Fresh Open Source Software Archive

Member "masakari-9.0.0/masakari/tests/unit/notifications/objects/test_notification.py" (13 May 2020, 10810 Bytes) of package /linux/misc/openstack/masakari-9.0.0.tar.gz:


As a special service "Fossies" has tried to format the requested source page into HTML format using (guessed) Python source code syntax highlighting (style: standard) with prefixed line numbers. Alternatively you can here view or download the uninterpreted source code file. See also the latest Fossies "Diffs" side-by-side code changes report for "test_notification.py": 8.0.0_vs_9.0.0.

    1 # Copyright (c) 2018 NTT DATA
    2 #
    3 # Licensed under the Apache License, Version 2.0 (the "License");
    4 # you may not use this file except in compliance with the License.
    5 # You may obtain a copy of the License at
    6 #
    7 #    http://www.apache.org/licenses/LICENSE-2.0
    8 #
    9 # Unless required by applicable law or agreed to in writing, software
   10 # distributed under the License is distributed on an "AS IS" BASIS,
   11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
   12 # implied.
   13 # See the License for the specific language governing permissions and
   14 # limitations under the License.
   15 
   16 import collections
   17 from unittest import mock
   18 
   19 from oslo_utils import timeutils
   20 from oslo_versionedobjects import fixture
   21 
   22 from masakari.notifications.objects import base as notification
   23 from masakari.objects import base
   24 from masakari.objects import fields
   25 from masakari import test
   26 
   27 
   28 class TestNotificationBase(test.NoDBTestCase):
   29 
   30     @base.MasakariObjectRegistry.register_if(False)
   31     class TestObject(base.MasakariObject):
   32         VERSION = '1.0'
   33         fields = {
   34             'field_1': fields.StringField(),
   35             'field_2': fields.IntegerField(),
   36             'not_important_field': fields.IntegerField(),
   37         }
   38 
   39     @base.MasakariObjectRegistry.register_if(False)
   40     class TestNotificationPayload(notification.NotificationPayloadBase):
   41         VERSION = '1.0'
   42 
   43         SCHEMA = {
   44             'field_1': ('source_field', 'field_1'),
   45             'field_2': ('source_field', 'field_2'),
   46         }
   47 
   48         fields = {
   49             'extra_field': fields.StringField(),  # filled by ctor
   50             'field_1': fields.StringField(),  # filled by the schema
   51             'field_2': fields.IntegerField(),   # filled by the schema
   52         }
   53 
   54         def populate_schema(self, source_field):
   55             super(TestNotificationBase.TestNotificationPayload,
   56                   self).populate_schema(source_field=source_field)
   57 
   58     @base.MasakariObjectRegistry.register_if(False)
   59     class TestNotificationPayloadEmptySchema(
   60         notification.NotificationPayloadBase):
   61         VERSION = '1.0'
   62 
   63         fields = {
   64             'extra_field': fields.StringField(),  # filled by ctor
   65         }
   66 
   67     @notification.notification_sample('test-update-1.json')
   68     @notification.notification_sample('test-update-2.json')
   69     @base.MasakariObjectRegistry.register_if(False)
   70     class TestNotification(notification.NotificationBase):
   71         VERSION = '1.0'
   72         fields = {
   73             'payload': fields.ObjectField('TestNotificationPayload')
   74         }
   75 
   76     @base.MasakariObjectRegistry.register_if(False)
   77     class TestNotificationEmptySchema(notification.NotificationBase):
   78         VERSION = '1.0'
   79         fields = {
   80             'payload': fields.ObjectField('TestNotificationPayloadEmptySchema')
   81         }
   82 
   83     fake_service = {
   84         'created_at': timeutils.utcnow().replace(microsecond=0),
   85         'updated_at': None,
   86         'deleted_at': None,
   87         'deleted': False,
   88         'id': 123,
   89         'host': 'fake-host',
   90         'binary': 'masakari-fake',
   91         'topic': 'fake-service-topic',
   92         'report_count': 1,
   93         'forced_down': False,
   94         'disabled': False,
   95         'disabled_reason': None,
   96         'last_seen_up': None,
   97         'version': 1}
   98 
   99     expected_payload = {
  100         'masakari_object.name': 'TestNotificationPayload',
  101         'masakari_object.data': {
  102             'extra_field': 'test string',
  103             'field_1': 'test1',
  104             'field_2': 42},
  105         'masakari_object.version': '1.0',
  106         'masakari_object.namespace': 'masakari'}
  107 
  108     def setUp(self):
  109         super(TestNotificationBase, self).setUp()
  110         mock_context = mock.Mock()
  111         mock_context.to_dict.return_value = {}
  112         self.publisher = notification.NotificationPublisher(
  113             context=mock_context, host='fake-host',
  114             binary='masakari-fake')
  115 
  116         self.my_obj = self.TestObject(field_1='test1',
  117                                       field_2=42,
  118                                       not_important_field=13)
  119 
  120         self.payload = self.TestNotificationPayload(
  121             extra_field='test string')
  122         self.payload.populate_schema(source_field=self.my_obj)
  123 
  124         self.notification = self.TestNotification(
  125             event_type=notification.EventType(
  126                 object='test_object',
  127                 action=fields.EventNotificationAction.SEGMENT_CREATE,
  128                 phase=fields.EventNotificationPhase.START),
  129             publisher=self.publisher,
  130             priority=fields.EventNotificationPriority.INFO,
  131             payload=self.payload)
  132 
  133     def _verify_notification(self, mock_notifier, mock_context,
  134                              expected_event_type,
  135                              expected_payload):
  136         mock_notifier.prepare.assert_called_once_with(
  137             publisher_id='masakari-fake:fake-host')
  138         mock_notify = mock_notifier.prepare.return_value.info
  139         self.assertTrue(mock_notify.called)
  140         self.assertEqual(mock_notify.call_args[0][0], mock_context)
  141         self.assertEqual(mock_notify.call_args[1]['event_type'],
  142                          expected_event_type)
  143         actual_payload = mock_notify.call_args[1]['payload']
  144         self.assertJsonEqual(expected_payload, actual_payload)
  145 
  146     @mock.patch('masakari.rpc.NOTIFIER')
  147     def test_emit_notification(self, mock_notifier):
  148 
  149         mock_context = mock.Mock()
  150         mock_context.to_dict.return_value = {}
  151         self.notification.emit(mock_context)
  152 
  153         self._verify_notification(
  154             mock_notifier,
  155             mock_context,
  156             expected_event_type='segment.create.start',
  157             expected_payload=self.expected_payload)
  158 
  159     @mock.patch('masakari.rpc.NOTIFIER')
  160     def test_emit_with_host_and_binary_as_publisher(self, mock_notifier):
  161         noti = self.TestNotification(
  162             event_type=notification.EventType(
  163                 object='test_object',
  164                 action=fields.EventNotificationAction.SEGMENT_CREATE),
  165             publisher=notification.NotificationPublisher(
  166                 host='fake-host', binary='masakari-fake'),
  167             priority=fields.EventNotificationPriority.INFO,
  168             payload=self.payload)
  169 
  170         mock_context = mock.Mock()
  171         mock_context.to_dict.return_value = {}
  172         noti.emit(mock_context)
  173 
  174         self._verify_notification(
  175             mock_notifier,
  176             mock_context,
  177             expected_event_type='segment.create',
  178             expected_payload=self.expected_payload)
  179 
  180     @mock.patch('masakari.rpc.NOTIFIER')
  181     def test_emit_event_type_without_phase(self, mock_notifier):
  182         noti = self.TestNotification(
  183             event_type=notification.EventType(
  184                 object='test_object',
  185                 action=fields.EventNotificationAction.SEGMENT_CREATE),
  186             publisher=self.publisher,
  187             priority=fields.EventNotificationPriority.INFO,
  188             payload=self.payload)
  189 
  190         mock_context = mock.Mock()
  191         mock_context.to_dict.return_value = {}
  192         noti.emit(mock_context)
  193 
  194         self._verify_notification(
  195             mock_notifier,
  196             mock_context,
  197             expected_event_type='segment.create',
  198             expected_payload=self.expected_payload)
  199 
  200     @mock.patch('masakari.rpc.NOTIFIER')
  201     def test_not_possible_to_emit_if_not_populated(self, mock_notifier):
  202         non_populated_payload = self.TestNotificationPayload(
  203             extra_field='test string')
  204         noti = self.TestNotification(
  205             event_type=notification.EventType(
  206                 object='test_object',
  207                 action=fields.EventNotificationAction.SEGMENT_CREATE),
  208             publisher=self.publisher,
  209             priority=fields.EventNotificationPriority.INFO,
  210             payload=non_populated_payload)
  211 
  212         mock_context = mock.Mock()
  213         self.assertRaises(AssertionError, noti.emit, mock_context)
  214         self.assertFalse(mock_notifier.called)
  215 
  216     @mock.patch('masakari.rpc.NOTIFIER')
  217     def test_empty_schema(self, mock_notifier):
  218         non_populated_payload = self.TestNotificationPayloadEmptySchema(
  219             extra_field='test string')
  220         noti = self.TestNotificationEmptySchema(
  221             event_type=notification.EventType(
  222                 object='test_object',
  223                 action=fields.EventNotificationAction.SEGMENT_CREATE),
  224             publisher=self.publisher,
  225             priority=fields.EventNotificationPriority.INFO,
  226             payload=non_populated_payload)
  227 
  228         mock_context = mock.Mock()
  229         mock_context.to_dict.return_value = {}
  230         noti.emit(mock_context)
  231 
  232         self._verify_notification(
  233             mock_notifier,
  234             mock_context,
  235             expected_event_type='segment.create',
  236             expected_payload={
  237                 'masakari_object.name': 'TestNotificationPayloadEmptySchema',
  238                 'masakari_object.data': {'extra_field': u'test string'},
  239                 'masakari_object.version': '1.0',
  240                 'masakari_object.namespace': 'masakari'})
  241 
  242     def test_sample_decorator(self):
  243         self.assertEqual(2, len(self.TestNotification.samples))
  244         self.assertIn('test-update-1.json', self.TestNotification.samples)
  245         self.assertIn('test-update-2.json', self.TestNotification.samples)
  246 
  247 
  248 class TestNotificationObjectVersions(test.NoDBTestCase):
  249 
  250     def test_notification_payload_version_depends_on_the_schema(self):
  251         @base.MasakariObjectRegistry.register_if(False)
  252         class TestNotificationPayload(notification.NotificationPayloadBase):
  253             VERSION = '1.0'
  254 
  255             SCHEMA = {
  256                 'field_1': ('source_field', 'field_1'),
  257                 'field_2': ('source_field', 'field_2'),
  258             }
  259 
  260             fields = {
  261                 'extra_field': fields.StringField(),  # filled by ctor
  262                 'field_1': fields.StringField(),  # filled by the schema
  263                 'field_2': fields.IntegerField(),   # filled by the schema
  264             }
  265 
  266         checker = fixture.ObjectVersionChecker(
  267             {'TestNotificationPayload': (TestNotificationPayload,)})
  268 
  269         old_hash = checker.get_hashes(extra_data_func=get_extra_data)
  270         TestNotificationPayload.SCHEMA['field_3'] = ('source_field',
  271                                                      'field_3')
  272         new_hash = checker.get_hashes(extra_data_func=get_extra_data)
  273 
  274         self.assertNotEqual(old_hash, new_hash)
  275 
  276 
  277 def get_extra_data(obj_class):
  278     extra_data = tuple()
  279 
  280     # Get the SCHEMA items to add to the fingerprint
  281     # if we are looking at a notification
  282     if issubclass(obj_class, notification.NotificationPayloadBase):
  283         schema_data = collections.OrderedDict(
  284             sorted(obj_class.SCHEMA.items()))
  285 
  286         extra_data += (schema_data,)
  287 
  288     return extra_data