1 # Copyright (c) 2017 Cable Television Laboratories, Inc. ("CableLabs")
2 # and others. All rights reserved.
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:
8 # http://www.apache.org/licenses/LICENSE-2.0
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.openstack.create_qos import QoSSettings, Consumer, OpenStackQoS
18 from urllib.request import URLError
20 from urllib2 import URLError
26 from snaps.openstack import create_volume_type
27 from snaps.openstack.create_volume_type import (
28 VolumeTypeSettings, VolumeTypeSettingsError, VolumeTypeEncryptionSettings,
30 from snaps.openstack.tests.os_source_file_test import OSIntegrationTestCase
31 from snaps.openstack.utils import cinder_utils
33 __author__ = 'spisarski'
35 logger = logging.getLogger('create_volume_type_tests')
38 class VolumeTypeSettingsUnitTests(unittest.TestCase):
40 Tests the construction of the VolumeTypeSettings class
43 def test_no_params(self):
44 with self.assertRaises(VolumeTypeSettingsError):
47 def test_empty_config(self):
48 with self.assertRaises(VolumeTypeSettingsError):
49 VolumeTypeSettings(**dict())
51 def test_name_only(self):
52 settings = VolumeTypeSettings(name='foo')
53 self.assertEqual('foo', settings.name)
54 self.assertIsNone(settings.description)
55 self.assertIsNone(settings.qos_spec_name)
56 self.assertIsNone(settings.encryption)
57 self.assertFalse(settings.public)
59 def test_config_with_name_only(self):
60 settings = VolumeTypeSettings(**{'name': 'foo'})
61 self.assertEqual('foo', settings.name)
62 self.assertIsNone(settings.description)
63 self.assertIsNone(settings.qos_spec_name)
64 self.assertIsNone(settings.encryption)
65 self.assertFalse(settings.public)
68 encryption_settings = VolumeTypeEncryptionSettings(
69 name='foo', provider_class='bar',
70 control_location=ControlLocation.back_end)
71 settings = VolumeTypeSettings(
72 name='foo', description='desc', encryption=encryption_settings,
73 qos_spec_name='spec_name', public=True)
74 self.assertEqual('foo', settings.name)
75 self.assertEqual('desc', settings.description)
76 self.assertEqual('spec_name', settings.qos_spec_name)
77 self.assertEqual(encryption_settings, settings.encryption)
78 self.assertTrue(True, settings.public)
80 def test_all_string(self):
81 encryption_settings = {
82 'name': 'foo', 'provider_class': 'bar',
83 'control_location': 'back-end'}
84 settings = VolumeTypeSettings(
85 name='foo', description='desc', encryption=encryption_settings,
86 qos_spec_name='spec_name', public='true')
87 self.assertEqual('foo', settings.name)
88 self.assertEqual('desc', settings.description)
89 self.assertEqual('spec_name', settings.qos_spec_name)
90 self.assertEqual(VolumeTypeEncryptionSettings(**encryption_settings),
92 self.assertTrue(settings.public)
94 def test_config_all(self):
95 encryption_settings = {
96 'name': 'foo', 'provider_class': 'bar',
97 'control_location': 'back-end'}
98 settings = VolumeTypeSettings(
99 **{'name': 'foo', 'description': 'desc',
100 'encryption': encryption_settings,
101 'qos_spec_name': 'spec_name', 'public': 'false'})
102 self.assertEqual('foo', settings.name)
103 self.assertEqual('desc', settings.description)
104 self.assertEqual('spec_name', settings.qos_spec_name)
105 self.assertEqual(VolumeTypeEncryptionSettings(**encryption_settings),
107 self.assertFalse(settings.public)
110 class CreateSimpleVolumeTypeSuccessTests(OSIntegrationTestCase):
112 Test for the OpenStackVolumeType class defined in create_volume_type.py
113 without any QoS Specs or Encryption
118 Instantiates the CreateVolumeType object that is responsible for
119 downloading and creating an OS volume type file within OpenStack
121 super(self.__class__, self).__start__()
124 self.volume_type_settings = VolumeTypeSettings(
125 name=self.__class__.__name__ + '-' + str(guid))
127 self.cinder = cinder_utils.cinder_client(self.os_creds)
128 self.volume_type_creator = None
132 Cleans the volume type
134 if self.volume_type_creator:
135 self.volume_type_creator.clean()
137 super(self.__class__, self).__clean__()
139 def test_create_volume_type(self):
141 Tests the creation of an OpenStack volume.
144 self.volume_type_creator = create_volume_type.OpenStackVolumeType(
145 self.os_creds, self.volume_type_settings)
146 created_volume_type = self.volume_type_creator.create()
147 self.assertIsNotNone(created_volume_type)
148 self.assertEqual(self.volume_type_settings.name,
149 created_volume_type.name)
151 retrieved_volume_type1 = cinder_utils.get_volume_type(
152 self.cinder, volume_type_settings=self.volume_type_settings)
153 self.assertIsNotNone(retrieved_volume_type1)
154 self.assertEqual(created_volume_type, retrieved_volume_type1)
156 retrieved_volume_type2 = cinder_utils.get_volume_type_by_id(
157 self.cinder, created_volume_type.id)
158 self.assertEqual(created_volume_type, retrieved_volume_type2)
160 def test_create_delete_volume_type(self):
162 Tests the creation then deletion of an OpenStack volume type to ensure
163 clean() does not raise an Exception.
166 self.volume_type_creator = create_volume_type.OpenStackVolumeType(
167 self.os_creds, self.volume_type_settings)
168 created_volume_type = self.volume_type_creator.create()
169 self.assertIsNotNone(created_volume_type)
171 retrieved_volume_type = cinder_utils.get_volume_type(
172 self.cinder, volume_type_settings=self.volume_type_settings)
173 self.assertIsNotNone(retrieved_volume_type)
174 self.assertEqual(created_volume_type, retrieved_volume_type)
176 # Delete VolumeType manually
177 cinder_utils.delete_volume_type(self.cinder, created_volume_type)
179 self.assertIsNone(cinder_utils.get_volume_type(
180 self.cinder, volume_type_settings=self.volume_type_settings))
182 # Must not throw an exception when attempting to cleanup non-existent
184 self.volume_type_creator.clean()
185 self.assertIsNone(self.volume_type_creator.get_volume_type())
187 def test_create_same_volume_type(self):
189 Tests the creation of an OpenStack volume_type when one already exists.
192 self.volume_type_creator = create_volume_type.OpenStackVolumeType(
193 self.os_creds, self.volume_type_settings)
194 volume_type1 = self.volume_type_creator.create()
196 retrieved_volume_type = cinder_utils.get_volume_type(
197 self.cinder, volume_type_settings=self.volume_type_settings)
198 self.assertEqual(volume_type1, retrieved_volume_type)
200 # Should be retrieving the instance data
201 os_volume_type_2 = create_volume_type.OpenStackVolumeType(
202 self.os_creds, self.volume_type_settings)
203 volume_type2 = os_volume_type_2.create()
204 self.assertEqual(volume_type2, volume_type2)
207 class CreateVolumeTypeComplexTests(OSIntegrationTestCase):
209 Test cases for the CreateVolumeType class that include QoS Specs and/or
214 super(self.__class__, self).__start__()
216 self.cinder = cinder_utils.cinder_client(self.os_creds)
218 guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
220 self.volume_type_name = guid + '-vol_type'
221 self.volume_type_creator = None
223 qos_settings = QoSSettings(
224 name=guid + '-qos-spec', consumer=Consumer.both)
225 self.qos_creator = OpenStackQoS(self.os_creds, qos_settings)
226 self.qos_creator.create()
229 if self.volume_type_creator:
230 self.volume_type_creator.clean()
233 self.qos_creator.clean()
235 super(self.__class__, self).__clean__()
237 def test_volume_type_with_qos(self):
239 Creates a Volume Type object with an associated QoS Spec
241 self.volume_type_creator = create_volume_type.OpenStackVolumeType(
244 name=self.volume_type_name,
245 qos_spec_name=self.qos_creator.qos_settings.name))
247 vol_type = self.volume_type_creator.create()
248 self.assertEqual(self.volume_type_creator.volume_type_settings.name,
250 self.assertEqual(self.volume_type_creator.volume_type_settings.name,
252 self.assertIsNotNone(vol_type.qos_spec)
254 self.volume_type_creator.volume_type_settings.qos_spec_name,
255 vol_type.qos_spec.name)
256 self.assertIsNone(vol_type.encryption)
258 vol_type_query = cinder_utils.get_volume_type_by_id(
259 self.cinder, vol_type.id)
260 self.assertIsNotNone(vol_type_query)
261 self.assertEqual(vol_type, vol_type_query)
263 def test_volume_type_with_encryption(self):
265 Creates a Volume Type object with encryption
267 encryption_settings = VolumeTypeEncryptionSettings(
268 name='foo', provider_class='bar',
269 control_location=ControlLocation.back_end)
270 self.volume_type_creator = create_volume_type.OpenStackVolumeType(
273 name=self.volume_type_name,
274 encryption=encryption_settings))
276 vol_type = self.volume_type_creator.create()
277 self.assertEqual(self.volume_type_creator.volume_type_settings.name,
279 self.assertEqual(self.volume_type_creator.volume_type_settings.name,
281 self.assertIsNone(vol_type.qos_spec)
282 self.assertIsNotNone(vol_type.encryption)
284 self.assertEqual(encryption_settings.control_location.value,
285 vol_type.encryption.control_location)
287 vol_type_query = cinder_utils.get_volume_type_by_id(
288 self.cinder, vol_type.id)
289 self.assertIsNotNone(vol_type_query)
290 self.assertEqual(vol_type, vol_type_query)
292 def test_volume_type_with_qos_and_encryption(self):
294 Creates a Volume Type object with encryption and an associated QoS Spec
296 encryption_settings = VolumeTypeEncryptionSettings(
297 name='foo', provider_class='bar',
298 control_location=ControlLocation.back_end)
299 self.volume_type_creator = create_volume_type.OpenStackVolumeType(
302 name=self.volume_type_name,
303 encryption=encryption_settings,
304 qos_spec_name=self.qos_creator.qos_settings.name))
306 vol_type = self.volume_type_creator.create()
307 self.assertEqual(self.volume_type_creator.volume_type_settings.name,
309 self.assertEqual(self.volume_type_creator.volume_type_settings.name,
311 self.assertIsNotNone(vol_type.qos_spec)
313 self.volume_type_creator.volume_type_settings.qos_spec_name,
314 vol_type.qos_spec.name)
315 self.assertIsNotNone(vol_type.encryption)
317 self.assertEqual(encryption_settings.control_location.value,
318 vol_type.encryption.control_location)
320 vol_type_query = cinder_utils.get_volume_type_by_id(
321 self.cinder, vol_type.id)
322 self.assertIsNotNone(vol_type_query)
323 self.assertEqual(vol_type, vol_type_query)