"Fossies" - the Fresh Open Source Software Archive

Member "masakari-9.0.0/masakari/tests/unit/objects/test_notifications.py" (13 May 2020, 11272 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_notifications.py": 8.0.0_vs_9.0.0.

    1 #    Copyright 2016 NTT DATA
    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 copy
   17 from unittest import mock
   18 
   19 from oslo_utils import timeutils
   20 from oslo_utils import uuidutils
   21 
   22 from masakari.api import utils as api_utils
   23 from masakari import db
   24 from masakari import exception
   25 from masakari.objects import fields
   26 from masakari.objects import notification
   27 from masakari.tests.unit.objects import test_objects
   28 from masakari.tests import uuidsentinel
   29 
   30 NOW = timeutils.utcnow().replace(microsecond=0)
   31 OPTIONAL = ['recovery_workflow_details']
   32 
   33 
   34 def _fake_db_notification(**kwargs):
   35     fake_notification = {
   36         'created_at': NOW,
   37         'updated_at': None,
   38         'deleted_at': None,
   39         'deleted': False,
   40         'id': 123,
   41         'notification_uuid': uuidsentinel.fake_notification,
   42         'generated_time': NOW,
   43         'type': 'COMPUTE_HOST',
   44         'payload': '{"fake_key": "fake_value"}',
   45         'status': 'new',
   46         'source_host_uuid': uuidsentinel.fake_host,
   47         }
   48     fake_notification.update(kwargs)
   49     return fake_notification
   50 
   51 
   52 def _fake_object_notification(**kwargs):
   53     fake_notification = {
   54         'created_at': NOW,
   55         'updated_at': None,
   56         'deleted_at': None,
   57         'deleted': False,
   58         'id': 123,
   59         'notification_uuid': uuidsentinel.fake_notification,
   60         'generated_time': NOW,
   61         'type': 'COMPUTE_HOST',
   62         'payload': {"fake_key": "fake_value"},
   63         'status': 'new',
   64         'source_host_uuid': uuidsentinel.fake_host,
   65         }
   66     fake_notification.update(kwargs)
   67     return fake_notification
   68 
   69 fake_object_notification = _fake_object_notification()
   70 
   71 fake_db_notification = _fake_db_notification()
   72 
   73 
   74 class TestNotificationObject(test_objects._LocalTest):
   75 
   76     def _test_query(self, db_method, obj_method, *args, **kwargs):
   77         with mock.patch.object(db, db_method) as mock_db:
   78 
   79             db_exception = kwargs.pop('db_exception', None)
   80             if db_exception:
   81                 mock_db.side_effect = db_exception
   82             else:
   83                 mock_db.return_value = fake_db_notification
   84 
   85             obj = getattr(notification.Notification, obj_method
   86                           )(self.context, *args, **kwargs)
   87             if db_exception:
   88                 self.assertIsNone(obj)
   89 
   90             self.compare_obj(obj, fake_object_notification,
   91                              allow_missing=OPTIONAL)
   92 
   93     def test_get_by_id(self):
   94         self._test_query('notification_get_by_id', 'get_by_id', 123)
   95 
   96     def test_get_by_uuid(self):
   97         self._test_query('notification_get_by_uuid', 'get_by_uuid',
   98                          uuidsentinel.fake_segment)
   99 
  100     def _notification_create_attributes(self, skip_uuid=False):
  101 
  102         notification_obj = notification.Notification(context=self.context)
  103         notification_obj.generated_time = NOW
  104         notification_obj.type = "COMPUTE_HOST"
  105         notification_obj.payload = {'fake_key': 'fake_value'}
  106         notification_obj.status = "new"
  107         if not skip_uuid:
  108             notification_obj.notification_uuid = uuidsentinel.fake_notification
  109         notification_obj.source_host_uuid = uuidsentinel.fake_host
  110 
  111         return notification_obj
  112 
  113     @mock.patch.object(api_utils, 'notify_about_notification_api')
  114     @mock.patch.object(db, 'notification_create')
  115     def test_create(self, mock_db_create, mock_notify_about_notification_api):
  116 
  117         mock_db_create.return_value = fake_db_notification
  118         notification_obj = self._notification_create_attributes()
  119         notification_obj.create()
  120 
  121         self.compare_obj(notification_obj, fake_object_notification,
  122                          allow_missing=OPTIONAL)
  123         mock_db_create.assert_called_once_with(self.context, {
  124             'source_host_uuid': uuidsentinel.fake_host,
  125             'notification_uuid': uuidsentinel.fake_notification,
  126             'generated_time': NOW, 'status': 'new',
  127             'type': 'COMPUTE_HOST', 'payload': '{"fake_key": "fake_value"}'})
  128         action = fields.EventNotificationAction.NOTIFICATION_CREATE
  129         phase_start = fields.EventNotificationPhase.START
  130         phase_end = fields.EventNotificationPhase.END
  131         notify_calls = [
  132             mock.call(self.context, notification_obj, action=action,
  133                       phase=phase_start),
  134             mock.call(self.context, notification_obj, action=action,
  135                       phase=phase_end)]
  136         mock_notify_about_notification_api.assert_has_calls(notify_calls)
  137 
  138     @mock.patch.object(api_utils, 'notify_about_notification_api')
  139     @mock.patch.object(db, 'notification_create')
  140     def test_recreate_fails(self, mock_notification_create,
  141                             mock_notify_about_notification_api):
  142         mock_notification_create.return_value = fake_db_notification
  143         notification_obj = self._notification_create_attributes()
  144         notification_obj.create()
  145 
  146         self.assertRaises(exception.ObjectActionError, notification_obj.create)
  147 
  148         mock_notification_create.assert_called_once_with(self.context, {
  149             'source_host_uuid': uuidsentinel.fake_host,
  150             'notification_uuid': uuidsentinel.fake_notification,
  151             'generated_time': NOW, 'status': 'new',
  152             'type': 'COMPUTE_HOST', 'payload': '{"fake_key": "fake_value"}'})
  153         action = fields.EventNotificationAction.NOTIFICATION_CREATE
  154         phase_start = fields.EventNotificationPhase.START
  155         phase_end = fields.EventNotificationPhase.END
  156         notify_calls = [
  157             mock.call(self.context, notification_obj, action=action,
  158                       phase=phase_start),
  159             mock.call(self.context, notification_obj, action=action,
  160                       phase=phase_end)]
  161         mock_notify_about_notification_api.assert_has_calls(notify_calls)
  162 
  163     @mock.patch.object(api_utils, 'notify_about_notification_api')
  164     @mock.patch.object(db, 'notification_create')
  165     @mock.patch.object(uuidutils, 'generate_uuid')
  166     def test_create_without_passing_uuid_in_updates(self, mock_generate_uuid,
  167                         mock_db_create, mock_notify_about_notification_api):
  168 
  169         mock_db_create.return_value = fake_db_notification
  170         mock_generate_uuid.return_value = uuidsentinel.fake_notification
  171         notification_obj = self._notification_create_attributes(skip_uuid=True)
  172 
  173         notification_obj.create()
  174 
  175         self.compare_obj(notification_obj, fake_object_notification,
  176                          allow_missing=OPTIONAL)
  177         mock_db_create.assert_called_once_with(self.context, {
  178             'source_host_uuid': uuidsentinel.fake_host,
  179             'notification_uuid': uuidsentinel.fake_notification,
  180             'generated_time': NOW, 'status': 'new',
  181             'type': 'COMPUTE_HOST', 'payload': '{"fake_key": "fake_value"}'})
  182         self.assertTrue(mock_generate_uuid.called)
  183         action = fields.EventNotificationAction.NOTIFICATION_CREATE
  184         phase_start = fields.EventNotificationPhase.START
  185         notify_calls = [
  186             mock.call(self.context, notification_obj, action=action,
  187                       phase=phase_start)]
  188         mock_notify_about_notification_api.assert_has_calls(notify_calls)
  189 
  190     @mock.patch.object(db, 'notification_delete')
  191     def test_destroy(self, mock_notification_delete):
  192         notification_obj = self._notification_create_attributes()
  193         notification_obj.id = 123
  194         notification_obj.destroy()
  195 
  196         (mock_notification_delete.
  197          assert_called_once_with(self.context, uuidsentinel.fake_notification))
  198         self.assertRaises(NotImplementedError, lambda: notification_obj.id)
  199 
  200     @mock.patch.object(db, 'notification_delete')
  201     def test_destroy_without_id(self, mock_destroy):
  202         notification_obj = self._notification_create_attributes()
  203         self.assertRaises(exception.ObjectActionError,
  204                           notification_obj.destroy)
  205         self.assertFalse(mock_destroy.called)
  206 
  207     @mock.patch.object(db, 'notification_delete')
  208     def test_destroy_without_notification_uuid(self, mock_destroy):
  209         notification_obj = self._notification_create_attributes(skip_uuid=True)
  210         notification_obj.id = 123
  211 
  212         self.assertRaises(exception.ObjectActionError,
  213                           notification_obj.destroy)
  214         self.assertFalse(mock_destroy.called)
  215 
  216     @mock.patch.object(db, 'notifications_get_all_by_filters')
  217     def test_get_notification_by_filters(self, mock_api_get):
  218         fake_db_notification2 = copy.deepcopy(fake_db_notification)
  219         fake_db_notification2['type'] = 'PROCESS'
  220         fake_db_notification2['id'] = 124
  221         fake_db_notification2[
  222             'notification_uuid'] = uuidsentinel.fake_db_notification2
  223 
  224         mock_api_get.return_value = [fake_db_notification2,
  225                                      fake_db_notification]
  226 
  227         filters = {'status': 'new'}
  228         notification_result = (notification.NotificationList.
  229                                get_all(self.context, filters=filters))
  230         self.assertEqual(2, len(notification_result))
  231         mock_api_get.assert_called_once_with(self.context, filters={
  232             'status': 'new'
  233         }, limit=None, marker=None, sort_dirs=None, sort_keys=None)
  234 
  235     @mock.patch.object(db, 'notifications_get_all_by_filters')
  236     def test_get_limit_and_marker_invalid_marker(self, mock_api_get):
  237         notification_uuid = uuidsentinel.fake_notification
  238         mock_api_get.side_effect = (exception.
  239                                     MarkerNotFound(marker=notification_uuid))
  240 
  241         self.assertRaises(exception.MarkerNotFound,
  242                           notification.NotificationList.get_all,
  243                           self.context, limit=5, marker=notification_uuid)
  244 
  245     @mock.patch.object(db, 'notification_update')
  246     def test_save(self, mock_notification_update):
  247 
  248         mock_notification_update.return_value = fake_db_notification
  249 
  250         notification_obj = self._notification_create_attributes()
  251         notification_obj.id = 123
  252         notification_obj.save()
  253 
  254         self.compare_obj(notification_obj, fake_object_notification,
  255                          allow_missing=OPTIONAL)
  256         (mock_notification_update.
  257          assert_called_once_with(self.context, uuidsentinel.fake_notification,
  258                                  {'source_host_uuid': uuidsentinel.fake_host,
  259                                   'notificati'
  260                                   'on_uuid': uuidsentinel.fake_notification,
  261                                   'status': 'new', 'generated_time': NOW,
  262                                   'payload': {'fake_key': u'fake_value'},
  263                                   'type': 'COMPUTE_HOST'}
  264                                  ))