"Fossies" - the Fresh Open Source Software Archive

Member "cinder-13.0.7/cinder/tests/unit/objects/test_service.py" (4 Oct 2019, 10819 Bytes) of package /linux/misc/openstack/cinder-13.0.7.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.

    1 #    Copyright 2015 Intel Corp.
    2 #
    3 #    Licensed under the Apache License, Version 2.0 (the "License"); you may
    4 #    not use this file except in compliance with the License. You may obtain
    5 #    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, WITHOUT
   11 #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
   12 #    License for the specific language governing permissions and limitations
   13 #    under the License.
   14 
   15 import datetime
   16 import ddt
   17 import mock
   18 from oslo_utils import timeutils
   19 import pytz
   20 import six
   21 
   22 from cinder import exception
   23 from cinder import objects
   24 from cinder.tests.unit import fake_cluster
   25 from cinder.tests.unit import fake_service
   26 from cinder.tests.unit import objects as test_objects
   27 
   28 
   29 @ddt.ddt
   30 class TestService(test_objects.BaseObjectsTestCase):
   31 
   32     @mock.patch('cinder.db.sqlalchemy.api.service_get')
   33     def test_get_by_id(self, service_get):
   34         db_service = fake_service.fake_db_service()
   35         service_get.return_value = db_service
   36         service = objects.Service.get_by_id(self.context, 1)
   37         self._compare(self, db_service, service)
   38         service_get.assert_called_once_with(self.context, 1)
   39 
   40     @ddt.data(True, False)
   41     @mock.patch('cinder.db.service_get')
   42     def test_get_by_host_and_topic(self, show_disabled, service_get):
   43         db_service = fake_service.fake_db_service()
   44         service_get.return_value = db_service
   45         service = objects.Service.get_by_host_and_topic(
   46             self.context, 'fake-host', 'fake-topic', disabled=show_disabled)
   47         self._compare(self, db_service, service)
   48         service_get.assert_called_once_with(
   49             self.context, disabled=show_disabled, host='fake-host',
   50             topic='fake-topic')
   51 
   52     @mock.patch('cinder.db.service_get')
   53     def test_get_by_args(self, service_get):
   54         db_service = fake_service.fake_db_service()
   55         service_get.return_value = db_service
   56         service = objects.Service.get_by_args(
   57             self.context, 'fake-host', 'fake-key')
   58         self._compare(self, db_service, service)
   59         service_get.assert_called_once_with(
   60             self.context, host='fake-host', binary='fake-key')
   61 
   62     @mock.patch('cinder.db.service_create')
   63     def test_create(self, service_create):
   64         db_service = fake_service.fake_db_service()
   65         service_create.return_value = db_service
   66         service = objects.Service(context=self.context)
   67         service.create()
   68         self.assertEqual(db_service['id'], service.id)
   69         service_create.assert_called_once_with(self.context,
   70                                                {'uuid': mock.ANY})
   71 
   72     @mock.patch('cinder.db.service_update')
   73     def test_save(self, service_update):
   74         db_service = fake_service.fake_db_service()
   75         service = objects.Service._from_db_object(
   76             self.context, objects.Service(), db_service)
   77         service.topic = 'foobar'
   78         service.save()
   79         service_update.assert_called_once_with(self.context, service.id,
   80                                                {'topic': 'foobar'})
   81 
   82     @mock.patch('oslo_utils.timeutils.utcnow', return_value=timeutils.utcnow())
   83     @mock.patch('cinder.db.sqlalchemy.api.service_destroy')
   84     def test_destroy(self, service_destroy, utcnow_mock):
   85         service_destroy.return_value = {
   86             'deleted': True,
   87             'deleted_at': utcnow_mock.return_value}
   88         db_service = fake_service.fake_db_service()
   89         service = objects.Service._from_db_object(
   90             self.context, objects.Service(), db_service)
   91         with mock.patch.object(service._context, 'elevated') as elevated_ctx:
   92             service.destroy()
   93             service_destroy.assert_called_once_with(elevated_ctx(), 123)
   94         self.assertTrue(service.deleted)
   95         self.assertEqual(utcnow_mock.return_value.replace(tzinfo=pytz.UTC),
   96                          service.deleted_at)
   97 
   98     @mock.patch('cinder.db.sqlalchemy.api.service_get')
   99     def test_refresh(self, service_get):
  100         db_service1 = fake_service.fake_db_service()
  101         db_service2 = db_service1.copy()
  102         db_service2['availability_zone'] = 'foobar'
  103 
  104         # On the second service_get, return the service with an updated
  105         # availability_zone
  106         service_get.side_effect = [db_service1, db_service2]
  107         service = objects.Service.get_by_id(self.context, 123)
  108         self._compare(self, db_service1, service)
  109 
  110         # availability_zone was updated, so a service refresh should have a
  111         # new value for that field
  112         service.refresh()
  113         self._compare(self, db_service2, service)
  114         if six.PY3:
  115             call_bool = mock.call.__bool__()
  116         else:
  117             call_bool = mock.call.__nonzero__()
  118         service_get.assert_has_calls([mock.call(self.context, 123),
  119                                       call_bool,
  120                                       mock.call(self.context, 123)])
  121 
  122     @mock.patch('cinder.db.service_get_all')
  123     def test_get_minimum_version(self, service_get_all):
  124         services_update = [
  125             {'rpc_current_version': '1.0', 'object_current_version': '1.3'},
  126             {'rpc_current_version': '1.1', 'object_current_version': '1.2'},
  127             {'rpc_current_version': '2.0', 'object_current_version': '2.5'},
  128         ]
  129         expected = ('1.0', '1.2')
  130         services = [fake_service.fake_db_service(**s) for s in services_update]
  131         service_get_all.return_value = services
  132 
  133         min_rpc = objects.Service.get_minimum_rpc_version(self.context, 'foo')
  134         self.assertEqual(expected[0], min_rpc)
  135         min_obj = objects.Service.get_minimum_obj_version(self.context, 'foo')
  136         self.assertEqual(expected[1], min_obj)
  137         service_get_all.assert_has_calls(
  138             [mock.call(self.context, binary='foo', disabled=None)] * 2)
  139 
  140     @mock.patch('cinder.db.service_get_all')
  141     def test_get_minimum_version_liberty(self, service_get_all):
  142         services_update = [
  143             {'rpc_current_version': '1.0', 'object_current_version': '1.3'},
  144             {'rpc_current_version': '1.1', 'object_current_version': None},
  145             {'rpc_current_version': None, 'object_current_version': '2.5'},
  146         ]
  147         services = [fake_service.fake_db_service(**s) for s in services_update]
  148         service_get_all.return_value = services
  149 
  150         self.assertRaises(exception.ServiceTooOld,
  151                           objects.Service.get_minimum_rpc_version,
  152                           self.context, 'foo')
  153         self.assertRaises(exception.ServiceTooOld,
  154                           objects.Service.get_minimum_obj_version,
  155                           self.context, 'foo')
  156 
  157     @mock.patch('cinder.db.service_get_all')
  158     def test_get_minimum_version_no_binary(self, service_get_all):
  159         services_update = [
  160             {'rpc_current_version': '1.0', 'object_current_version': '1.3'},
  161             {'rpc_current_version': '1.1', 'object_current_version': '1.2'},
  162             {'rpc_current_version': '2.0', 'object_current_version': '2.5'},
  163         ]
  164         services = [fake_service.fake_db_service(**s) for s in services_update]
  165         service_get_all.return_value = services
  166 
  167         min_obj = objects.Service.get_minimum_obj_version(self.context)
  168         self.assertEqual('1.2', min_obj)
  169         service_get_all.assert_called_once_with(self.context, binary=None,
  170                                                 disabled=None)
  171 
  172     @mock.patch('cinder.db.sqlalchemy.api.cluster_get')
  173     def test_lazy_loading_cluster_field(self, cluster_get):
  174         cluster_orm = fake_cluster.fake_cluster_orm(name='mycluster')
  175         cluster_get.return_value = cluster_orm
  176         cluster = objects.Cluster._from_db_object(self.context,
  177                                                   objects.Cluster(),
  178                                                   cluster_orm)
  179 
  180         service = fake_service.fake_service_obj(self.context,
  181                                                 cluster_name='mycluster')
  182         self.assertEqual(cluster, service.cluster)
  183         cluster_get.assert_called_once_with(self.context, None,
  184                                             name='mycluster')
  185 
  186     def test_service_is_up(self):
  187         # NOTE(mdovgal): don't use @ddt.data with the real timestamp value
  188         # for this test.
  189         # When using ddt decorators ddt.data seems to have been calculated
  190         # not at the time of test's execution but at the tests's beginning.
  191         # And this one depends on utcnow func. So it won't be utcnow at the
  192         # execution moment and the result will be unexpected.
  193         down_time = 5
  194         self.flags(service_down_time=down_time)
  195 
  196         # test if service is up
  197         service = fake_service.fake_service_obj(self.context)
  198         self.assertTrue(service.is_up)
  199 
  200         service.updated_at = timeutils.utcnow()
  201         self.assertTrue(service.is_up)
  202 
  203         # test is service is down now
  204         past_time = timeutils.utcnow() - datetime.timedelta(seconds=64)
  205         service.updated_at = past_time
  206         self.assertFalse(service.is_up)
  207 
  208 
  209 class TestServiceList(test_objects.BaseObjectsTestCase):
  210     @mock.patch('cinder.db.service_get_all')
  211     def test_get_all(self, service_get_all):
  212         db_service = fake_service.fake_db_service()
  213         service_get_all.return_value = [db_service]
  214 
  215         filters = {'host': 'host', 'binary': 'foo', 'disabled': False}
  216         services = objects.ServiceList.get_all(self.context, filters)
  217         service_get_all.assert_called_once_with(self.context, **filters)
  218         self.assertEqual(1, len(services))
  219         TestService._compare(self, db_service, services[0])
  220 
  221     @mock.patch('cinder.db.service_get_all')
  222     def test_get_all_by_topic(self, service_get_all):
  223         db_service = fake_service.fake_db_service()
  224         service_get_all.return_value = [db_service]
  225 
  226         services = objects.ServiceList.get_all_by_topic(
  227             self.context, 'foo', 'bar')
  228         service_get_all.assert_called_once_with(
  229             self.context, topic='foo', disabled='bar')
  230         self.assertEqual(1, len(services))
  231         TestService._compare(self, db_service, services[0])
  232 
  233     @mock.patch('cinder.db.service_get_all')
  234     def test_get_all_by_binary(self, service_get_all):
  235         db_service = fake_service.fake_db_service()
  236         service_get_all.return_value = [db_service]
  237 
  238         services = objects.ServiceList.get_all_by_binary(
  239             self.context, 'foo', 'bar')
  240         service_get_all.assert_called_once_with(
  241             self.context, binary='foo', disabled='bar')
  242         self.assertEqual(1, len(services))
  243         TestService._compare(self, db_service, services[0])