"Fossies" - the Fresh Open Source Software Archive

Member "masakari-9.0.0/masakari/tests/unit/objects/test_hosts.py" (13 May 2020, 12394 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_hosts.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 
   21 from masakari.api import utils as api_utils
   22 from masakari import db
   23 from masakari import exception
   24 from masakari.objects import fields
   25 from masakari.objects import host
   26 from masakari.objects import segment
   27 from masakari.tests.unit.objects import test_objects
   28 from masakari.tests import uuidsentinel
   29 
   30 NOW = timeutils.utcnow().replace(microsecond=0)
   31 
   32 fake_segment_dict = {
   33     'name': 'fake_segment',
   34     'recovery_method': 'auto',
   35     'description': 'fake',
   36     'service_type': 'CINDER',
   37     'id': 123,
   38     'uuid': uuidsentinel.fake_segment,
   39     'created_at': NOW,
   40     'updated_at': None,
   41     'deleted_at': None,
   42     'deleted': False
   43 }
   44 
   45 
   46 def _fake_host(**kwargs):
   47     fake_host = {
   48         'created_at': NOW,
   49         'updated_at': None,
   50         'deleted_at': None,
   51         'deleted': False,
   52         'id': 123,
   53         'uuid': uuidsentinel.fake_host,
   54         'name': 'fake-host',
   55         'reserved': False,
   56         'on_maintenance': False,
   57         'control_attributes': 'fake_control_attributes',
   58         'type': 'SSH',
   59         'failover_segment': fake_segment_dict,
   60         'failover_segment_id': uuidsentinel.fake_segment,
   61         }
   62     fake_host.update(kwargs)
   63     return fake_host
   64 
   65 fake_host = _fake_host()
   66 
   67 
   68 class TestHostObject(test_objects._LocalTest):
   69 
   70     def _compare_segment_and_host_data(self, obj):
   71 
   72         self.compare_obj(obj.failover_segment, fake_segment_dict)
   73         self.assertEqual(obj.name, fake_host.get('name'))
   74         self.assertEqual(obj.reserved, fake_host.get('reserved'))
   75         self.assertEqual(obj.on_maintenance, fake_host.get('on_maintenance'))
   76         self.assertEqual(obj.type, fake_host.get('type'))
   77         self.assertEqual(obj.control_attributes, fake_host.get('control_attri'
   78                                                                'butes'))
   79         self.assertEqual(obj.id, 123)
   80 
   81     def _test_query(self, db_method, obj_method, *args, **kwargs):
   82         with mock.patch.object(db, db_method) as mock_db:
   83 
   84             db_exception = kwargs.pop('db_exception', None)
   85             if db_exception:
   86                 mock_db.side_effect = db_exception
   87             else:
   88                 mock_db.return_value = fake_host
   89 
   90             obj = getattr(host.Host, obj_method)(self.context, *args, **kwargs)
   91             if db_exception:
   92                 self.assertIsNone(obj)
   93 
   94             self._compare_segment_and_host_data(obj)
   95 
   96     def test_get_by_id(self):
   97         self._test_query('host_get_by_id', 'get_by_id', 123)
   98 
   99     def test_get_by_uuid(self):
  100         self._test_query('host_get_by_uuid', 'get_by_uuid',
  101                          uuidsentinel.fake_segment)
  102 
  103     def test_get_by_name(self):
  104         self._test_query('host_get_by_name', 'get_by_name',
  105                          'fake-host')
  106 
  107     def _host_create_attributes(self):
  108 
  109         host_obj = host.Host(context=self.context)
  110         host_obj.name = 'foo-host'
  111         host_obj.failover_segment_id = uuidsentinel.fake_segment
  112         host_obj.type = 'fake-type'
  113         host_obj.reserved = False
  114         host_obj.on_maintenance = False
  115         host_obj.control_attributes = 'fake_attributes'
  116         host_obj.uuid = uuidsentinel.fake_host
  117 
  118         return host_obj
  119 
  120     @mock.patch.object(api_utils, 'notify_about_host_api')
  121     @mock.patch.object(db, 'host_create')
  122     def test_create(self, mock_db_create, mock_notify_about_host_api):
  123 
  124         mock_db_create.return_value = fake_host
  125         host_obj = self._host_create_attributes()
  126         host_obj.create()
  127         self._compare_segment_and_host_data(host_obj)
  128         mock_db_create.assert_called_once_with(self.context, {
  129             'failover_segment_id': uuidsentinel.fake_segment,
  130             'on_maintenance': False, 'uuid': uuidsentinel.fake_host,
  131             'reserved': False, 'name': u'foo-host',
  132             'control_attributes': u'fake_attributes',
  133             'type': u'fake-type'})
  134         action = fields.EventNotificationAction.HOST_CREATE
  135         phase_start = fields.EventNotificationPhase.START
  136         phase_end = fields.EventNotificationPhase.END
  137         notify_calls = [
  138             mock.call(self.context, host_obj, action=action,
  139                       phase=phase_start),
  140             mock.call(self.context, host_obj, action=action,
  141                       phase=phase_end)]
  142         mock_notify_about_host_api.assert_has_calls(notify_calls)
  143 
  144     @mock.patch.object(api_utils, 'notify_about_host_api')
  145     @mock.patch.object(db, 'host_create')
  146     def test_recreate_fails(self, mock_host_create,
  147                             mock_notify_about_host_api):
  148         mock_host_create.return_value = fake_host
  149         host_obj = self._host_create_attributes()
  150         host_obj.create()
  151 
  152         self.assertRaises(exception.ObjectActionError, host_obj.create)
  153         action = fields.EventNotificationAction.HOST_CREATE
  154         phase_start = fields.EventNotificationPhase.START
  155         phase_end = fields.EventNotificationPhase.END
  156         notify_calls = [
  157             mock.call(self.context, host_obj, action=action,
  158                       phase=phase_start),
  159             mock.call(self.context, host_obj, action=action,
  160                       phase=phase_end)]
  161         mock_notify_about_host_api.assert_has_calls(notify_calls)
  162 
  163         mock_host_create.assert_called_once_with(self.context, {
  164             'uuid': uuidsentinel.fake_host, 'name': 'foo-host',
  165             'failover_segment_id': uuidsentinel.fake_segment,
  166             'type': 'fake-type', 'reserved': False, 'on_maintenance': False,
  167             'control_attributes': 'fake_attributes'})
  168 
  169     @mock.patch.object(api_utils, 'notify_about_host_api')
  170     @mock.patch.object(db, 'host_delete')
  171     def test_destroy(self, mock_host_destroy, mock_notify_about_host_api):
  172         host_obj = self._host_create_attributes()
  173         host_obj.id = 123
  174         host_obj.destroy()
  175 
  176         mock_host_destroy.assert_called_once_with(
  177             self.context, uuidsentinel.fake_host)
  178         action = fields.EventNotificationAction.HOST_DELETE
  179         phase_start = fields.EventNotificationPhase.START
  180         phase_end = fields.EventNotificationPhase.END
  181         notify_calls = [
  182             mock.call(self.context, host_obj, action=action,
  183                       phase=phase_start),
  184             mock.call(self.context, host_obj, action=action,
  185                       phase=phase_end)]
  186         mock_notify_about_host_api.assert_has_calls(notify_calls)
  187 
  188     @mock.patch.object(api_utils, 'notify_about_host_api')
  189     @mock.patch.object(db, 'host_delete')
  190     def test_destroy_host_not_found(self, mock_host_destroy,
  191                                     mock_notify_about_host_api):
  192         mock_host_destroy.side_effect = exception.HostNotFound(id=123)
  193         host_obj = self._host_create_attributes()
  194         host_obj.id = 123
  195         self.assertRaises(exception.HostNotFound, host_obj.destroy)
  196         action = fields.EventNotificationAction.HOST_DELETE
  197         phase_start = fields.EventNotificationPhase.START
  198         notify_calls = [
  199             mock.call(self.context, host_obj, action=action,
  200                       phase=phase_start)]
  201         mock_notify_about_host_api.assert_has_calls(notify_calls)
  202 
  203     @mock.patch.object(db, 'host_get_all_by_filters')
  204     def test_get_host_by_filters(self, mock_api_get):
  205         fake_host2 = copy.deepcopy(fake_host)
  206         fake_host2['name'] = 'fake_host22'
  207 
  208         mock_api_get.return_value = [fake_host2, fake_host]
  209 
  210         filters = {'reserved': False}
  211         host_result = host.HostList.get_all(self.context, filters=filters)
  212         self.assertEqual(2, len(host_result))
  213         mock_api_get.assert_called_once_with(self.context, filters={
  214             'reserved': False
  215         }, limit=None, marker=None, sort_dirs=None, sort_keys=None)
  216 
  217     @mock.patch.object(db, 'host_get_all_by_filters')
  218     def test_get_limit_and_marker_invalid_marker(self, mock_api_get):
  219         host_name = 'fake-host'
  220         mock_api_get.side_effect = exception.MarkerNotFound(marker=host_name)
  221 
  222         self.assertRaises(exception.MarkerNotFound,
  223                           host.HostList.get_all,
  224                           self.context, limit=5, marker=host_name)
  225 
  226     @mock.patch.object(api_utils, 'notify_about_host_api')
  227     @mock.patch.object(db, 'host_update')
  228     def test_save(self, mock_host_update, mock_notify_about_host_api):
  229 
  230         mock_host_update.return_value = fake_host
  231 
  232         host_obj = self._host_create_attributes()
  233         host_obj.id = 123
  234         host_obj.save()
  235         self._compare_segment_and_host_data(host_obj)
  236         (mock_host_update.
  237          assert_called_once_with(self.context, uuidsentinel.fake_host,
  238                                  {'control_attributes': u'fake_attributes',
  239                                   'type': u'fake-type',
  240                                   'failover_segment_id':
  241                                       uuidsentinel.fake_segment,
  242                                   'name': u'foo-host',
  243                                   'uuid': uuidsentinel.fake_host,
  244                                   'reserved': False, 'on_maintenance': False
  245                                   }))
  246         action = fields.EventNotificationAction.HOST_UPDATE
  247         phase_start = fields.EventNotificationPhase.START
  248         phase_end = fields.EventNotificationPhase.END
  249         notify_calls = [
  250             mock.call(self.context, host_obj, action=action,
  251                       phase=phase_start),
  252             mock.call(self.context, host_obj, action=action,
  253                       phase=phase_end)]
  254         mock_notify_about_host_api.assert_has_calls(notify_calls)
  255 
  256     @mock.patch.object(db, 'host_update')
  257     def test_save_lazy_attribute_changed(self, mock_host_update):
  258 
  259         mock_host_update.return_value = fake_host
  260 
  261         host_obj = self._host_create_attributes()
  262         host_obj.failover_segment = (segment.
  263                                      FailoverSegment(context=self.context))
  264         host_obj.id = 123
  265         self.assertRaises(exception.ObjectActionError, host_obj.save)
  266 
  267     @mock.patch.object(api_utils, 'notify_about_host_api')
  268     @mock.patch.object(db, 'host_update')
  269     def test_save_host_already_exists(self, mock_host_update,
  270                                       mock_notify_about_host_api):
  271 
  272         mock_host_update.side_effect = exception.HostExists(name="foo-host")
  273 
  274         host_object = host.Host(context=self.context)
  275         host_object.name = "foo-host"
  276         host_object.id = 123
  277         host_object.uuid = uuidsentinel.fake_host
  278 
  279         self.assertRaises(exception.HostExists, host_object.save)
  280         action = fields.EventNotificationAction.HOST_UPDATE
  281         phase_start = fields.EventNotificationPhase.START
  282         notify_calls = [
  283             mock.call(self.context, host_object, action=action,
  284                       phase=phase_start)]
  285         mock_notify_about_host_api.assert_has_calls(notify_calls)
  286 
  287     @mock.patch.object(api_utils, 'notify_about_host_api')
  288     @mock.patch.object(db, 'host_update')
  289     def test_save_host_not_found(self, mock_host_update,
  290                                  mock_notify_about_host_api):
  291 
  292         mock_host_update.side_effect = exception.HostNotFound(id="foo-host")
  293 
  294         host_object = host.Host(context=self.context)
  295         host_object.name = "foo-host"
  296         host_object.id = 123
  297         host_object.uuid = uuidsentinel.fake_host
  298 
  299         self.assertRaises(exception.HostNotFound, host_object.save)
  300         action = fields.EventNotificationAction.HOST_UPDATE
  301         phase_start = fields.EventNotificationPhase.START
  302         notify_calls = [
  303             mock.call(self.context, host_object, action=action,
  304                       phase=phase_start)]
  305         mock_notify_about_host_api.assert_has_calls(notify_calls)