From 530153597deb5030c296358431d9549d13b7288b Mon Sep 17 00:00:00 2001 From: spisarski Date: Mon, 16 Oct 2017 15:54:51 -0600 Subject: First of several patches for adding volume support. * Added volume API version attribute to OSCreds * Created utility for interfacing with the Cinder APIs * Created QoS creator * Added new tests to test_suite_builder.py JIRA: SNAPS-195, SNAPS-194 Change-Id: I0c6a53b4cba6efea3e92d909b94b259fa07a35c3 Signed-off-by: spisarski --- snaps/openstack/utils/cinder_utils.py | 110 ++++++++++++++++ snaps/openstack/utils/tests/cinder_utils_tests.py | 154 ++++++++++++++++++++++ 2 files changed, 264 insertions(+) create mode 100644 snaps/openstack/utils/cinder_utils.py create mode 100644 snaps/openstack/utils/tests/cinder_utils_tests.py (limited to 'snaps/openstack/utils') diff --git a/snaps/openstack/utils/cinder_utils.py b/snaps/openstack/utils/cinder_utils.py new file mode 100644 index 0000000..5f847a1 --- /dev/null +++ b/snaps/openstack/utils/cinder_utils.py @@ -0,0 +1,110 @@ +# 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 + """ diff --git a/snaps/openstack/utils/tests/cinder_utils_tests.py b/snaps/openstack/utils/tests/cinder_utils_tests.py new file mode 100644 index 0000000..e6ad2a0 --- /dev/null +++ b/snaps/openstack/utils/tests/cinder_utils_tests.py @@ -0,0 +1,154 @@ +# 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)) -- cgit 1.2.3-korg