ceilometer/tests/tempest/scenario/test_autoscaling.py

#    Licensed under the Apache License, Version 2.0 (the "License"); you may
#    not use this file except in compliance with the License. You may obtain
#    a copy of the License at
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
#    License for the specific language governing permissions and limitations
#    under the License.

import os
import unittest

from gabbi import driver
from tempest import config
from tempest.scenario import manager


class TestAutoscalingGabbi(manager.ScenarioTest):
    credentials = ['admin', 'primary']

    @classmethod
    def skip_checks(cls):
        super(TestAutoscalingGabbi, cls).skip_checks()
        for name in ["aodh_plugin", "gnocchi", "nova", "heat", "panko",
                     "ceilometer", "glance"]:
            cls._check_service(name)

    @classmethod
    def _check_service(cls, name):
        if not getattr(config.CONF.service_available, name, False):
            raise cls.skipException("%s support is required" %
                                    name.capitalize())

    @classmethod
    def resource_setup(cls):
        super(TestAutoscalingGabbi, cls).resource_setup()
        test_dir = os.path.join(os.path.dirname(__file__), '..', '..',
                                'integration', 'gabbi', 'gabbits-live')
        cls.tests = driver.build_tests(
            test_dir, unittest.TestLoader(),
            host='localhost', port='13245',
            test_loader_name='tempest.scenario.telemetry-autoscaling.test')

        auth = cls.os_admin.auth_provider.get_auth()
        os.environ["ADMIN_TOKEN"] = auth[0]
        os.environ["AODH_SERVICE_URL"] = cls._get_endpoint_for(
            auth, "alarming_plugin")
        os.environ["GNOCCHI_SERVICE_URL"] = cls._get_endpoint_for(
            auth, "metric")
        os.environ["PANKO_SERVICE_URL"] = cls._get_endpoint_for(
            auth, "event")
        os.environ["HEAT_SERVICE_URL"] = cls._get_endpoint_for(
            auth, "orchestration")
        os.environ["NOVA_SERVICE_URL"] = cls._get_endpoint_for(auth, "compute")
        os.environ["GLANCE_SERVICE_URL"] = cls._get_endpoint_for(auth, "image")

    @staticmethod
    def clear_credentials():
        # FIXME(sileht): We don't want the token to be invalided, but
        # for some obcurs reason, clear_credentials is called before/during run
        # So, make the one used by tearDropClass a dump, and call it manually
        # in run()
        pass

    def run(self, result=None):
        self.setUp()
        os.environ["GLANCE_IMAGE_NAME"] = self.glance_image_create()
        try:
            self.tests.run(result)
        finally:
            super(TestAutoscalingGabbi, self).clear_credentials()
            self.tearDown()

    @staticmethod
    def _get_endpoint_for(auth, service):
        opt_section = getattr(config.CONF, service)
        endpoint_type = opt_section.endpoint_type
        is_keystone_v3 = 'catalog' in auth[1]

        if is_keystone_v3:
            if endpoint_type.endswith("URL"):
                endpoint_type = endpoint_type[:-3]
            catalog = auth[1]['catalog']
            endpoints = [e['endpoints'] for e in catalog
                         if e['type'] == opt_section.catalog_type]
            if not endpoints:
                raise Exception("%s endpoint not found" %
                                config.CONF.metric.catalog_type)
            endpoints = [e['url'] for e in endpoints[0]
                         if e['interface'] == endpoint_type]
            if not endpoints:
                raise Exception("%s interface not found for endpoint %s" %
                                (endpoint_type,
                                 config.CONF.metric.catalog_type))
            return endpoints[0]

        else:
            if not endpoint_type.endswith("URL"):
                endpoint_type += "URL"
            catalog = auth[1]['serviceCatalog']
            endpoints = [e for e in catalog
                         if e['type'] == opt_section.catalog_type]
            if not endpoints:
                raise Exception("%s endpoint not found" %
                                config.CONF.metric.catalog_type)
            return endpoints[0]['endpoints'][0][endpoint_type]

    @staticmethod
    def test_fake():
        # NOTE(sileht): A fake test is needed to have the class loaded
        # by the test runner
        pass