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
  16 try:
  17     from urllib.request import URLError
  18 except ImportError:
  19     from urllib2 import URLError
  20
  21 import logging
  22 import unittest
  23 import uuid
  24
  25 from snaps.openstack import create_qos
  26 from snaps.openstack.create_qos import (QoSSettings, QoSSettingsError,
  27                                         Consumer)
  28 from snaps.openstack.tests.os_source_file_test import OSIntegrationTestCase
  29 from snaps.openstack.utils import cinder_utils
  30
  31 __author__ = 'spisarski'
  32
  33 logger = logging.getLogger('create_qos_tests')
  34
  35
  36 class QoSSettingsUnitTests(unittest.TestCase):
  37     """
  38     Tests the construction of the QoSSettings class
  39     """
  40
  41     def test_no_params(self):
  42         with self.assertRaises(QoSSettingsError):
  43             QoSSettings()
  44
  45     def test_empty_config(self):
  46         with self.assertRaises(QoSSettingsError):
  47             QoSSettings(**dict())
  48
  49     def test_name_only(self):
  50         with self.assertRaises(QoSSettingsError):
  51             QoSSettings(name='foo')
  52
  53     def test_config_with_name_only(self):
  54         with self.assertRaises(QoSSettingsError):
  55             QoSSettings(**{'name': 'foo'})
  56
  57     def test_invalid_consumer(self):
  58         with self.assertRaises(QoSSettingsError):
  59             QoSSettings(name='foo', consumer='bar')
  60
  61     def test_config_with_invalid_consumer(self):
  62         with self.assertRaises(QoSSettingsError):
  63             QoSSettings(**{'name': 'foo', 'consumer': 'bar'})
  64
  65     def test_name_consumer(self):
  66         settings = QoSSettings(name='foo', consumer=Consumer.front_end)
  67
  68         self.assertEqual('foo', settings.name)
  69         self.assertEqual(Consumer.front_end, settings.consumer)
  70         self.assertEqual(dict(), settings.specs)
  71
  72     def test_name_consumer_front_end_strings(self):
  73         settings = QoSSettings(name='foo', consumer='front-end')
  74
  75         self.assertEqual('foo', settings.name)
  76         self.assertEqual(Consumer.front_end, settings.consumer)
  77         self.assertEqual(dict(), settings.specs)
  78
  79     def test_name_consumer_back_end_strings(self):
  80         settings = QoSSettings(name='foo', consumer='back-end')
  81
  82         self.assertEqual('foo', settings.name)
  83         self.assertEqual(Consumer.back_end, settings.consumer)
  84         self.assertEqual(dict(), settings.specs)
  85
  86     def test_name_consumer_both_strings(self):
  87         settings = QoSSettings(name='foo', consumer='both')
  88
  89         self.assertEqual('foo', settings.name)
  90         self.assertEqual(Consumer.both, settings.consumer)
  91         self.assertEqual(dict(), settings.specs)
  92
  93     def test_all(self):
  94         specs = {'spec1': 'val1', 'spec2': 'val2'}
  95         settings = QoSSettings(name='foo', consumer=Consumer.both,
  96                                specs=specs)
  97
  98         self.assertEqual('foo', settings.name)
  99         self.assertEqual(Consumer.both, settings.consumer)
 100         self.assertEqual(specs, settings.specs)
 101
 102     def test_config_all(self):
 103         settings = QoSSettings(
 104             **{'name': 'foo', 'consumer': 'both', 'specs': {'spec1': 'val1'}})
 105
 106         self.assertEqual('foo', settings.name)
 107         self.assertEqual(Consumer.both, settings.consumer)
 108         self.assertEqual({'spec1': 'val1'}, settings.specs)
 109
 110
 111 class CreateQoSTests(OSIntegrationTestCase):
 112     """
 113     Test for the CreateQoS class defined in create_qos.py
 114     """
 115
 116     def setUp(self):
 117         """
 118         Instantiates the CreateQoS object that is responsible for
 119         downloading and creating an OS QoS Spec file within OpenStack
 120         """
 121         super(self.__class__, self).__start__()
 122
 123         guid = uuid.uuid4()
 124         self.qos_settings = QoSSettings(
 125             name=self.__class__.__name__ + '-' + str(guid),
 126             consumer=Consumer.both)
 127
 128         self.cinder = cinder_utils.cinder_client(self.os_creds)
 129         self.qos_creator = None
 130
 131     def tearDown(self):
 132         """
 133         Cleans the Qos Spec
 134         """
 135         if self.qos_creator:
 136             self.qos_creator.clean()
 137
 138         super(self.__class__, self).__clean__()
 139
 140     def test_create_qos(self):
 141         """
 142         Tests the creation of an OpenStack qos.
 143         """
 144         # Create QoS
 145         self.qos_creator = create_qos.OpenStackQoS(
 146             self.os_creds, self.qos_settings)
 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_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         self.qos_creator = create_qos.OpenStackQoS(
 163             self.os_creds, self.qos_settings)
 164         created_qos = self.qos_creator.create()
 165         self.assertIsNotNone(created_qos)
 166
 167         retrieved_qos = cinder_utils.get_qos(
 168             self.cinder, qos_settings=self.qos_settings)
 169         self.assertIsNotNone(retrieved_qos)
 170         self.assertEqual(created_qos, retrieved_qos)
 171
 172         # Delete QoS manually
 173         cinder_utils.delete_qos(self.cinder, created_qos)
 174
 175         self.assertIsNone(cinder_utils.get_qos(
 176             self.cinder, qos_settings=self.qos_settings))
 177
 178         # Must not raise an exception when attempting to cleanup non-existent
 179         # qos
 180         self.qos_creator.clean()
 181         self.assertIsNone(self.qos_creator.get_qos())
 182
 183     def test_create_same_qos(self):
 184         """
 185         Tests the creation of an OpenStack qos when one already exists.
 186         """
 187         # Create QoS
 188         self.qos_creator = create_qos.OpenStackQoS(
 189             self.os_creds, self.qos_settings)
 190         qos1 = self.qos_creator.create()
 191
 192         retrieved_qos = cinder_utils.get_qos(
 193             self.cinder, qos_settings=self.qos_settings)
 194         self.assertEqual(qos1, retrieved_qos)
 195
 196         # Should be retrieving the instance data
 197         os_qos_2 = create_qos.OpenStackQoS(
 198             self.os_creds, self.qos_settings)
 199         qos2 = os_qos_2.create()
 200         self.assertEqual(qos1, qos2)