snaps/openstack/tests/create_qos_tests.py

   1 # Copyright (c) 2017 Cable Television Laboratories, Inc. ("CableLabs")
   2 #                    and others.  All rights reserved.
   3 #
   4 # Licensed under the Apache License, Version 2.0 (the "License");
   5 # you may not use this file except in compliance with the License.
   6 # You may obtain a copy of the License at:
   7 #
   8 #     http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 # Unless required by applicable law or agreed to in writing, software
  11 # distributed under the License is distributed on an "AS IS" BASIS,
  12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 # See the License for the specific language governing permissions and
  14 # limitations under the License.
  15 from snaps.config.qos import Consumer, QoSConfigError, QoSConfig
  16 from snaps.openstack.create_qos import Consumer as Consumer_old
  17
  18 try:
  19     from urllib.request import URLError
  20 except ImportError:
  21     from urllib2 import URLError
  22
  23 import logging
  24 import unittest
  25 import uuid
  26
  27 from snaps.openstack import create_qos
  28 from snaps.openstack.create_qos import QoSSettings
  29 from snaps.openstack.tests.os_source_file_test import OSIntegrationTestCase
  30 from snaps.openstack.utils import cinder_utils
  31
  32 __author__ = 'spisarski'
  33
  34 logger = logging.getLogger('create_qos_tests')
  35
  36
  37 class QoSSettingsUnitTests(unittest.TestCase):
  38     """
  39     Tests the construction of the QoSSettings class
  40     """
  41
  42     def test_no_params(self):
  43         with self.assertRaises(QoSConfigError):
  44             QoSSettings()
  45
  46     def test_empty_config(self):
  47         with self.assertRaises(QoSConfigError):
  48             QoSSettings(**dict())
  49
  50     def test_name_only(self):
  51         with self.assertRaises(QoSConfigError):
  52             QoSSettings(name='foo')
  53
  54     def test_config_with_name_only(self):
  55         with self.assertRaises(QoSConfigError):
  56             QoSSettings(**{'name': 'foo'})
  57
  58     def test_invalid_consumer(self):
  59         with self.assertRaises(QoSConfigError):
  60             QoSSettings(name='foo', consumer='bar')
  61
  62     def test_config_with_invalid_consumer(self):
  63         with self.assertRaises(QoSConfigError):
  64             QoSSettings(**{'name': 'foo', 'consumer': 'bar'})
  65
  66     def test_name_consumer(self):
  67         settings = QoSSettings(name='foo', consumer=Consumer_old.front_end)
  68
  69         self.assertEqual('foo', settings.name)
  70         self.assertEqual(Consumer_old.front_end.value, settings.consumer.value)
  71         self.assertEqual(dict(), settings.specs)
  72
  73     def test_name_consumer_front_end_strings(self):
  74         settings = QoSSettings(name='foo', consumer='front-end')
  75
  76         self.assertEqual('foo', settings.name)
  77         self.assertEqual(Consumer_old.front_end.value, settings.consumer.value)
  78         self.assertEqual(dict(), settings.specs)
  79
  80     def test_name_consumer_back_end_strings(self):
  81         settings = QoSSettings(name='foo', consumer='back-end')
  82
  83         self.assertEqual('foo', settings.name)
  84         self.assertEqual(Consumer_old.back_end.value, settings.consumer.value)
  85         self.assertEqual(dict(), settings.specs)
  86
  87     def test_name_consumer_both_strings(self):
  88         settings = QoSSettings(name='foo', consumer='both')
  89
  90         self.assertEqual('foo', settings.name)
  91         self.assertEqual(Consumer_old.both.value, settings.consumer.value)
  92         self.assertEqual(dict(), settings.specs)
  93
  94     def test_all(self):
  95         specs = {'spec1': 'val1', 'spec2': 'val2'}
  96         settings = QoSSettings(name='foo', consumer=Consumer_old.both,
  97                                specs=specs)
  98
  99         self.assertEqual('foo', settings.name)
 100         self.assertEqual(Consumer_old.both.value, settings.consumer.value)
 101         self.assertEqual(specs, settings.specs)
 102
 103     def test_config_all(self):
 104         settings = QoSSettings(
 105             **{'name': 'foo', 'consumer': 'both', 'specs': {'spec1': 'val1'}})
 106
 107         self.assertEqual('foo', settings.name)
 108         self.assertEqual(Consumer.both, settings.consumer)
 109         self.assertEqual({'spec1': 'val1'}, settings.specs)
 110
 111
 112 class CreateQoSTests(OSIntegrationTestCase):
 113     """
 114     Test for the CreateQoS class defined in create_qos.py
 115     """
 116
 117     def setUp(self):
 118         """
 119         Instantiates the CreateQoS object that is responsible for
 120         downloading and creating an OS QoS Spec file within OpenStack
 121         """
 122         super(self.__class__, self).__start__()
 123
 124         guid = uuid.uuid4()
 125         self.qos_settings = QoSConfig(
 126             name=self.__class__.__name__ + '-' + str(guid),
 127             consumer=Consumer.both)
 128
 129         self.cinder = cinder_utils.cinder_client(self.os_creds)
 130         self.qos_creator = None
 131
 132     def tearDown(self):
 133         """
 134         Cleans the Qos Spec
 135         """
 136         if self.qos_creator:
 137             self.qos_creator.clean()
 138
 139         super(self.__class__, self).__clean__()
 140
 141     def test_create_qos(self):
 142         """
 143         Tests the creation of an OpenStack qos.
 144         """
 145         # Create QoS
 146         self.qos_creator = create_qos.OpenStackQoS(
 147             self.os_creds, self.qos_settings)
 148         created_qos = self.qos_creator.create()
 149         self.assertIsNotNone(created_qos)
 150
 151         retrieved_qos = cinder_utils.get_qos(
 152             self.cinder, qos_settings=self.qos_settings)
 153
 154         self.assertIsNotNone(retrieved_qos)
 155         self.assertEqual(created_qos, retrieved_qos)
 156
 157     def test_create_delete_qos(self):
 158         """
 159         Tests the creation then deletion of an OpenStack QoS Spec to ensure
 160         clean() does not raise an Exception.
 161         """
 162         # Create QoS
 163         self.qos_creator = create_qos.OpenStackQoS(
 164             self.os_creds, self.qos_settings)
 165         created_qos = self.qos_creator.create()
 166         self.assertIsNotNone(created_qos)
 167
 168         retrieved_qos = cinder_utils.get_qos(
 169             self.cinder, qos_settings=self.qos_settings)
 170         self.assertIsNotNone(retrieved_qos)
 171         self.assertEqual(created_qos, retrieved_qos)
 172
 173         # Delete QoS manually
 174         cinder_utils.delete_qos(self.cinder, created_qos)
 175
 176         self.assertIsNone(cinder_utils.get_qos(
 177             self.cinder, qos_settings=self.qos_settings))
 178
 179         # Must not raise an exception when attempting to cleanup non-existent
 180         # qos
 181         self.qos_creator.clean()
 182         self.assertIsNone(self.qos_creator.get_qos())
 183
 184     def test_create_same_qos(self):
 185         """
 186         Tests the creation of an OpenStack qos when one already exists.
 187         """
 188         # Create QoS
 189         self.qos_creator = create_qos.OpenStackQoS(
 190             self.os_creds, self.qos_settings)
 191         qos1 = self.qos_creator.create()
 192
 193         retrieved_qos = cinder_utils.get_qos(
 194             self.cinder, qos_settings=self.qos_settings)
 195         self.assertEqual(qos1, retrieved_qos)
 196
 197         # Should be retrieving the instance data
 198         os_qos_2 = create_qos.OpenStackQoS(
 199             self.os_creds, self.qos_settings)
 200         qos2 = os_qos_2.create()
 201         self.assertEqual(qos1, qos2)