path: root/snaps/openstack/utils/tests/magnum_utils_tests.py

# Copyright (c) 2017 Cable Television Laboratories, Inc. ("CableLabs")
#                    and others.  All rights reserved.
#
# 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 logging
import uuid

from magnumclient.common.apiclient.exceptions import BadRequest, NotFound

from snaps.config.cluster_template import (
    ClusterTemplateConfig, ServerType,  ContainerOrchestrationEngine,
    DockerStorageDriver)
from snaps.config.flavor import FlavorConfig
from snaps.config.keypair import KeypairConfig
from snaps.openstack.create_flavor import OpenStackFlavor
from snaps.openstack.create_image import OpenStackImage
from snaps.openstack.create_keypairs import OpenStackKeypair
from snaps.openstack.os_credentials import OSCreds
from snaps.openstack.tests import openstack_tests
from snaps.openstack.tests.os_source_file_test import OSComponentTestCase
from snaps.openstack.utils import magnum_utils

__author__ = 'spisarski'

logger = logging.getLogger('magnum_utils_tests')


class MagnumSmokeTests(OSComponentTestCase):
    """
    Tests to ensure that the magnum client can communicate with the cloud
    """

    def test_connect_success(self):
        """
        Tests to ensure that the proper credentials can connect.
        """
        magnum = magnum_utils.magnum_client(self.os_creds)

        # This should not throw an exception
        self.assertIsNotNone(magnum.clusters.list())

    def test_nova_connect_fail(self):
        """
        Tests to ensure that the improper credentials cannot connect.
        """

        with self.assertRaises(RuntimeError):
            magnum_utils.magnum_client(
                OSCreds(username='user', password='pass',
                        auth_url=self.os_creds.auth_url,
                        project_name=self.os_creds.project_name,
                        proxy_settings=self.os_creds.proxy_settings))


class MagnumUtilsClusterTypeTests(OSComponentTestCase):
    """
    Tests individual functions within magnum_utils.py
    """

    def setUp(self):
        self.guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
        self.cluster_type_name = self.guid + '-cluster-type'
        self.magnum = magnum_utils.magnum_client(self.os_creds)

        metadata = self.image_metadata
        if not metadata:
            metadata = dict()
        if 'extra_properties' not in metadata:
            metadata['extra_properties'] = dict()
        metadata['extra_properties']['os_distro'] = 'cirros'

        os_image_settings = openstack_tests.cirros_image_settings(
            name=self.guid + '-image', image_metadata=metadata)

        self.image_creator = OpenStackImage(self.os_creds, os_image_settings)

        self.flavor_creator = OpenStackFlavor(
            self.os_creds, FlavorConfig(
                name=self.guid + '-flavor', ram=512, disk=10, vcpus=1))

        keypair_priv_filepath = 'tmp/' + self.guid
        keypair_pub_filepath = keypair_priv_filepath + '.pub'

        self.keypair_creator = OpenStackKeypair(
            self.os_creds, KeypairConfig(
                name=self.guid + '-keypair',
                public_filepath=keypair_pub_filepath,
                private_filepath=keypair_priv_filepath))

        self.cluster_template = None

        try:
            self.image_creator.create()
            self.flavor_creator.create()
            self.keypair_creator.create()
        except:
            self.tearDown()
            raise

    def tearDown(self):
        if self.cluster_template:
            try:
                magnum_utils.delete_cluster_template(
                    self.magnum, self.cluster_template.id)
            except:
                pass
        if self.keypair_creator:
            try:
                self.keypair_creator.clean()
            except:
                pass
        if self.flavor_creator:
            try:
                self.flavor_creator.clean()
            except:
                pass
        if self.image_creator:
            try:
                self.image_creator.clean()
            except:
                pass

    def test_create_cluster_template_simple(self):
        config = ClusterTemplateConfig(
            name=self.cluster_type_name,
            image=self.image_creator.image_settings.name,
            keypair=self.keypair_creator.keypair_settings.name,
            external_net=self.ext_net_name,
            flavor=self.flavor_creator.flavor_settings.name)

        self.cluster_template = magnum_utils.create_cluster_template(
            self.magnum, config)
        self.assertIsNotNone(self.cluster_template)
        self.assertTrue(
            validate_cluster_template(config, self.cluster_template))

        template_by_name = magnum_utils.get_cluster_template(
            self.magnum, template_name=config.name)
        self.assertEqual(self.cluster_template, template_by_name)
        template_by_id = magnum_utils.get_cluster_template_by_id(
            self.magnum, self.cluster_template.id)
        self.assertEqual(self.cluster_template, template_by_id)

    def test_create_cluster_template_all(self):
        config = ClusterTemplateConfig(
            name=self.cluster_type_name,
            image=self.image_creator.image_settings.name,
            keypair=self.keypair_creator.keypair_settings.name,
            network_driver='flannel', external_net=self.ext_net_name,
            floating_ip_enabled=True, docker_volume_size=100,
            server_type=ServerType.vm,
            flavor=self.flavor_creator.flavor_settings.name,
            master_flavor=self.flavor_creator.flavor_settings.name,
            coe=ContainerOrchestrationEngine.kubernetes,
            fixed_net='foo', fixed_subnet='bar',
            registry_enabled=True, insecure_registry='localhost',
            docker_storage_driver=DockerStorageDriver.overlay,
            dns_nameserver='8.8.4.4', public=True, tls_disabled=True,
            http_proxy=None, https_proxy=None, volume_driver='cinder',
            master_lb_enabled=False, labels={'foo': 'bar'})

        self.cluster_template = magnum_utils.create_cluster_template(
            self.magnum, config)
        self.assertIsNotNone(self.cluster_template)
        self.assertTrue(
            validate_cluster_template(config, self.cluster_template))

        template_by_name = magnum_utils.get_cluster_template(
            self.magnum, template_name=config.name)
        self.assertEqual(self.cluster_template, template_by_name)
        template_by_id = magnum_utils.get_cluster_template_by_id(
            self.magnum, self.cluster_template.id)
        self.assertEqual(self.cluster_template, template_by_id)

    def test_create_cluster_template_bad_image(self):
        config = ClusterTemplateConfig(
            name=self.cluster_type_name,
            image='foo',
            keypair=self.keypair_creator.keypair_settings.name,
            external_net=self.ext_net_name,
            flavor=self.flavor_creator.flavor_settings.name)

        with self.assertRaises(BadRequest):
            self.cluster_template = magnum_utils.create_cluster_template(
                self.magnum, config)

    def test_create_cluster_template_bad_kp(self):
        config = ClusterTemplateConfig(
            name=self.cluster_type_name,
            image=self.image_creator.image_settings.name,
            keypair='foo',
            external_net=self.ext_net_name,
            flavor=self.flavor_creator.flavor_settings.name)

        with self.assertRaises(NotFound):
            self.cluster_template = magnum_utils.create_cluster_template(
                self.magnum, config)

    def test_create_cluster_template_bad_ext_net(self):
        config = ClusterTemplateConfig(
            name=self.cluster_type_name,
            image=self.image_creator.image_settings.name,
            keypair=self.keypair_creator.keypair_settings.name,
            external_net='foo',
            flavor=self.flavor_creator.flavor_settings.name)

        with self.assertRaises(BadRequest):
            self.cluster_template = magnum_utils.create_cluster_template(
                self.magnum, config)

    def test_create_cluster_template_bad_flavor(self):
        config = ClusterTemplateConfig(
            name=self.cluster_type_name,
            image=self.image_creator.image_settings.name,
            keypair=self.keypair_creator.keypair_settings.name,
            external_net=self.ext_net_name,
            flavor='foo')

        with self.assertRaises(BadRequest):
            self.cluster_template = magnum_utils.create_cluster_template(
                self.magnum, config)

    def test_create_cluster_template_bad_master_flavor(self):
        config = ClusterTemplateConfig(
            name=self.cluster_type_name,
            image=self.image_creator.image_settings.name,
            keypair=self.keypair_creator.keypair_settings.name,
            external_net=self.ext_net_name,
            flavor=self.flavor_creator.flavor_settings.name,
            master_flavor='foo')

        with self.assertRaises(BadRequest):
            self.cluster_template = magnum_utils.create_cluster_template(
                self.magnum, config)

    def test_create_cluster_template_bad_network_driver(self):
        config = ClusterTemplateConfig(
            name=self.cluster_type_name,
            image=self.image_creator.image_settings.name,
            keypair=self.keypair_creator.keypair_settings.name,
            external_net=self.ext_net_name,
            network_driver='foo')

        with self.assertRaises(BadRequest):
            self.cluster_template = magnum_utils.create_cluster_template(
                self.magnum, config)

    def test_create_cluster_template_bad_volume_driver(self):
        config = ClusterTemplateConfig(
            name=self.cluster_type_name,
            image=self.image_creator.image_settings.name,
            keypair=self.keypair_creator.keypair_settings.name,
            external_net=self.ext_net_name,
            volume_driver='foo')

        with self.assertRaises(BadRequest):
            self.cluster_template = magnum_utils.create_cluster_template(
                self.magnum, config)


