python-keystoneclient>=3.8.0 # Apache-2.0
python-glanceclient>=2.5.0 # Apache-2.0
python-heatclient>=1.6.1 # Apache-2.0
+python-cinderclient
ansible>=2.1.0,<2.4
wrapt>=1.7.0 # BSD License
scp
--- /dev/null
+# 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 unittest
+from snaps.domain.volume import QoSSpec
+
+
+class QoSSpecDomainObjectTests(unittest.TestCase):
+ """
+ Tests the construction of the snaps.domain.volume.QoSSpec class
+ """
+
+ def test_construction_positional(self):
+ qos_spec = QoSSpec('name', 'id', 'consumer')
+ self.assertEqual('name', qos_spec.name)
+ self.assertEqual('id', qos_spec.id)
+ self.assertEqual('consumer', qos_spec.consumer)
+
+ def test_construction_named(self):
+ qos_spec = QoSSpec(consumer='consumer', spec_id='id', name='name')
+ self.assertEqual('name', qos_spec.name)
+ self.assertEqual('id', qos_spec.id)
+ self.assertEqual('consumer', qos_spec.consumer)
--- /dev/null
+# 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.
+
+
+class QoSSpec:
+ """
+ SNAPS domain object for Volume Types. Should contain attributes that
+ are shared amongst cloud providers
+ """
+ def __init__(self, name, spec_id, consumer):
+ """
+ Constructor
+ :param name: the volume's name
+ :param spec_id: the QoS Spec's id
+ """
+ self.name = name
+ self.id = spec_id
+ self.consumer = consumer
+
+ def __eq__(self, other):
+ return (self.name == other.name and self.id == other.id
+ and self.consumer == other.consumer)
--- /dev/null
+# 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 enum
+from cinderclient.exceptions import NotFound
+
+from snaps.openstack.openstack_creator import OpenStackVolumeObject
+from snaps.openstack.utils import cinder_utils
+
+__author__ = 'spisarski'
+
+logger = logging.getLogger('create_qos')
+
+IMAGE_ACTIVE_TIMEOUT = 600
+POLL_INTERVAL = 3
+STATUS_ACTIVE = 'active'
+
+
+class OpenStackQoS(OpenStackVolumeObject):
+ """
+ Class responsible for managing an qos in OpenStack
+ """
+
+ def __init__(self, os_creds, qos_settings):
+ """
+ Constructor
+ :param os_creds: The OpenStack connection credentials
+ :param qos_settings: The qos settings
+ :return:
+ """
+ super(self.__class__, self).__init__(os_creds)
+
+ self.qos_settings = qos_settings
+ self.__qos = None
+
+ def initialize(self):
+ """
+ Loads the existing QoS
+ :return: The QoS domain object or None
+ """
+ super(self.__class__, self).initialize()
+
+ self.__qos = cinder_utils.get_qos(
+ self._cinder, qos_settings=self.qos_settings)
+
+ return self.__qos
+
+ def create(self):
+ """
+ Creates the qos in OpenStack if it does not already exist and returns
+ the domain QoS object
+ :return: The QoS domain object or None
+ """
+ self.initialize()
+
+ if not self.__qos:
+ self.__qos = cinder_utils.create_qos(
+ self._cinder, self.qos_settings)
+
+ logger.info(
+ 'Created qos with name - %s', self.qos_settings.name)
+
+ return self.__qos
+
+ def clean(self):
+ """
+ Cleanse environment of all artifacts
+ :return: void
+ """
+ if self.__qos:
+ try:
+ cinder_utils.delete_qos(self._cinder, self.__qos)
+ except NotFound:
+ pass
+
+ self.__qos = None
+
+ def get_qos(self):
+ """
+ Returns the domain QoS object as it was populated when create() was
+ called
+ :return: the object
+ """
+ return self.__qos
+
+
+class Consumer(enum.Enum):
+ """
+ QoS Specification consumer types
+ """
+ front_end = 'front-end'
+ back_end = 'back-end'
+ both = 'both'
+
+
+class QoSSettings:
+ def __init__(self, **kwargs):
+ """
+ Constructor
+ :param name: the qos's name (required)
+ :param consumer: the qos's consumer type (required)
+ :param specs: dict of key/values
+ """
+
+ self.name = kwargs.get('name')
+
+ if kwargs.get('consumer'):
+ self.consumer = map_consumer(kwargs['consumer'])
+ else:
+ self.consumer = None
+
+ self.specs = kwargs.get('specs')
+ if not self.specs:
+ self.specs = dict()
+
+ if not self.name or not self.consumer:
+ raise QoSSettingsError(
+ "The attributes name and consumer are required")
+
+
+def map_consumer(consumer):
+ """
+ Takes a the protocol value maps it to the Consumer enum. When None return
+ None
+ :param consumer: the value to map to the Enum
+ :return: the Protocol enum object
+ :raise: Exception if value is invalid
+ """
+ if not consumer:
+ return None
+ elif isinstance(consumer, Consumer):
+ return consumer
+ else:
+ proto_str = str(consumer)
+ if proto_str == 'front-end':
+ return Consumer.front_end
+ elif proto_str == 'back-end':
+ return Consumer.back_end
+ elif proto_str == 'both':
+ return Consumer.both
+ else:
+ raise QoSSettingsError('Invalid Consumer - ' + proto_str)
+
+
+class QoSSettingsError(Exception):
+ """
+ Exception to be thrown when an qos settings are incorrect
+ """
+
+ def __init__(self, message):
+ Exception.__init__(self, message)
+
+
+class QoSCreationError(Exception):
+ """
+ Exception to be thrown when an qos cannot be created
+ """
+
+ def __init__(self, message):
+ Exception.__init__(self, message)
# See the License for the specific language governing permissions and
# limitations under the License.
from snaps.domain.creator import CloudObject
-from snaps.openstack.utils import nova_utils, neutron_utils, keystone_utils
+from snaps.openstack.utils import (nova_utils, neutron_utils, keystone_utils,
+ cinder_utils)
__author__ = 'spisarski'
def clean(self):
raise NotImplementedError('Do not override abstract method')
+
+
+class OpenStackVolumeObject(OpenStackCloudObject):
+ """
+ Abstract class for all OpenStack compute creators
+ """
+
+ def __init__(self, os_creds):
+ """
+ Constructor
+ :param os_creds: the OpenStack credentials object
+ """
+ super(OpenStackVolumeObject, self).__init__(os_creds)
+ self._cinder = None
+
+ def initialize(self):
+ self._cinder = cinder_utils.cinder_client(self._os_creds)
+
+ def create(self):
+ raise NotImplementedError('Do not override abstract method')
+
+ def clean(self):
+ raise NotImplementedError('Do not override abstract method')
from neutronclient.common.utils import str2bool
import numbers
from snaps import file_utils
-from snaps.openstack.utils import glance_utils, keystone_utils
+from snaps.openstack.utils import glance_utils, keystone_utils, cinder_utils
__author__ = 'spisarski'
clients
:param heat_api_version: The OpenStack's API version to use for Heat
clients
+ :param volume_api_version: The OpenStack's API version to use
+ for Cinder clients
:param user_domain_id: Used for v3 APIs (default='default')
:param user_domain_name: Used for v3 APIs (default='Default')
:param project_domain_id: Used for v3 APIs (default='default')
else:
self.heat_api_version = float(kwargs['heat_api_version'])
+ if kwargs.get('volume_api_version') is None:
+ self.volume_api_version = cinder_utils.VERSION_2
+ else:
+ self.volume_api_version = float(
+ kwargs['volume_api_version'])
+
self.user_domain_id = kwargs.get('user_domain_id', 'default')
if kwargs.get('user_domain_name') is None:
--- /dev/null
+# 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.
+
+try:
+ from urllib.request import URLError
+except ImportError:
+ from urllib2 import URLError
+
+import logging
+import unittest
+import uuid
+
+from snaps.openstack import create_qos
+from snaps.openstack.create_qos import (QoSSettings, QoSSettingsError,
+ Consumer)
+from snaps.openstack.tests.os_source_file_test import OSIntegrationTestCase
+from snaps.openstack.utils import cinder_utils
+
+__author__ = 'spisarski'
+
+logger = logging.getLogger('create_qos_tests')
+
+
+class QoSSettingsUnitTests(unittest.TestCase):
+ """
+ Tests the construction of the QoSSettings class
+ """
+
+ def test_no_params(self):
+ with self.assertRaises(QoSSettingsError):
+ QoSSettings()
+
+ def test_empty_config(self):
+ with self.assertRaises(QoSSettingsError):
+ QoSSettings(**dict())
+
+ def test_name_only(self):
+ with self.assertRaises(QoSSettingsError):
+ QoSSettings(name='foo')
+
+ def test_config_with_name_only(self):
+ with self.assertRaises(QoSSettingsError):
+ QoSSettings(**{'name': 'foo'})
+
+ def test_invalid_consumer(self):
+ with self.assertRaises(QoSSettingsError):
+ QoSSettings(name='foo', consumer='bar')
+
+ def test_config_with_invalid_consumer(self):
+ with self.assertRaises(QoSSettingsError):
+ QoSSettings(**{'name': 'foo', 'consumer': 'bar'})
+
+ def test_name_consumer(self):
+ settings = QoSSettings(name='foo', consumer=Consumer.front_end)
+
+ self.assertEqual('foo', settings.name)
+ self.assertEqual(Consumer.front_end, settings.consumer)
+ self.assertEqual(dict(), settings.specs)
+
+ def test_name_consumer_front_end_strings(self):
+ settings = QoSSettings(name='foo', consumer='front-end')
+
+ self.assertEqual('foo', settings.name)
+ self.assertEqual(Consumer.front_end, settings.consumer)
+ self.assertEqual(dict(), settings.specs)
+
+ def test_name_consumer_back_end_strings(self):
+ settings = QoSSettings(name='foo', consumer='back-end')
+
+ self.assertEqual('foo', settings.name)
+ self.assertEqual(Consumer.back_end, settings.consumer)
+ self.assertEqual(dict(), settings.specs)
+
+ def test_name_consumer_both_strings(self):
+ settings = QoSSettings(name='foo', consumer='both')
+
+ self.assertEqual('foo', settings.name)
+ self.assertEqual(Consumer.both, settings.consumer)
+ self.assertEqual(dict(), settings.specs)
+
+ def test_all(self):
+ specs = {'spec1': 'val1', 'spec2': 'val2'}
+ settings = QoSSettings(name='foo', consumer=Consumer.both,
+ specs=specs)
+
+ self.assertEqual('foo', settings.name)
+ self.assertEqual(Consumer.both, settings.consumer)
+ self.assertEqual(specs, settings.specs)
+
+ def test_config_all(self):
+ settings = QoSSettings(
+ **{'name': 'foo', 'consumer': 'both', 'specs': {'spec1': 'val1'}})
+
+ self.assertEqual('foo', settings.name)
+ self.assertEqual(Consumer.both, settings.consumer)
+ self.assertEqual({'spec1': 'val1'}, settings.specs)
+
+
+class CreateQoSTests(OSIntegrationTestCase):
+ """
+ Test for the CreateQoS class defined in create_qos.py
+ """
+
+ def setUp(self):
+ """
+ Instantiates the CreateQoS object that is responsible for
+ downloading and creating an OS QoS Spec file within OpenStack
+ """
+ super(self.__class__, self).__start__()
+
+ guid = uuid.uuid4()
+ self.qos_settings = QoSSettings(
+ name=self.__class__.__name__ + '-' + str(guid),
+ consumer=Consumer.both)
+
+ self.cinder = cinder_utils.cinder_client(self.os_creds)
+ self.qos_creator = None
+
+ def tearDown(self):
+ """
+ Cleans the Qos Spec
+ """
+ if self.qos_creator:
+ self.qos_creator.clean()
+
+ super(self.__class__, self).__clean__()
+
+ def test_create_qos(self):
+ """
+ Tests the creation of an OpenStack qos.
+ """
+ # Create QoS
+ self.qos_creator = create_qos.OpenStackQoS(
+ self.os_creds, self.qos_settings)
+ created_qos = self.qos_creator.create()
+ self.assertIsNotNone(created_qos)
+
+ retrieved_qos = cinder_utils.get_qos(
+ self.cinder, qos_settings=self.qos_settings)
+
+ self.assertIsNotNone(retrieved_qos)
+ self.assertEqual(created_qos, retrieved_qos)
+
+ def test_create_delete_qos(self):
+ """
+ Tests the creation then deletion of an OpenStack QoS Spec to ensure
+ clean() does not raise an Exception.
+ """
+ # Create QoS
+ self.qos_creator = create_qos.OpenStackQoS(
+ self.os_creds, self.qos_settings)
+ created_qos = self.qos_creator.create()
+ self.assertIsNotNone(created_qos)
+
+ retrieved_qos = cinder_utils.get_qos(
+ self.cinder, qos_settings=self.qos_settings)
+ self.assertIsNotNone(retrieved_qos)
+ self.assertEqual(created_qos, retrieved_qos)
+
+ # Delete QoS manually
+ cinder_utils.delete_qos(self.cinder, created_qos)
+
+ self.assertIsNone(cinder_utils.get_qos(
+ self.cinder, qos_settings=self.qos_settings))
+
+ # Must not raise an exception when attempting to cleanup non-existent
+ # qos
+ self.qos_creator.clean()
+ self.assertIsNone(self.qos_creator.get_qos())
+
+ def test_create_same_qos(self):
+ """
+ Tests the creation of an OpenStack qos when one already exists.
+ """
+ # Create QoS
+ self.qos_creator = create_qos.OpenStackQoS(
+ self.os_creds, self.qos_settings)
+ qos1 = self.qos_creator.create()
+
+ retrieved_qos = cinder_utils.get_qos(
+ self.cinder, qos_settings=self.qos_settings)
+ self.assertEqual(qos1, retrieved_qos)
+
+ # Should be retrieving the instance data
+ os_qos_2 = create_qos.OpenStackQoS(
+ self.os_creds, self.qos_settings)
+ qos2 = os_qos_2.create()
+ self.assertEqual(qos1, qos2)
'user_domain_name': config.get('OS_USER_DOMAIN_NAME'),
'project_domain_id': config.get('OS_PROJECT_DOMAIN_ID'),
'project_domain_name': config.get('OS_PROJECT_DOMAIN_NAME'),
+ 'volume_api_version': config.get('OS_VOLUME_API_VERSION'),
'interface': interface,
'proxy_settings': proxy_settings,
'cacert': https_cacert,
'user_domain_name': config.get('user_domain_name'),
'project_domain_id': config.get('project_domain_id'),
'project_domain_name': config.get('project_domain_name'),
+ 'volume_api_version': config.get('volume_api_version'),
'interface': config.get('interface'),
'proxy_settings': proxy_settings,
'cacert': config.get('cacert'),
--- /dev/null
+# 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
+
+from cinderclient.client import Client
+
+from snaps.domain.volume import QoSSpec
+from snaps.openstack.utils import keystone_utils
+
+__author__ = 'spisarski'
+
+logger = logging.getLogger('cinder_utils')
+
+VERSION_1 = 1
+VERSION_2 = 2
+VERSION_3 = 3
+
+"""
+Utilities for basic neutron API calls
+"""
+
+
+def cinder_client(os_creds):
+ """
+ Creates and returns a cinder client object
+ :return: the cinder client
+ """
+ return Client(version=os_creds.volume_api_version,
+ session=keystone_utils.keystone_session(os_creds),
+ region_name=os_creds.region_name)
+
+
+def get_qos(cinder, qos_name=None, qos_settings=None):
+ """
+ Returns an OpenStack QoS object for a given name
+ :param cinder: the Cinder client
+ :param qos_name: the qos name to lookup
+ :param qos_settings: the qos settings used for lookups
+ :return: the qos object or None
+ """
+ if not qos_name and not qos_settings:
+ return None
+
+ qos_name = qos_name
+ if qos_settings:
+ qos_name = qos_settings.name
+
+ qoss = cinder.qos_specs.list()
+ for qos in qoss:
+ if qos.name == qos_name:
+ if qos_settings:
+ if qos_settings.consumer.value == qos.consumer:
+ return QoSSpec(name=qos.name, spec_id=qos.id,
+ consumer=qos.consumer)
+ else:
+ return QoSSpec(name=qos.name, spec_id=qos.id,
+ consumer=qos.consumer)
+
+
+def get_qos_by_id(cinder, qos_id):
+ """
+ Returns an OpenStack qos object for a given name
+ :param cinder: the Cinder client
+ :param qos_id: the qos ID to lookup
+ :return: the SNAPS-OO Domain Volume object or None
+ """
+ qos = cinder.qos_specs.get(qos_id)
+ return QoSSpec(name=qos.name, spec_id=qos.id, consumer=qos.consumer)
+
+
+def create_qos(cinder, qos_settings):
+ """
+ Creates and returns OpenStack qos object with an external URL
+ :param cinder: the cinder client
+ :param qos_settings: the qos settings object
+ :return: the qos domain object
+ :raise Exception if using a file and it cannot be found
+ """
+ specs = qos_settings.specs
+ specs['consumer'] = qos_settings.consumer.value
+ qos = cinder.qos_specs.create(qos_settings.name, qos_settings.specs)
+ return QoSSpec(name=qos.name, spec_id=qos.id, consumer=qos.consumer)
+
+
+def delete_qos(cinder, qos):
+ """
+ Deletes an QoS from OpenStack
+ :param cinder: the cinder client
+ :param qos: the qos domain object to delete
+ """
+ logger.info('Deleting QoS named - %s', qos.name)
+ cinder.qos_specs.delete(qos.id)
+
+
+class CinderException(Exception):
+ """
+ Exception when calls to the Cinder client cannot be served properly
+ """
--- /dev/null
+# 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 cinderclient.exceptions import NotFound
+
+from snaps.openstack.create_qos import QoSSettings, Consumer
+from snaps.openstack.tests import validation_utils
+from snaps.openstack.tests.os_source_file_test import OSComponentTestCase
+from snaps.openstack.utils import cinder_utils
+
+__author__ = 'spisarski'
+
+
+logger = logging.getLogger('cinder_utils_tests')
+
+
+class CinderSmokeTests(OSComponentTestCase):
+ """
+ Tests to ensure that the neutron client can communicate with the cloud
+ """
+
+ def test_cinder_connect_success(self):
+ """
+ Tests to ensure that the proper credentials can connect.
+ """
+ cinder = cinder_utils.cinder_client(self.os_creds)
+ volumes = cinder.volumes.list()
+ self.assertIsNotNone(volumes)
+ self.assertTrue(isinstance(volumes, list))
+
+ def test_cinder_connect_fail(self):
+ """
+ Tests to ensure that the improper credentials cannot connect.
+ """
+ from snaps.openstack.os_credentials import OSCreds
+
+ with self.assertRaises(Exception):
+ cinder = cinder_utils.cinder_client(OSCreds(
+ username='user', password='pass', auth_url='url',
+ project_name='project'))
+ cinder.volumes.list()
+
+
+class CinderUtilsQoSTests(OSComponentTestCase):
+ """
+ Test for the CreateQos class defined in create_qos.py
+ """
+
+ def setUp(self):
+ """
+ Creates objects for testing cinder_utils.py
+ """
+ guid = uuid.uuid4()
+ self.qos_name = self.__class__.__name__ + '-' + str(guid)
+ self.specs = {'foo': 'bar '}
+ self.qos = None
+ self.cinder = cinder_utils.cinder_client(self.os_creds)
+
+ def tearDown(self):
+ """
+ Cleans the remote OpenStack objects
+ """
+ if self.qos:
+ try:
+ cinder_utils.delete_qos(self.cinder, self.qos)
+ except NotFound:
+ pass
+
+ def test_create_qos_both(self):
+ """
+ Tests the cinder_utils.create_qos()
+ """
+ qos_settings = QoSSettings(name=self.qos_name, specs=self.specs,
+ consumer=Consumer.both)
+ self.qos = cinder_utils.create_qos(
+ self.cinder, qos_settings)
+ self.assertIsNotNone(self.qos)
+
+ qos1 = cinder_utils.get_qos(self.cinder, qos_settings=qos_settings)
+ self.assertIsNotNone(qos1)
+ validation_utils.objects_equivalent(self.qos, qos1)
+
+ qos2 = cinder_utils.get_qos(self.cinder, qos_name=qos_settings.name)
+ self.assertIsNotNone(qos2)
+ validation_utils.objects_equivalent(self.qos, qos2)
+
+ def test_create_qos_front(self):
+ """
+ Tests the cinder_utils.create_qos()
+ """
+ qos_settings = QoSSettings(name=self.qos_name, specs=self.specs,
+ consumer=Consumer.front_end)
+ self.qos = cinder_utils.create_qos(
+ self.cinder, qos_settings)
+ self.assertIsNotNone(self.qos)
+
+ qos1 = cinder_utils.get_qos(self.cinder, qos_settings=qos_settings)
+ self.assertIsNotNone(qos1)
+ validation_utils.objects_equivalent(self.qos, qos1)
+
+ qos2 = cinder_utils.get_qos(self.cinder, qos_name=qos_settings.name)
+ self.assertIsNotNone(qos2)
+ validation_utils.objects_equivalent(self.qos, qos2)
+
+ def test_create_qos_back(self):
+ """
+ Tests the cinder_utils.create_qos()
+ """
+ qos_settings = QoSSettings(name=self.qos_name, specs=self.specs,
+ consumer=Consumer.back_end)
+ self.qos = cinder_utils.create_qos(
+ self.cinder, qos_settings)
+ self.assertIsNotNone(self.qos)
+
+ qos1 = cinder_utils.get_qos(self.cinder, qos_settings=qos_settings)
+ self.assertIsNotNone(qos1)
+ validation_utils.objects_equivalent(self.qos, qos1)
+
+ qos2 = cinder_utils.get_qos(self.cinder, qos_name=qos_settings.name)
+ self.assertIsNotNone(qos2)
+ validation_utils.objects_equivalent(self.qos, qos2)
+
+ def test_create_delete_qos(self):
+ """
+ Tests the cinder_utils.create_qos()
+ """
+ qos_settings = QoSSettings(name=self.qos_name, consumer=Consumer.both)
+ self.qos = cinder_utils.create_qos(
+ self.cinder, qos_settings)
+ self.assertIsNotNone(self.qos)
+ self.assertEqual(self.qos_name, self.qos.name)
+
+ qos = cinder_utils.get_qos(
+ self.cinder, qos_settings=qos_settings)
+ self.assertIsNotNone(qos)
+ validation_utils.objects_equivalent(self.qos, qos)
+
+ cinder_utils.delete_qos(self.cinder, self.qos)
+ self.assertIsNone(cinder_utils.get_qos(
+ self.cinder, qos_settings=qos_settings))
from snaps.domain.test.user_tests import UserDomainObjectTests
from snaps.domain.test.vm_inst_tests import (
VmInstDomainObjectTests, FloatingIpDomainObjectTests)
+from snaps.domain.test.volume_tests import QoSSpecDomainObjectTests
from snaps.openstack.tests.conf.os_credentials_tests import (
ProxySettingsUnitTests, OSCredsUnitTests)
from snaps.openstack.tests.create_flavor_tests import (
from snaps.openstack.tests.create_project_tests import (
CreateProjectSuccessTests, ProjectSettingsUnitTests,
CreateProjectUserTests)
+from snaps.openstack.tests.create_qos_tests import (QoSSettingsUnitTests,
+ CreateQoSTests)
from snaps.openstack.tests.create_router_tests import (
CreateRouterSuccessTests, CreateRouterNegativeTests,
RouterSettingsUnitTests)
UserSettingsUnitTests, CreateUserSuccessTests)
from snaps.openstack.tests.os_source_file_test import (
OSComponentTestCase, OSIntegrationTestCase)
+from snaps.openstack.utils.tests.cinder_utils_tests import (CinderSmokeTests,
+ CinderUtilsQoSTests)
from snaps.openstack.utils.tests.glance_utils_tests import (
GlanceSmokeTests, GlanceUtilsTests)
from snaps.openstack.utils.tests.heat_utils_tests import (
ResourceDomainObjectTests))
suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
StackSettingsUnitTests))
+ suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
+ QoSSpecDomainObjectTests))
suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
VmInstDomainObjectTests))
suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
FloatingIpDomainObjectTests))
+ suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
+ QoSSettingsUnitTests))
def add_openstack_client_tests(suite, os_creds, ext_net_name,
OSComponentTestCase.parameterize(
HeatSmokeTests, os_creds=os_creds, ext_net_name=ext_net_name,
log_level=log_level))
+ suite.addTest(
+ OSComponentTestCase.parameterize(
+ CinderSmokeTests, os_creds=os_creds, ext_net_name=ext_net_name,
+ log_level=log_level))
def add_openstack_api_tests(suite, os_creds, ext_net_name, use_keystone=True,
HeatUtilsCreateComplexStackTests, os_creds=os_creds,
ext_net_name=ext_net_name, log_level=log_level,
image_metadata=image_metadata))
+ suite.addTest(OSComponentTestCase.parameterize(
+ CinderUtilsQoSTests, os_creds=os_creds,
+ ext_net_name=ext_net_name, log_level=log_level,
+ image_metadata=image_metadata))
def add_openstack_integration_tests(suite, os_creds, ext_net_name,
ext_net_name=ext_net_name, use_keystone=use_keystone,
flavor_metadata=flavor_metadata, image_metadata=image_metadata,
log_level=log_level))
+ suite.addTest(OSIntegrationTestCase.parameterize(
+ CreateQoSTests, os_creds=os_creds,
+ ext_net_name=ext_net_name, use_keystone=use_keystone,
+ flavor_metadata=flavor_metadata, image_metadata=image_metadata,
+ log_level=log_level))
# VM Instances
suite.addTest(OSIntegrationTestCase.parameterize(
ext_net_name=ext_net_name, use_keystone=use_keystone,
flavor_metadata=flavor_metadata, image_metadata=image_metadata,
log_level=log_level))
- # TODO - uncomment after all OPNFV projects have cut a stable/euphrates
- # branch as this test was not meant to be exercised until F
- # suite.addTest(OSIntegrationTestCase.parameterize(
- # CreateComplexStackTests, os_creds=os_creds,
- # ext_net_name=ext_net_name, use_keystone=use_keystone,
- # flavor_metadata=flavor_metadata, image_metadata=image_metadata,
- # log_level=log_level))
+ suite.addTest(OSIntegrationTestCase.parameterize(
+ CreateComplexStackTests, os_creds=os_creds,
+ ext_net_name=ext_net_name, use_keystone=use_keystone,
+ flavor_metadata=flavor_metadata, image_metadata=image_metadata,
+ log_level=log_level))
suite.addTest(OSIntegrationTestCase.parameterize(
AnsibleProvisioningTests, os_creds=os_creds,
ext_net_name=ext_net_name, use_keystone=use_keystone,
ext_net_name=ext_net_name, log_level=log_level))
suite.addTest(OSIntegrationTestCase.parameterize(
CreateInstancePubPrivNetTests, os_creds=os_creds,
- ext_net_name=ext_net_name, use_keystone=use_keystone,
- flavor_metadata=flavor_metadata, image_metadata=image_metadata,
- log_level=log_level))
+ ext_net_name=ext_net_name, log_level=log_level))
Code Review / snaps.git / commitdiff