"Fossies" - the Fresh Open Source Software Archive

Member "cloudkitty-13.0.0/cloudkitty/tests/gabbi/fixtures.py" (14 Oct 2020, 17358 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 "fixtures.py": 12.1.0_vs_13.0.0.

    1 # -*- coding: utf-8 -*-
    2 # Copyright 2015 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 import abc
   17 import collections
   18 import datetime
   19 import decimal
   20 import os
   21 from unittest import mock
   22 
   23 from dateutil import tz
   24 from gabbi import fixture
   25 from oslo_config import cfg
   26 from oslo_config import fixture as conf_fixture
   27 from oslo_db.sqlalchemy import utils
   28 import oslo_messaging
   29 from oslo_messaging import conffixture
   30 from oslo_policy import opts as policy_opts
   31 import six
   32 from stevedore import driver
   33 from stevedore import extension
   34 import webob.dec
   35 from wsme import types as wtypes
   36 import wsmeext.pecan as wsme_pecan
   37 
   38 from cloudkitty.api import app
   39 from cloudkitty.api import middleware
   40 from cloudkitty.api.v2.dataframes import dataframes as v2_api_dataframes
   41 from cloudkitty.api.v2.summary import summary as v2_api_summary
   42 from cloudkitty import dataframe
   43 from cloudkitty import db
   44 from cloudkitty.db import api as ck_db_api
   45 from cloudkitty import messaging
   46 from cloudkitty import rating
   47 from cloudkitty import storage
   48 from cloudkitty.storage.v1.sqlalchemy import models
   49 from cloudkitty import storage_state
   50 from cloudkitty import tests
   51 from cloudkitty.tests.storage.v2 import influx_utils
   52 from cloudkitty.tests import utils as test_utils
   53 from cloudkitty import utils as ck_utils
   54 from cloudkitty.utils import tz as tzutils
   55 
   56 
   57 INITIAL_DT = datetime.datetime(2015, 1, 1, tzinfo=tz.tzutc())
   58 
   59 
   60 class UUIDFixture(fixture.GabbiFixture):
   61     def start_fixture(self):
   62         FAKE_UUID = '6c1b8a30-797f-4b7e-ad66-9879b79059fb'
   63         patcher = mock.patch(
   64             'oslo_utils.uuidutils.generate_uuid',
   65             return_value=FAKE_UUID)
   66         patcher.start()
   67         self.patcher = patcher
   68 
   69     def stop_fixture(self):
   70         self.patcher.stop()
   71 
   72 
   73 @six.add_metaclass(abc.ABCMeta)
   74 class BaseExtensionFixture(fixture.GabbiFixture):
   75     klass = None
   76     namespace = None
   77     stevedore_mgr = None
   78     assert_args = {}
   79 
   80     @abc.abstractmethod
   81     def setup_fake_modules(self):
   82         pass
   83 
   84     def start_fixture(self):
   85         fake_extensions = self.setup_fake_modules()
   86         self.mock = mock.patch(self.klass)
   87         fake_mgr = self.stevedore_mgr.make_test_instance(
   88             fake_extensions,
   89             self.namespace)
   90         self.patch = self.mock.start()
   91         self.patch.return_value = fake_mgr
   92 
   93     def stop_fixture(self):
   94         self.patch.assert_called_with(
   95             self.namespace,
   96             **self.assert_args)
   97         self.mock.stop()
   98 
   99 
  100 class CollectorExtensionsFixture(BaseExtensionFixture):
  101     klass = 'stevedore.driver.DriverManager'
  102     namespace = 'cloudkitty.collector.backends'
  103     stevedore_mgr = driver.DriverManager
  104     assert_args = {
  105         'invoke_kwds': {'period': 3600},
  106         'invoke_on_load': True}
  107 
  108     def setup_fake_modules(self):
  109         def fake_metric(start,
  110                         end=None,
  111                         project_id=None,
  112                         q_filter=None):
  113             return None
  114 
  115         fake_module1 = tests.FakeCollectorModule()
  116         fake_module1.collector_name = 'fake1'
  117         fake_module1.get_compute = fake_metric
  118         fake_module2 = tests.FakeCollectorModule()
  119         fake_module2.collector_name = 'fake2'
  120         fake_module2.get_volume = fake_metric
  121         fake_module3 = tests.FakeCollectorModule()
  122         fake_module3.collector_name = 'fake3'
  123         fake_module3.get_compute = fake_metric
  124         fake_extensions = [
  125             extension.Extension(
  126                 'fake1',
  127                 'cloudkitty.tests.FakeCollectorModule1',
  128                 None,
  129                 fake_module1),
  130             extension.Extension(
  131                 'fake2',
  132                 'cloudkitty.tests.FakeCollectorModule2',
  133                 None,
  134                 fake_module2),
  135             extension.Extension(
  136                 'fake3',
  137                 'cloudkitty.tests.FakeCollectorModule3',
  138                 None,
  139                 fake_module3)]
  140         return fake_extensions[0]
  141 
  142 
  143 class RatingModulesFixture(BaseExtensionFixture):
  144     klass = 'stevedore.extension.ExtensionManager'
  145     namespace = 'cloudkitty.rating.processors'
  146     stevedore_mgr = extension.ExtensionManager
  147     assert_args = {
  148         'invoke_on_load': True}
  149 
  150     def setup_fake_modules(self):
  151         class FakeConfigController(rating.RatingRestControllerBase):
  152             _custom_actions = {
  153                 'test': ['GET']
  154             }
  155 
  156             @wsme_pecan.wsexpose(wtypes.text)
  157             def get_test(self):
  158                 """Return the list of every mapping type available.
  159 
  160                 """
  161                 return 'OK'
  162 
  163         fake_module1 = tests.FakeRatingModule()
  164         fake_module1.module_name = 'fake1'
  165         fake_module1.set_priority(3)
  166         fake_module2 = tests.FakeRatingModule()
  167         fake_module2.module_name = 'fake2'
  168         fake_module2.config_controller = FakeConfigController
  169         fake_module2.set_priority(1)
  170         fake_module3 = tests.FakeRatingModule()
  171         fake_module3.module_name = 'fake3'
  172         fake_module3.set_priority(2)
  173         fake_extensions = [
  174             extension.Extension(
  175                 'fake1',
  176                 'cloudkitty.tests.FakeRatingModule1',
  177                 None,
  178                 fake_module1),
  179             extension.Extension(
  180                 'fake2',
  181                 'cloudkitty.tests.FakeRatingModule2',
  182                 None,
  183                 fake_module2),
  184             extension.Extension(
  185                 'fake3',
  186                 'cloudkitty.tests.FakeRatingModule3',
  187                 None,
  188                 fake_module3)]
  189         return fake_extensions
  190 
  191 
  192 class ConfigFixture(fixture.GabbiFixture):
  193     auth_strategy = 'noauth'
  194 
  195     def start_fixture(self):
  196         self.conf = None
  197         conf = conf_fixture.Config().conf
  198         policy_opts.set_defaults(conf)
  199         msg_conf = conffixture.ConfFixture(conf)
  200         msg_conf.transport_url = 'fake:/'
  201         conf.import_group('api', 'cloudkitty.api.app')
  202         conf.set_override('auth_strategy', self.auth_strategy)
  203         conf.set_override('connection', 'sqlite:///', 'database')
  204         conf.set_override('policy_file',
  205                           os.path.abspath('etc/cloudkitty/policy.json'),
  206                           group='oslo_policy')
  207         conf.set_override('api_paste_config',
  208                           os.path.abspath(
  209                               'cloudkitty/tests/gabbi/gabbi_paste.ini')
  210                           )
  211         conf.import_group('storage', 'cloudkitty.storage')
  212         conf.set_override('backend', 'sqlalchemy', 'storage')
  213         conf.set_override('version', '1', 'storage')
  214         self.conf = conf
  215         self.conn = ck_db_api.get_instance()
  216         migration = self.conn.get_migration()
  217         migration.upgrade('head')
  218 
  219     def stop_fixture(self):
  220         if self.conf:
  221             self.conf.reset()
  222         db.get_engine().dispose()
  223 
  224 
  225 class ConfigFixtureStorageV2(ConfigFixture):
  226 
  227     def start_fixture(self):
  228         super(ConfigFixtureStorageV2, self).start_fixture()
  229         self.conf.set_override('backend', 'influxdb', 'storage')
  230         self.conf.set_override('version', '2', 'storage')
  231 
  232 
  233 class ConfigFixtureKeystoneAuth(ConfigFixture):
  234     auth_strategy = 'keystone'
  235 
  236     def start_fixture(self):
  237         # Mocking the middleware process_request which check for credentials
  238         # here, the only check done is that the hardcoded token is the one
  239         # send by the query. If not, 401, else 200.
  240         def _mock_proc_request(self, request):
  241             token = 'c93e3e31342e4e32ba201fd3d70878b5'
  242             http_code = 401
  243             if 'X-Auth-Token' in request.headers and \
  244                request.headers['X-Auth-Token'] == token:
  245                 http_code = 200
  246 
  247             return webob.Response(
  248                 status_code=http_code,
  249                 content_type='application/json'
  250             )
  251 
  252         self._orig_func = middleware.auth_token.AuthProtocol.process_request
  253         middleware.auth_token.AuthProtocol.process_request = _mock_proc_request
  254 
  255         super(ConfigFixtureKeystoneAuth, self).start_fixture()
  256 
  257     def stop_fixture(self):
  258         super(ConfigFixtureKeystoneAuth, self).stop_fixture()
  259         middleware.auth_token.AuthProtocol.process_request = self._orig_func
  260 
  261 
  262 class BaseFakeRPC(fixture.GabbiFixture):
  263     endpoint = None
  264 
  265     def start_fixture(self):
  266         messaging.setup()
  267         target = oslo_messaging.Target(topic='cloudkitty',
  268                                        server=cfg.CONF.host,
  269                                        version='1.0')
  270         endpoints = [
  271             self.endpoint()
  272         ]
  273         self.server = messaging.get_server(target, endpoints)
  274         self.server.start()
  275 
  276     def stop_fixture(self):
  277         self.server.stop()
  278 
  279 
  280 class ScopeStateResetFakeRPC(BaseFakeRPC):
  281     class FakeRPCEndpoint(object):
  282         target = oslo_messaging.Target(version='1.0')
  283 
  284         def reset_state(self, ctxt, res_data):
  285             pass
  286 
  287     endpoint = FakeRPCEndpoint
  288 
  289 
  290 class QuoteFakeRPC(BaseFakeRPC):
  291     class FakeRPCEndpoint(object):
  292         target = oslo_messaging.Target(namespace='rating',
  293                                        version='1.0')
  294 
  295         def quote(self, ctxt, res_data):
  296             return str(1.0)
  297 
  298     endpoint = FakeRPCEndpoint
  299 
  300 
  301 class BaseStorageDataFixture(fixture.GabbiFixture):
  302     def create_fake_data(self, begin, end, project_id):
  303 
  304         cpu_point = dataframe.DataPoint(
  305             unit="nothing",
  306             qty=1,
  307             groupby={"fake_meta": 1.0, "project_id": project_id},
  308             metadata={"dummy": True},
  309             price=decimal.Decimal('1.337'),
  310         )
  311         image_point = dataframe.DataPoint(
  312             unit="nothing",
  313             qty=1,
  314             groupby={"fake_meta": 1.0, "project_id": project_id},
  315             metadata={"dummy": True},
  316             price=decimal.Decimal('0.121'),
  317         )
  318         data = [
  319             dataframe.DataFrame(
  320                 start=begin, end=end,
  321                 usage=collections.OrderedDict({"cpu": [cpu_point, cpu_point]}),
  322             ),
  323             dataframe.DataFrame(
  324                 start=begin, end=end,
  325                 usage=collections.OrderedDict(
  326                     {"image.size": [image_point, image_point]}),
  327             ),
  328         ]
  329         return data
  330 
  331     def start_fixture(self):
  332         auth = mock.patch(
  333             'keystoneauth1.loading.load_auth_from_conf_options',
  334             return_value=dict())
  335         session = mock.patch(
  336             'keystoneauth1.loading.load_session_from_conf_options',
  337             return_value=dict())
  338         with auth:
  339             with session:
  340                 self.storage = storage.get_storage(conf=test_utils.load_conf())
  341         self.storage.init()
  342         self.initialize_data()
  343 
  344     def stop_fixture(self):
  345         model = models.RatedDataFrame
  346         session = db.get_session()
  347         q = utils.model_query(
  348             model,
  349             session)
  350         q.delete()
  351 
  352 
  353 class StorageDataFixture(BaseStorageDataFixture):
  354     def initialize_data(self):
  355         nodata_duration = (24 * 3 + 12) * 3600
  356         hour_delta = datetime.timedelta(seconds=3600)
  357         tenant_list = ['8f82cc70-e50c-466e-8624-24bdea811375',
  358                        '7606a24a-b8ad-4ae0-be6c-3d7a41334a2e']
  359         data_dt = INITIAL_DT + datetime.timedelta(
  360             seconds=nodata_duration + 3600)
  361         data_duration = datetime.timedelta(seconds=(24 * 2 + 8) * 3600)
  362 
  363         iter_dt = data_dt
  364         while iter_dt < data_dt + data_duration:
  365             data = self.create_fake_data(
  366                 iter_dt, iter_dt + hour_delta, tenant_list[0])
  367             self.storage.push(data, tenant_list[0])
  368             iter_dt += hour_delta
  369 
  370         iter_dt = data_dt
  371         while iter_dt < data_dt + data_duration / 2:
  372             data = self.create_fake_data(
  373                 iter_dt, iter_dt + hour_delta, tenant_list[1])
  374             self.storage.push(data, tenant_list[1])
  375             iter_dt += hour_delta
  376 
  377 
  378 class NowStorageDataFixture(BaseStorageDataFixture):
  379     def initialize_data(self):
  380         dt = tzutils.get_month_start(naive=True).replace(tzinfo=tz.tzutc())
  381         hour_delta = datetime.timedelta(seconds=3600)
  382         limit = dt + hour_delta * 12
  383         while dt < limit:
  384             project_id = '3d9a1b33-482f-42fd-aef9-b575a3da9369'
  385             data = self.create_fake_data(dt, dt + hour_delta, project_id)
  386             self.storage.push(data, project_id)
  387             dt += hour_delta
  388 
  389 
  390 class ScopeStateFixture(fixture.GabbiFixture):
  391 
  392     def start_fixture(self):
  393         self.sm = storage_state.StateManager()
  394         self.sm.init()
  395         data = [
  396             ('aaaa', datetime.datetime(2019, 1, 1), 'fet1', 'col1', 'key1'),
  397             ('bbbb', datetime.datetime(2019, 2, 2), 'fet1', 'col1', 'key2'),
  398             ('cccc', datetime.datetime(2019, 3, 3), 'fet1', 'col2', 'key1'),
  399             ('dddd', datetime.datetime(2019, 4, 4), 'fet1', 'col2', 'key2'),
  400             ('eeee', datetime.datetime(2019, 5, 5), 'fet2', 'col1', 'key1'),
  401             ('ffff', datetime.datetime(2019, 6, 6), 'fet2', 'col1', 'key2'),
  402             ('gggg', datetime.datetime(2019, 6, 6), 'fet2', 'col2', 'key1'),
  403             ('hhhh', datetime.datetime(2019, 6, 6), 'fet2', 'col2', 'key2'),
  404         ]
  405         for d in data:
  406             self.sm.set_state(
  407                 d[0], d[1], fetcher=d[2], collector=d[3], scope_key=d[4])
  408 
  409     def stop_fixture(self):
  410         session = db.get_session()
  411         q = utils.model_query(
  412             self.sm.model,
  413             session)
  414         q.delete()
  415 
  416 
  417 class CORSConfigFixture(fixture.GabbiFixture):
  418     """Inject mock configuration for the CORS middleware."""
  419 
  420     def start_fixture(self):
  421         # Here we monkeypatch GroupAttr.__getattr__, necessary because the
  422         # paste.ini method of initializing this middleware creates its own
  423         # ConfigOpts instance, bypassing the regular config fixture.
  424 
  425         def _mock_getattr(instance, key):
  426             if key != 'allowed_origin':
  427                 return self._original_call_method(instance, key)
  428             return "http://valid.example.com"
  429 
  430         self._original_call_method = cfg.ConfigOpts.GroupAttr.__getattr__
  431         cfg.ConfigOpts.GroupAttr.__getattr__ = _mock_getattr
  432 
  433     def stop_fixture(self):
  434         """Remove the monkeypatch."""
  435         cfg.ConfigOpts.GroupAttr.__getattr__ = self._original_call_method
  436 
  437 
  438 class MetricsConfFixture(fixture.GabbiFixture):
  439     """Inject Metrics configuration mock to the get_metrics_conf() function"""
  440 
  441     def start_fixture(self):
  442         self._original_function = ck_utils.load_conf
  443         ck_utils.load_conf = mock.Mock(
  444             return_value=tests.samples.METRICS_CONF,
  445         )
  446 
  447     def stop_fixture(self):
  448         """Remove the get_metrics_conf() monkeypatch."""
  449         ck_utils.load_conf = self._original_function
  450 
  451 
  452 class NowInfluxStorageDataFixture(NowStorageDataFixture):
  453 
  454     def start_fixture(self):
  455         cli = influx_utils.FakeInfluxClient()
  456         st = storage.get_storage()
  457         st._conn = cli
  458 
  459         self._get_storage_patch = mock.patch(
  460             'cloudkitty.storage.get_storage',
  461             new=lambda **kw: st,
  462         )
  463         self._get_storage_patch.start()
  464         v2_api_summary.Summary.reload()
  465         v2_api_dataframes.DataFrameList.reload()
  466 
  467         super(NowInfluxStorageDataFixture, self).start_fixture()
  468 
  469     def initialize_data(self):
  470         data = test_utils.generate_v2_storage_data(
  471             start=tzutils.get_month_start(),
  472             end=tzutils.localized_now().replace(hour=0),
  473         )
  474         self.storage.push([data])
  475 
  476     def stop_fixture(self):
  477         self._get_storage_patch.stop()
  478 
  479 
  480 class InfluxStorageDataFixture(StorageDataFixture):
  481 
  482     def start_fixture(self):
  483         cli = influx_utils.FakeInfluxClient()
  484         st = storage.get_storage()
  485         st._conn = cli
  486 
  487         self._get_storage_patch = mock.patch(
  488             'cloudkitty.storage.get_storage',
  489             new=lambda **kw: st,
  490         )
  491         self._get_storage_patch.start()
  492         v2_api_summary.Summary.reload()
  493         v2_api_dataframes.DataFrameList.reload()
  494 
  495         super(InfluxStorageDataFixture, self).start_fixture()
  496 
  497     def stop_fixture(self):
  498         self._get_storage_patch.stop()
  499 
  500 
  501 class UTCFixture(fixture.GabbiFixture):
  502     """Set the local timezone to UTC"""
  503     def start_fixture(self):
  504         self._tzmock = mock.patch('cloudkitty.utils.tz._LOCAL_TZ', tz.tzutc())
  505         self._tzmock.start()
  506 
  507     def stop_fixture(self):
  508         self._tzmock.stop()
  509 
  510 
  511 def setup_app():
  512     messaging.setup()
  513     # FIXME(sheeprine): Extension fixtures are interacting with transformers
  514     # loading, since collectors are not needed here we shunt them
  515     no_collector = mock.patch(
  516         'cloudkitty.collector.get_collector',
  517         return_value=None)
  518     with no_collector:
  519         return app.load_app()