def validate_cluster_template(tmplt_config, tmplt_obj):
    """
    Returns true if the configuration matches the ClusterTemplate object
    :param tmplt_config: the ClusterTemplateConfig object
    :param tmplt_obj: the ClusterTemplate domain object
    :return: T/F
    """
    if not tmplt_config.network_driver:
        network_driver = 'flannel'
    else:
        network_driver = tmplt_config.network_driver

    return (
        tmplt_config.coe.value == tmplt_obj.coe and
        tmplt_config.dns_nameserver == tmplt_obj.dns_nameserver and
        tmplt_config.docker_storage_driver.value
        == tmplt_obj.docker_storage_driver and
        tmplt_config.docker_volume_size == tmplt_obj.docker_volume_size and
        tmplt_config.external_net == tmplt_obj.external_net and
        tmplt_config.fixed_net == tmplt_obj.fixed_net and
        tmplt_config.fixed_subnet == tmplt_obj.fixed_subnet and
        tmplt_config.flavor == tmplt_obj.flavor and
        tmplt_config.floating_ip_enabled == tmplt_obj.floating_ip_enabled and
        tmplt_config.http_proxy == tmplt_obj.http_proxy and
        tmplt_config.https_proxy == tmplt_obj.https_proxy and
        tmplt_config.no_proxy == tmplt_obj.no_proxy and
        tmplt_config.image == tmplt_obj.image and
        tmplt_config.insecure_registry == tmplt_obj.insecure_registry and
        tmplt_config.keypair == tmplt_obj.keypair and
        tmplt_config.labels == tmplt_obj.labels and
        tmplt_config.master_flavor == tmplt_obj.master_flavor and
        tmplt_config.master_lb_enabled == tmplt_obj.master_lb_enabled and
        tmplt_config.name == tmplt_obj.name and
        network_driver == tmplt_obj.network_driver and
        tmplt_config.no_proxy == tmplt_obj.no_proxy and
        tmplt_config.public == tmplt_obj.public and
        tmplt_config.registry_enabled == tmplt_obj.registry_enabled and
        tmplt_config.server_type.value == tmplt_obj.server_type and
        tmplt_config.tls_disabled == tmplt_obj.tls_disabled and
        tmplt_config.volume_driver == tmplt_obj.volume_driver
    )