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         qos_settings = QoSConfig(
 126             name=self.__class__.__name__ + '-' + str(guid),
 127             consumer=Consumer.both)
 128
 129         self.cinder = cinder_utils.cinder_client(self.admin_os_creds)
 130         self.qos_creator = create_qos.OpenStackQoS(
 131             self.admin_os_creds, qos_settings)
 132
 133     def tearDown(self):
 134         """
 135         Cleans the Qos Spec
 136         """
 137         if self.qos_creator:
 138             self.qos_creator.clean()
 139
 140         super(self.__class__, self).__clean__()
 141
 142     def test_create_qos(self):
 143         """
 144         Tests the creation of an OpenStack qos.
 145         """
 146         # Create QoS
 147         created_qos = self.qos_creator.create()
 148         self.assertIsNotNone(created_qos)
 149
 150         retrieved_qos = cinder_utils.get_qos(
 151             self.cinder, qos_settings=self.qos_creator.qos_settings)
 152
 153         self.assertIsNotNone(retrieved_qos)
 154         self.assertEqual(created_qos, retrieved_qos)
 155
 156     def test_create_delete_qos(self):
 157         """
 158         Tests the creation then deletion of an OpenStack QoS Spec to ensure
 159         clean() does not raise an Exception.
 160         """
 161         # Create QoS
 162         created_qos = self.qos_creator.create()
 163         self.assertIsNotNone(created_qos)
 164
 165         retrieved_qos = cinder_utils.get_qos(
 166             self.cinder, qos_settings=self.qos_creator.qos_settings)
 167         self.assertIsNotNone(retrieved_qos)
 168         self.assertEqual(created_qos, retrieved_qos)
 169
 170         # Delete QoS manually
 171         cinder_utils.delete_qos(self.cinder, created_qos)
 172
 173         self.assertIsNone(cinder_utils.get_qos(
 174             self.cinder, qos_settings=self.qos_creator.qos_settings))
 175
 176         # Must not raise an exception when attempting to cleanup non-existent
 177         # qos
 178         self.qos_creator.clean()
 179         self.assertIsNone(self.qos_creator.get_qos())
 180
 181     def test_create_same_qos(self):
 182         """
 183         Tests the creation of an OpenStack qos when one already exists.
 184         """
 185         # Create QoS
 186         qos1 = self.qos_creator.create()
 187
 188         retrieved_qos = cinder_utils.get_qos(
 189             self.cinder, qos_settings=self.qos_creator.qos_settings)
 190         self.assertEqual(qos1, retrieved_qos)
 191
 192         # Should be retrieving the instance data
 193         os_qos_2 = create_qos.OpenStackQoS(
 194             self.admin_os_creds, self.qos_creator.qos_settings)
 195         qos2 = os_qos_2.create()
 196         self.assertEqual(qos1, qos2)