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.
21 from snaps.domain.flavor import Flavor
22 from snaps.domain.volume import (
23 Volume, VolumeType, VolumeTypeEncryption, QoSSpec)
24 from snaps.openstack import (
25 create_image, create_network, create_router, create_flavor,
26 create_keypairs, create_instance)
27 from snaps.openstack.create_network import (
28 NetworkSettings, OpenStackNetwork, SubnetSettings)
29 from snaps.openstack.create_qos import Consumer
30 from snaps.openstack.create_security_group import (
31 SecurityGroupRuleSettings, Direction, Protocol, OpenStackSecurityGroup,
32 SecurityGroupSettings)
33 from snaps.openstack.tests import openstack_tests
34 from snaps.openstack.tests.os_source_file_test import OSComponentTestCase
35 from snaps.openstack.utils import (
36 neutron_utils, settings_utils, nova_utils, glance_utils)
38 __author__ = 'spisarski'
40 logger = logging.getLogger('nova_utils_tests')
43 class SettingsUtilsNetworkingTests(OSComponentTestCase):
45 Tests the ability to reverse engineer NetworkSettings objects from existing
46 networks deployed to OpenStack
51 Instantiates OpenStack instances that cannot be spawned by Heat
53 guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
54 self.network_name = guid + '-net'
55 self.subnet_name = guid + '-subnet'
56 self.net_creator = None
57 self.neutron = neutron_utils.neutron_client(self.os_creds)
61 Cleans the image and downloaded image file
65 self.net_creator.clean()
69 def test_derive_net_settings_no_subnet(self):
71 Validates the utility function settings_utils#create_network_settings
72 returns an acceptable NetworkSettings object and ensures that the
73 new settings object will not cause the new OpenStackNetwork instance
74 to create another network
76 net_settings = NetworkSettings(name=self.network_name)
77 self.net_creator = OpenStackNetwork(self.os_creds, net_settings)
78 network = self.net_creator.create()
80 derived_settings = settings_utils.create_network_settings(
81 self.neutron, network)
83 self.assertIsNotNone(derived_settings)
84 self.assertEqual(net_settings.name, derived_settings.name)
85 self.assertEqual(net_settings.admin_state_up,
86 derived_settings.admin_state_up)
87 self.assertEqual(net_settings.external, derived_settings.external)
88 self.assertEqual(len(net_settings.subnet_settings),
89 len(derived_settings.subnet_settings))
91 net_creator = OpenStackNetwork(self.os_creds, derived_settings)
92 derived_network = net_creator.create()
94 self.assertEqual(network, derived_network)
96 def test_derive_net_settings_two_subnets(self):
98 Validates the utility function settings_utils#create_network_settings
99 returns an acceptable NetworkSettings object
101 subnet_settings = list()
102 subnet_settings.append(SubnetSettings(name='sub1', cidr='10.0.0.0/24'))
103 subnet_settings.append(SubnetSettings(name='sub2', cidr='10.0.1.0/24'))
104 net_settings = NetworkSettings(name=self.network_name,
105 subnet_settings=subnet_settings)
106 self.net_creator = OpenStackNetwork(self.os_creds, net_settings)
107 network = self.net_creator.create()
109 derived_settings = settings_utils.create_network_settings(
110 self.neutron, network)
112 self.assertIsNotNone(derived_settings)
113 self.assertEqual(net_settings.name, derived_settings.name)
114 self.assertEqual(net_settings.admin_state_up,
115 derived_settings.admin_state_up)
116 self.assertEqual(net_settings.external, derived_settings.external)
117 self.assertEqual(len(net_settings.subnet_settings),
118 len(derived_settings.subnet_settings))
120 # Validate the first subnet
121 orig_sub1 = net_settings.subnet_settings[0]
123 for derived_sub in derived_settings.subnet_settings:
124 if orig_sub1.name == derived_sub.name:
125 self.assertEqual(orig_sub1.cidr, derived_sub.cidr)
128 self.assertTrue(found)
130 # Validate the second subnet
131 orig_sub2 = net_settings.subnet_settings[1]
133 for derived_sub in derived_settings.subnet_settings:
134 if orig_sub2.name == derived_sub.name:
135 self.assertEqual(orig_sub2.cidr, derived_sub.cidr)
136 self.assertEqual(orig_sub2.ip_version, derived_sub.ip_version)
139 self.assertTrue(found)
142 class SettingsUtilsVmInstTests(OSComponentTestCase):
144 Tests the ability to reverse engineer VmInstanceSettings objects from
145 existing VMs/servers deployed to OpenStack
150 Instantiates the CreateImage object that is responsible for downloading
151 and creating an OS image file within OpenStack
153 self.nova = nova_utils.nova_client(self.os_creds)
154 self.glance = glance_utils.glance_client(self.os_creds)
155 self.neutron = neutron_utils.neutron_client(self.os_creds)
157 guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
158 self.keypair_priv_filepath = 'tmp/' + guid
159 self.keypair_pub_filepath = self.keypair_priv_filepath + '.pub'
160 self.keypair_name = guid + '-kp'
161 self.vm_inst_name = guid + '-inst'
162 self.test_file_local_path = 'tmp/' + guid + '-hello.txt'
163 self.port_1_name = guid + '-port-1'
164 self.port_2_name = guid + '-port-2'
165 self.floating_ip_name = guid + 'fip1'
167 # Setup members to cleanup just in case they don't get created
168 self.inst_creator = None
169 self.keypair_creator = None
170 self.sec_grp_creator = None
171 self.flavor_creator = None
172 self.router_creator = None
173 self.network_creator = None
174 self.image_creator = None
178 os_image_settings = openstack_tests.cirros_image_settings(
179 name=guid + '-' + '-image',
180 image_metadata=self.image_metadata)
181 self.image_creator = create_image.OpenStackImage(self.os_creds,
183 self.image_creator.create()
185 # First network is public
186 self.pub_net_config = openstack_tests.get_pub_net_config(
187 net_name=guid + '-pub-net', subnet_name=guid + '-pub-subnet',
188 router_name=guid + '-pub-router',
189 external_net=self.ext_net_name)
191 self.network_creator = create_network.OpenStackNetwork(
192 self.os_creds, self.pub_net_config.network_settings)
193 self.network_creator.create()
196 self.router_creator = create_router.OpenStackRouter(
197 self.os_creds, self.pub_net_config.router_settings)
198 self.router_creator.create()
201 self.flavor_creator = create_flavor.OpenStackFlavor(
203 create_flavor.FlavorSettings(name=guid + '-flavor-name',
204 ram=256, disk=1, vcpus=1))
205 self.flavor_creator.create()
208 self.keypair_creator = create_keypairs.OpenStackKeypair(
209 self.os_creds, create_keypairs.KeypairSettings(
210 name=self.keypair_name,
211 public_filepath=self.keypair_pub_filepath,
212 private_filepath=self.keypair_priv_filepath))
213 self.keypair_creator.create()
215 # Create Security Group
216 sec_grp_name = guid + '-sec-grp'
217 rule1 = SecurityGroupRuleSettings(sec_grp_name=sec_grp_name,
218 direction=Direction.ingress,
219 protocol=Protocol.icmp)
220 rule2 = SecurityGroupRuleSettings(sec_grp_name=sec_grp_name,
221 direction=Direction.ingress,
222 protocol=Protocol.tcp,
225 self.sec_grp_creator = OpenStackSecurityGroup(
227 SecurityGroupSettings(name=sec_grp_name,
228 rule_settings=[rule1, rule2]))
229 self.sec_grp_creator.create()
232 ports_settings = list()
233 ports_settings.append(
234 create_network.PortSettings(
235 name=self.port_1_name,
236 network_name=self.pub_net_config.network_settings.name))
238 instance_settings = create_instance.VmInstanceSettings(
239 name=self.vm_inst_name,
240 flavor=self.flavor_creator.flavor_settings.name,
241 port_settings=ports_settings,
242 floating_ip_settings=[create_instance.FloatingIpSettings(
243 name=self.floating_ip_name, port_name=self.port_1_name,
244 router_name=self.pub_net_config.router_settings.name)])
246 self.inst_creator = create_instance.OpenStackVmInstance(
247 self.os_creds, instance_settings,
248 self.image_creator.image_settings,
249 keypair_settings=self.keypair_creator.keypair_settings)
256 Cleans the created objects
258 if self.inst_creator:
260 self.inst_creator.clean()
264 if self.sec_grp_creator:
266 self.sec_grp_creator.clean()
270 if self.keypair_creator:
272 self.keypair_creator.clean()
276 if self.flavor_creator:
278 self.flavor_creator.clean()
282 if os.path.isfile(self.keypair_pub_filepath):
284 os.remove(self.keypair_pub_filepath)
288 if os.path.isfile(self.keypair_priv_filepath):
290 os.remove(self.keypair_priv_filepath)
294 if self.router_creator:
296 self.router_creator.clean()
300 if self.network_creator:
302 self.network_creator.clean()
306 if self.image_creator and not self.image_creator.image_settings.exists:
308 self.image_creator.clean()
312 if os.path.isfile(self.test_file_local_path):
313 os.remove(self.test_file_local_path)
315 # super(self.__class__, self).__clean__()
317 def test_derive_vm_inst_settings(self):
319 Validates the utility function settings_utils#create_vm_inst_settings
320 returns an acceptable VmInstanceSettings object
322 self.inst_creator.create(block=True)
324 server = nova_utils.get_server(
325 self.nova, vm_inst_settings=self.inst_creator.instance_settings)
326 derived_vm_settings = settings_utils.create_vm_inst_settings(
327 self.nova, self.neutron, server)
328 self.assertIsNotNone(derived_vm_settings)
329 self.assertIsNotNone(derived_vm_settings.port_settings)
330 self.assertIsNotNone(derived_vm_settings.floating_ip_settings)
332 def test_derive_image_settings(self):
334 Validates the utility function settings_utils#create_image_settings
335 returns an acceptable ImageSettings object
337 self.inst_creator.create(block=True)
339 server = nova_utils.get_server(
340 self.nova, vm_inst_settings=self.inst_creator.instance_settings)
341 derived_image_settings = settings_utils.determine_image_settings(
342 self.glance, server, [self.image_creator.image_settings])
343 self.assertIsNotNone(derived_image_settings)
344 self.assertEqual(self.image_creator.image_settings.name,
345 derived_image_settings.name)
348 class SettingsUtilsUnitTests(unittest.TestCase):
350 Exercises the settings_utils.py functions around volumes
353 def test_vol_settings_from_vol(self):
355 name='vol-name', volume_id='vol-id', description='desc', size=99,
356 vol_type='vol-type', availability_zone='zone1', multi_attach=True)
357 settings = settings_utils.create_volume_settings(volume)
358 self.assertEqual(volume.name, settings.name)
359 self.assertEqual(volume.description, settings.description)
360 self.assertEqual(volume.size, settings.size)
361 self.assertEqual(volume.type, settings.type_name)
362 self.assertEqual(volume.availability_zone, settings.availability_zone)
363 self.assertEqual(volume.multi_attach, settings.multi_attach)
365 def test_vol_type_settings_from_vol(self):
366 encryption = VolumeTypeEncryption(
367 volume_encryption_id='vol-encrypt-id', volume_type_id='vol-typ-id',
368 control_location='front-end', provider='FooClass', cipher='1',
370 qos_spec = QoSSpec(name='qos-spec-name', spec_id='qos-spec-id',
371 consumer=Consumer.back_end)
372 volume_type = VolumeType(
373 name='vol-type-name', volume_type_id='vol-type-id', public=True,
374 encryption=encryption, qos_spec=qos_spec)
376 settings = settings_utils.create_volume_type_settings(volume_type)
377 self.assertEqual(volume_type.name, settings.name)
378 self.assertEqual(volume_type.public, settings.public)
380 encrypt_settings = settings.encryption
381 self.assertIsNotNone(encrypt_settings)
382 self.assertEqual(encryption.control_location,
383 encrypt_settings.control_location.value)
384 self.assertEqual(encryption.cipher, encrypt_settings.cipher)
385 self.assertEqual(encryption.key_size, encrypt_settings.key_size)
387 self.assertEqual(qos_spec.name, settings.qos_spec_name)
389 def test_flavor_settings_from_flavor(self):
391 name='flavor-name', flavor_id='flavor-id', ram=99, disk=101,
392 vcpus=9, ephemeral=3, swap=5, rxtx_factor=7, is_public=False)
393 settings = settings_utils.create_flavor_settings(flavor)
394 self.assertEqual(flavor.name, settings.name)
395 self.assertEqual(flavor.id, settings.flavor_id)
396 self.assertEqual(flavor.ram, settings.ram)
397 self.assertEqual(flavor.disk, settings.disk)
398 self.assertEqual(flavor.vcpus, settings.vcpus)
399 self.assertEqual(flavor.ephemeral, settings.ephemeral)
400 self.assertEqual(flavor.swap, settings.swap)
401 self.assertEqual(flavor.rxtx_factor, settings.rxtx_factor)
402 self.assertEqual(flavor.is_public, settings.is_public)