- Volume Type - snaps.openstack.create\_qos.OpenStackQoS
- - snaps.openstack.create\_qos.QoSSettings
+ - snaps.openstack.qos.QoSConfig
- name - the volume type's name (required)
- consumer - the qos's consumer type of the enum type Consumer (required)
.. code:: python
- from snaps.openstack.create_qos import QoSSettings, OpenStackQoS
+ from snaps.openstack.qos import QoSConfig
+ from snaps.openstack.create_qos import OpenStackQoS
- qos_settings = QoSSettings(name='stack-name', consumer=Consumer.front-end)
+ qos_settings = QoSConfig(name='stack-name', consumer=Consumer.front-end)
qos_creator = OpenStackQoS(os_creds, vol_type_settings)
qos_creator.create()
Ensures that all required members are included when constructing a
VolumeTypeEncryption domain object (for Cinder)
+QoSConfigUnitTests
+------------------
+
+Ensures that all required members are included when constructing a
+QoSConfig object
+
QoSSettingsUnitTests
--------------------
Ensures that all required members are included when constructing a
-QoSSettings object
+deprecated QoSSettings object
QoSSpecDomainObjectTests
------------------------
from snaps.config.image import ImageConfig
from snaps.config.keypair import KeypairConfig
from snaps.config.project import ProjectConfig
+from snaps.config.qos import QoSConfig
from snaps.config.router import RouterConfig
from snaps.config.user import UserConfig
from snaps.openstack.create_flavor import OpenStackFlavor
from snaps.openstack.create_network import (
PortSettings, NetworkSettings, OpenStackNetwork)
from snaps.openstack.create_project import OpenStackProject
-from snaps.openstack.create_qos import QoSSettings, OpenStackQoS
+from snaps.openstack.create_qos import OpenStackQoS
from snaps.openstack.create_router import OpenStackRouter
from snaps.openstack.create_security_group import (
OpenStackSecurityGroup, SecurityGroupSettings)
# Create QoS specs
qos_dict = __create_instances(
- os_creds_dict, OpenStackQoS, QoSSettings,
+ os_creds_dict, OpenStackQoS, QoSConfig,
os_config.get('qos_specs'), 'qos_spec', clean, users_dict)
creators.append(qos_dict)
--- /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 enum
+
+
+class Consumer(enum.Enum):
+ """
+ QoS Specification consumer types
+ """
+ front_end = 'front-end'
+ back_end = 'back-end'
+ both = 'both'
+
+
+class QoSConfig(object):
+ def __init__(self, **kwargs):
+ """
+ Constructor
+ :param name: the qos's name (required)
+ :param consumer: the qos's consumer type of the enum type Consumer
+ (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 QoSConfigError(
+ "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
+ elif isinstance(consumer, str):
+ 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 QoSConfigError('Invalid Consumer - ' + proto_str)
+ else:
+ if consumer.value == 'front-end':
+ return Consumer.front_end
+ elif consumer.value == 'back-end':
+ return Consumer.back_end
+ elif consumer.value == 'both':
+ return Consumer.both
+ else:
+ raise QoSConfigError('Invalid Consumer - ' + consumer.value)
+
+
+class QoSConfigError(Exception):
+ """
+ Exception to be thrown when an qos settings are incorrect
+ """
+
+ def __init__(self, message):
+ Exception.__init__(self, message)
--- /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.config.qos import QoSConfig, QoSConfigError, Consumer
+
+
+class QoSConfigUnitTests(unittest.TestCase):
+ """
+ Tests the construction of the QoSConfig class
+ """
+
+ def test_no_params(self):
+ with self.assertRaises(QoSConfigError):
+ QoSConfig()
+
+ def test_empty_config(self):
+ with self.assertRaises(QoSConfigError):
+ QoSConfig(**dict())
+
+ def test_name_only(self):
+ with self.assertRaises(QoSConfigError):
+ QoSConfig(name='foo')
+
+ def test_config_with_name_only(self):
+ with self.assertRaises(QoSConfigError):
+ QoSConfig(**{'name': 'foo'})
+
+ def test_invalid_consumer(self):
+ with self.assertRaises(QoSConfigError):
+ QoSConfig(name='foo', consumer='bar')
+
+ def test_config_with_invalid_consumer(self):
+ with self.assertRaises(QoSConfigError):
+ QoSConfig(**{'name': 'foo', 'consumer': 'bar'})
+
+ def test_name_consumer(self):
+ settings = QoSConfig(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 = QoSConfig(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 = QoSConfig(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 = QoSConfig(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 = QoSConfig(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 = QoSConfig(
+ **{'name': 'foo', 'consumer': 'both', 'specs': {'spec1': 'val1'}})
+
+ self.assertEqual('foo', settings.name)
+ self.assertEqual(Consumer.both, settings.consumer)
+ self.assertEqual({'spec1': 'val1'}, settings.specs)
import enum
from cinderclient.exceptions import NotFound
+from snaps.config.qos import QoSConfig
from snaps.openstack.openstack_creator import OpenStackVolumeObject
from snaps.openstack.utils import cinder_utils
class Consumer(enum.Enum):
"""
QoS Specification consumer types
+ deprecated - use snaps.config.qos.Consumer
"""
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 of the enum type Consumer
- (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):
+class QoSSettings(QoSConfig):
"""
- Exception to be thrown when an qos settings are incorrect
+ Class to hold the configuration settings required for creating OpenStack
+ QoS objects
+ deprecated
"""
- def __init__(self, message):
- Exception.__init__(self, message)
+ def __init__(self, **kwargs):
+ from warnings import warn
+ warn('Use snaps.config.qos.QoSConfig instead',
+ DeprecationWarning)
+ super(self.__class__, self).__init__(**kwargs)
class QoSCreationError(Exception):
# 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.
+from snaps.config.qos import Consumer, QoSConfigError, QoSConfig
+from snaps.openstack.create_qos import Consumer as Consumer_old
try:
from urllib.request import URLError
import uuid
from snaps.openstack import create_qos
-from snaps.openstack.create_qos import (QoSSettings, QoSSettingsError,
- Consumer)
+from snaps.openstack.create_qos import QoSSettings
from snaps.openstack.tests.os_source_file_test import OSIntegrationTestCase
from snaps.openstack.utils import cinder_utils
"""
def test_no_params(self):
- with self.assertRaises(QoSSettingsError):
+ with self.assertRaises(QoSConfigError):
QoSSettings()
def test_empty_config(self):
- with self.assertRaises(QoSSettingsError):
+ with self.assertRaises(QoSConfigError):
QoSSettings(**dict())
def test_name_only(self):
- with self.assertRaises(QoSSettingsError):
+ with self.assertRaises(QoSConfigError):
QoSSettings(name='foo')
def test_config_with_name_only(self):
- with self.assertRaises(QoSSettingsError):
+ with self.assertRaises(QoSConfigError):
QoSSettings(**{'name': 'foo'})
def test_invalid_consumer(self):
- with self.assertRaises(QoSSettingsError):
+ with self.assertRaises(QoSConfigError):
QoSSettings(name='foo', consumer='bar')
def test_config_with_invalid_consumer(self):
- with self.assertRaises(QoSSettingsError):
+ with self.assertRaises(QoSConfigError):
QoSSettings(**{'name': 'foo', 'consumer': 'bar'})
def test_name_consumer(self):
- settings = QoSSettings(name='foo', consumer=Consumer.front_end)
+ settings = QoSSettings(name='foo', consumer=Consumer_old.front_end)
self.assertEqual('foo', settings.name)
- self.assertEqual(Consumer.front_end, settings.consumer)
+ self.assertEqual(Consumer_old.front_end.value, settings.consumer.value)
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(Consumer_old.front_end.value, settings.consumer.value)
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(Consumer_old.back_end.value, settings.consumer.value)
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(Consumer_old.both.value, settings.consumer.value)
self.assertEqual(dict(), settings.specs)
def test_all(self):
specs = {'spec1': 'val1', 'spec2': 'val2'}
- settings = QoSSettings(name='foo', consumer=Consumer.both,
+ settings = QoSSettings(name='foo', consumer=Consumer_old.both,
specs=specs)
self.assertEqual('foo', settings.name)
- self.assertEqual(Consumer.both, settings.consumer)
+ self.assertEqual(Consumer_old.both.value, settings.consumer.value)
self.assertEqual(specs, settings.specs)
def test_config_all(self):
super(self.__class__, self).__start__()
guid = uuid.uuid4()
- self.qos_settings = QoSSettings(
+ self.qos_settings = QoSConfig(
name=self.__class__.__name__ + '-' + str(guid),
consumer=Consumer.both)
# 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.
-from snaps.openstack.create_qos import QoSSettings, Consumer, OpenStackQoS
+from snaps.config.qos import QoSConfig, Consumer
+from snaps.openstack.create_qos import OpenStackQoS
try:
from urllib.request import URLError
import unittest
import uuid
-from snaps.openstack import create_volume_type
from snaps.openstack.create_volume_type import (
VolumeTypeSettings, VolumeTypeSettingsError, VolumeTypeEncryptionSettings,
- ControlLocation)
+ ControlLocation, OpenStackVolumeType)
from snaps.openstack.tests.os_source_file_test import OSIntegrationTestCase
from snaps.openstack.utils import cinder_utils
class CreateSimpleVolumeTypeSuccessTests(OSIntegrationTestCase):
"""
- Test for the OpenStackVolumeType class defined in create_volume_type.py
+ Test for the OpenStackVolumeType class defined in py
without any QoS Specs or Encryption
"""
Tests the creation of an OpenStack volume.
"""
# Create VolumeType
- self.volume_type_creator = create_volume_type.OpenStackVolumeType(
+ self.volume_type_creator = OpenStackVolumeType(
self.os_creds, self.volume_type_settings)
created_volume_type = self.volume_type_creator.create()
self.assertIsNotNone(created_volume_type)
clean() does not raise an Exception.
"""
# Create VolumeType
- self.volume_type_creator = create_volume_type.OpenStackVolumeType(
+ self.volume_type_creator = OpenStackVolumeType(
self.os_creds, self.volume_type_settings)
created_volume_type = self.volume_type_creator.create()
self.assertIsNotNone(created_volume_type)
Tests the creation of an OpenStack volume_type when one already exists.
"""
# Create VolumeType
- self.volume_type_creator = create_volume_type.OpenStackVolumeType(
+ self.volume_type_creator = OpenStackVolumeType(
self.os_creds, self.volume_type_settings)
volume_type1 = self.volume_type_creator.create()
self.assertEqual(volume_type1, retrieved_volume_type)
# Should be retrieving the instance data
- os_volume_type_2 = create_volume_type.OpenStackVolumeType(
+ os_volume_type_2 = OpenStackVolumeType(
self.os_creds, self.volume_type_settings)
volume_type2 = os_volume_type_2.create()
self.assertEqual(volume_type2, volume_type2)
self.volume_type_name = guid + '-vol_type'
self.volume_type_creator = None
- qos_settings = QoSSettings(
+ qos_settings = QoSConfig(
name=guid + '-qos-spec', consumer=Consumer.both)
self.qos_creator = OpenStackQoS(self.os_creds, qos_settings)
self.qos_creator.create()
"""
Creates a Volume Type object with an associated QoS Spec
"""
- self.volume_type_creator = create_volume_type.OpenStackVolumeType(
+ self.volume_type_creator = OpenStackVolumeType(
self.os_creds,
VolumeTypeSettings(
name=self.volume_type_name,
encryption_settings = VolumeTypeEncryptionSettings(
name='foo', provider_class='bar',
control_location=ControlLocation.back_end)
- self.volume_type_creator = create_volume_type.OpenStackVolumeType(
+ self.volume_type_creator = OpenStackVolumeType(
self.os_creds,
VolumeTypeSettings(
name=self.volume_type_name,
encryption_settings = VolumeTypeEncryptionSettings(
name='foo', provider_class='bar',
control_location=ControlLocation.back_end)
- self.volume_type_creator = create_volume_type.OpenStackVolumeType(
+ self.volume_type_creator = OpenStackVolumeType(
self.os_creds,
VolumeTypeSettings(
name=self.volume_type_name,
import time
from cinderclient.exceptions import NotFound, BadRequest
+from snaps.config.qos import Consumer, QoSConfig
from snaps.openstack import create_volume
-from snaps.openstack.create_qos import QoSSettings, Consumer
from snaps.openstack.create_volume import VolumeSettings
from snaps.openstack.create_volume_type import (
VolumeTypeSettings, VolumeTypeEncryptionSettings, ControlLocation)
"""
Tests the cinder_utils.create_qos()
"""
- qos_settings = QoSSettings(name=self.qos_name, specs=self.specs,
- consumer=Consumer.both)
+ qos_settings = QoSConfig(
+ name=self.qos_name, specs=self.specs, consumer=Consumer.both)
self.qos = cinder_utils.create_qos(self.cinder, qos_settings)
self.assertIsNotNone(self.qos)
"""
Tests the cinder_utils.create_qos()
"""
- qos_settings = QoSSettings(name=self.qos_name, specs=self.specs,
- consumer=Consumer.front_end)
+ qos_settings = QoSConfig(
+ 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)
"""
Tests the cinder_utils.create_qos()
"""
- qos_settings = QoSSettings(name=self.qos_name, specs=self.specs,
- consumer=Consumer.back_end)
+ qos_settings = QoSConfig(
+ 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)
"""
Tests the cinder_utils.create_qos()
"""
- qos_settings = QoSSettings(name=self.qos_name, consumer=Consumer.both)
+ qos_settings = QoSConfig(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)
self.vol_type_name = self.__class__.__name__ + '-' + str(guid)
self.specs = {'foo': 'bar'}
self.cinder = cinder_utils.cinder_client(self.os_creds)
- qos_settings = QoSSettings(name=self.qos_name, specs=self.specs,
- consumer=Consumer.both)
+ qos_settings = QoSConfig(
+ name=self.qos_name, specs=self.specs, consumer=Consumer.both)
self.qos = cinder_utils.create_qos(self.cinder, qos_settings)
self.volume_type = None
from snaps.config.flavor import FlavorConfig
from snaps.config.keypair import KeypairConfig
+from snaps.config.qos import Consumer
from snaps.domain.flavor import Flavor
from snaps.domain.volume import (
Volume, VolumeType, VolumeTypeEncryption, QoSSpec)
create_keypairs, create_instance)
from snaps.openstack.create_network import (
NetworkSettings, OpenStackNetwork, SubnetSettings)
-from snaps.openstack.create_qos import Consumer
from snaps.openstack.create_security_group import (
SecurityGroupRuleSettings, Direction, Protocol, OpenStackSecurityGroup,
SecurityGroupSettings)
import logging
import unittest
+from snaps.config.tests.qos_tests import QoSConfigUnitTests
from snaps.config.tests.stack_tests import StackConfigUnitTests
from snaps.config.tests.router_tests import RouterConfigUnitTests
from snaps.config.tests.user_tests import UserConfigUnitTests
VmInstDomainObjectTests))
suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
FloatingIpDomainObjectTests))
+ suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
+ QoSConfigUnitTests))
suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
QoSSettingsUnitTests))
suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
Code Review / snaps.git / commitdiff