"Fossies" - the Fresh Open Source Software Archive

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