"Fossies" - the Fresh Open Source Software Archive

Member "cloudkitty-13.0.0/cloudkitty/tests/storage/v1/test_hybrid_storage.py" (14 Oct 2020, 4610 Bytes) of package /linux/misc/openstack/cloudkitty-13.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_hybrid_storage.py": 12.1.0_vs_13.0.0.

    1 # -*- coding: utf-8 -*-
    2 # Copyright 2017 Objectif Libre
    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 from unittest import mock
   17 
   18 from gnocchiclient import exceptions as gexc
   19 
   20 from cloudkitty import storage
   21 from cloudkitty import tests
   22 from cloudkitty.tests import utils as test_utils
   23 
   24 
   25 class BaseHybridStorageTest(tests.TestCase):
   26 
   27     @mock.patch('cloudkitty.utils.load_conf', new=test_utils.load_conf)
   28     def setUp(self):
   29         super(BaseHybridStorageTest, self).setUp()
   30         self.conf.set_override('backend', 'hybrid', 'storage')
   31         self.conf.set_override('version', '1', 'storage')
   32         self.storage = storage.get_storage(conf=test_utils.load_conf())
   33         with mock.patch.object(
   34                 self.storage.storage._hybrid_backend, 'init'):
   35             self.storage.init()
   36 
   37 
   38 class PermissiveDict(object):
   39     """Allows to check a single key of a dict in an assertion.
   40 
   41     Example:
   42     >>> mydict = {'a': 'A', 'b': 'B'}
   43     >>> checker = PermissiveDict('A', key='a')
   44     >>> checker == mydict
   45     True
   46     """
   47     def __init__(self, value, key='name'):
   48         self.key = key
   49         self.value = value
   50 
   51     def __eq__(self, other):
   52         return self.value == other.get(self.key)
   53 
   54 
   55 class HybridStorageTestGnocchi(BaseHybridStorageTest):
   56 
   57     def setUp(self):
   58         super(HybridStorageTestGnocchi, self).setUp()
   59 
   60     def tearDown(self):
   61         super(HybridStorageTestGnocchi, self).tearDown()
   62 
   63     def _init_storage(self, archive_policy=False, res_type=False):
   64         with mock.patch.object(self.storage.storage._hybrid_backend._conn,
   65                                'archive_policy',
   66                                spec=['get', 'create']) as pol_mock:
   67             if not archive_policy:
   68                 pol_mock.get.side_effect = gexc.ArchivePolicyNotFound
   69             else:
   70                 pol_mock.create.side_effect = gexc.ArchivePolicyAlreadyExists
   71             with mock.patch.object(self.storage.storage._hybrid_backend._conn,
   72                                    'resource_type',
   73                                    spec=['get', 'create']) as rtype_mock:
   74                 if not res_type:
   75                     rtype_mock.get.side_effect = gexc.ResourceTypeNotFound
   76                 else:
   77                     rtype_mock.create.side_effect \
   78                         = gexc.ResourceTypeAlreadyExists
   79 
   80                 self.storage.init()
   81                 rtype_data = (self.storage.storage
   82                               ._hybrid_backend._resource_type_data)
   83                 rtype_calls = list()
   84                 for val in rtype_data.values():
   85                     rtype_calls.append(
   86                         mock.call(PermissiveDict(val['name'], key='name')))
   87                 if res_type:
   88                     rtype_mock.create.assert_not_called()
   89                 else:
   90                     rtype_mock.create.assert_has_calls(
   91                         rtype_calls, any_order=True)
   92             pol_mock.get.assert_called_once_with(
   93                 self.storage.storage._hybrid_backend._archive_policy_name)
   94             if archive_policy:
   95                 pol_mock.create.assert_not_called()
   96             else:
   97                 apolicy = {
   98                     'name': (self.storage.storage
   99                              ._hybrid_backend._archive_policy_name),
  100                     'back_window': 0,
  101                     'aggregation_methods':
  102                         ['std', 'count', 'min', 'max', 'sum', 'mean'],
  103                 }
  104                 apolicy['definition'] = (self.storage.storage
  105                                          ._hybrid_backend
  106                                          ._archive_policy_definition)
  107                 pol_mock.create.assert_called_once_with(apolicy)
  108 
  109     def test_init_no_res_type_no_policy(self):
  110         self._init_storage()
  111 
  112     def test_init_with_res_type_no_policy(self):
  113         self._init_storage(res_type=True)
  114 
  115     def test_init_no_res_type_with_policy(self):
  116         self._init_storage(archive_policy=True)
  117 
  118     def test_init_with_res_type_with_policy(self):
  119         self._init_storage(res_type=True, archive_policy=True)