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.config.network import SubnetConfig, NetworkConfig, PortConfig
22 from snaps.config.flavor import FlavorConfig
23 from snaps.config.keypair import KeypairConfig
24 from snaps.config.qos import Consumer
25 from snaps.config.security_group import (
26 SecurityGroupRuleConfig, Direction, Protocol, SecurityGroupConfig)
27 from snaps.config.vm_inst import VmInstanceConfig, FloatingIpConfig
28 from snaps.domain.flavor import Flavor
29 from snaps.domain.volume import (
30 Volume, VolumeType, VolumeTypeEncryption, QoSSpec)
31 from snaps.openstack import (
32 create_image, create_network, create_router, create_flavor,
33 create_keypairs, create_instance)
34 from snaps.openstack.create_qos import Consumer
35 from snaps.openstack.create_network import OpenStackNetwork
36 from snaps.openstack.create_security_group import OpenStackSecurityGroup
37 from snaps.openstack.tests import openstack_tests
38 from snaps.openstack.tests.os_source_file_test import OSComponentTestCase
39 from snaps.openstack.utils import (
40 neutron_utils, settings_utils, nova_utils, glance_utils, keystone_utils)
42 __author__ = 'spisarski'
44 logger = logging.getLogger('nova_utils_tests')
47 class SettingsUtilsNetworkingTests(OSComponentTestCase):
49 Tests the ability to reverse engineer NetworkConfig objects from existing
50 networks deployed to OpenStack
55 Instantiates OpenStack instances that cannot be spawned by Heat
57 guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
58 self.network_name = guid + '-net'
59 self.subnet_name = guid + '-subnet'
60 self.net_creator = None
61 self.neutron = neutron_utils.neutron_client(
62 self.os_creds, self.os_session)
66 Cleans the image and downloaded image file
70 self.net_creator.clean()
74 super(self.__class__, self).__clean__()
76 def test_derive_net_settings_no_subnet(self):
78 Validates the utility function settings_utils#create_network_config
79 returns an acceptable NetworkConfig object and ensures that the
80 new settings object will not cause the new OpenStackNetwork instance
81 to create another network
83 net_settings = NetworkConfig(name=self.network_name)
84 self.net_creator = OpenStackNetwork(self.os_creds, net_settings)
85 network = self.net_creator.create()
87 derived_settings = settings_utils.create_network_config(
88 self.neutron, network)
90 self.assertIsNotNone(derived_settings)
91 self.assertEqual(net_settings.name, derived_settings.name)
92 self.assertEqual(net_settings.admin_state_up,
93 derived_settings.admin_state_up)
94 self.assertEqual(net_settings.external, derived_settings.external)
95 self.assertEqual(len(net_settings.subnet_settings),
96 len(derived_settings.subnet_settings))
98 net_creator = OpenStackNetwork(self.os_creds, derived_settings)
99 derived_network = net_creator.create()
101 self.assertEqual(network, derived_network)
103 def test_derive_net_settings_two_subnets(self):
105 Validates the utility function settings_utils#create_network_config
106 returns an acceptable NetworkConfig object
108 subnet_settings = list()
109 subnet_settings.append(SubnetConfig(name='sub1', cidr='10.0.0.0/24'))
110 subnet_settings.append(SubnetConfig(name='sub2', cidr='10.0.1.0/24'))
111 net_settings = NetworkConfig(
112 name=self.network_name, subnet_settings=subnet_settings)
113 self.net_creator = OpenStackNetwork(self.os_creds, net_settings)
114 network = self.net_creator.create()
116 derived_settings = settings_utils.create_network_config(
117 self.neutron, network)
119 self.assertIsNotNone(derived_settings)
120 self.assertEqual(net_settings.name, derived_settings.name)
121 self.assertEqual(net_settings.admin_state_up,
122 derived_settings.admin_state_up)
123 self.assertEqual(net_settings.external, derived_settings.external)
124 self.assertEqual(len(net_settings.subnet_settings),
125 len(derived_settings.subnet_settings))
127 # Validate the first subnet
128 orig_sub1 = net_settings.subnet_settings[0]
130 for derived_sub in derived_settings.subnet_settings:
131 if orig_sub1.name == derived_sub.name:
132 self.assertEqual(orig_sub1.cidr, derived_sub.cidr)
135 self.assertTrue(found)
137 # Validate the second subnet
138 orig_sub2 = net_settings.subnet_settings[1]
140 for derived_sub in derived_settings.subnet_settings:
141 if orig_sub2.name == derived_sub.name:
142 self.assertEqual(orig_sub2.cidr, derived_sub.cidr)
143 self.assertEqual(orig_sub2.ip_version, derived_sub.ip_version)
146 self.assertTrue(found)
149 class SettingsUtilsVmInstTests(OSComponentTestCase):
151 Tests the ability to reverse engineer VmInstanceConfig objects from
152 existing VMs/servers deployed to OpenStack
157 Instantiates the CreateImage object that is responsible for downloading
158 and creating an OS image file within OpenStack
160 self.nova = nova_utils.nova_client(
161 self.os_creds, self.os_session)
162 self.keystone = keystone_utils.keystone_client(
163 self.os_creds, self.os_session)
164 self.glance = glance_utils.glance_client(
165 self.os_creds, self.os_session)
166 self.neutron = neutron_utils.neutron_client(
167 self.os_creds, self.os_session)
169 guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
170 self.keypair_priv_filepath = 'tmp/' + guid
171 self.keypair_pub_filepath = self.keypair_priv_filepath + '.pub'
172 self.keypair_name = guid + '-kp'
173 self.vm_inst_name = guid + '-inst'
174 self.test_file_local_path = 'tmp/' + guid + '-hello.txt'
175 self.port_1_name = guid + '-port-1'
176 self.port_2_name = guid + '-port-2'
177 self.floating_ip_name = guid + 'fip1'
179 # Setup members to cleanup just in case they don't get created
180 self.inst_creator = None
181 self.keypair_creator = None
182 self.sec_grp_creator = None
183 self.flavor_creator = None
184 self.router_creator = None
185 self.network_creator = None
186 self.image_creator = None
190 os_image_settings = openstack_tests.cirros_image_settings(
191 name=guid + '-' + '-image',
192 image_metadata=self.image_metadata)
193 self.image_creator = create_image.OpenStackImage(self.os_creds,
195 self.image_creator.create()
197 # First network is public
198 self.pub_net_config = openstack_tests.get_pub_net_config(
199 net_name=guid + '-pub-net', subnet_name=guid + '-pub-subnet',
200 router_name=guid + '-pub-router',
201 external_net=self.ext_net_name)
203 self.network_creator = create_network.OpenStackNetwork(
204 self.os_creds, self.pub_net_config.network_settings)
205 self.network_creator.create()
208 self.router_creator = create_router.OpenStackRouter(
209 self.os_creds, self.pub_net_config.router_settings)
210 self.router_creator.create()
213 self.flavor_creator = create_flavor.OpenStackFlavor(
216 name=guid + '-flavor-name', ram=256, disk=1, vcpus=1))
217 self.flavor_creator.create()
220 self.keypair_creator = create_keypairs.OpenStackKeypair(
221 self.os_creds, KeypairConfig(
222 name=self.keypair_name,
223 public_filepath=self.keypair_pub_filepath,
224 private_filepath=self.keypair_priv_filepath))
225 self.keypair_creator.create()
227 # Create Security Group
228 sec_grp_name = guid + '-sec-grp'
229 rule1 = SecurityGroupRuleConfig(
230 sec_grp_name=sec_grp_name, direction=Direction.ingress,
231 protocol=Protocol.icmp)
232 rule2 = SecurityGroupRuleConfig(
233 sec_grp_name=sec_grp_name, direction=Direction.ingress,
234 protocol=Protocol.tcp, port_range_min=22, port_range_max=22)
235 self.sec_grp_creator = OpenStackSecurityGroup(
238 name=sec_grp_name, rule_settings=[rule1, rule2]))
239 self.sec_grp_creator.create()
242 ports_settings = list()
243 ports_settings.append(
245 name=self.port_1_name,
246 network_name=self.pub_net_config.network_settings.name))
248 instance_settings = VmInstanceConfig(
249 name=self.vm_inst_name,
250 flavor=self.flavor_creator.flavor_settings.name,
251 port_settings=ports_settings,
252 floating_ip_settings=[FloatingIpConfig(
253 name=self.floating_ip_name, port_name=self.port_1_name,
254 router_name=self.pub_net_config.router_settings.name)])
256 self.inst_creator = create_instance.OpenStackVmInstance(
257 self.os_creds, instance_settings,
258 self.image_creator.image_settings,
259 keypair_settings=self.keypair_creator.keypair_settings)
266 Cleans the created objects
268 if self.inst_creator:
270 self.inst_creator.clean()
274 if self.sec_grp_creator:
276 self.sec_grp_creator.clean()
280 if self.keypair_creator:
282 self.keypair_creator.clean()
286 if self.flavor_creator:
288 self.flavor_creator.clean()
292 if os.path.isfile(self.keypair_pub_filepath):
294 os.remove(self.keypair_pub_filepath)
298 if os.path.isfile(self.keypair_priv_filepath):
300 os.remove(self.keypair_priv_filepath)
304 if self.router_creator:
306 self.router_creator.clean()
310 if self.network_creator:
312 self.network_creator.clean()
316 if self.image_creator and not self.image_creator.image_settings.exists:
318 self.image_creator.clean()
322 if os.path.isfile(self.test_file_local_path):
323 os.remove(self.test_file_local_path)
325 super(self.__class__, self).__clean__()
327 def test_derive_vm_inst_config(self):
329 Validates the utility function settings_utils#create_vm_inst_config
330 returns an acceptable VmInstanceConfig object
332 self.inst_creator.create(block=True)
334 server = nova_utils.get_server(
335 self.nova, self.neutron, self.keystone,
336 vm_inst_settings=self.inst_creator.instance_settings)
337 derived_vm_settings = settings_utils.create_vm_inst_config(
338 self.nova, self.keystone, self.neutron, server,
339 self.os_creds.project_name)
340 self.assertIsNotNone(derived_vm_settings)
341 self.assertIsNotNone(derived_vm_settings.port_settings)
342 self.assertIsNotNone(derived_vm_settings.floating_ip_settings)
344 def test_derive_image_settings(self):
346 Validates the utility function settings_utils#create_image_settings
347 returns an acceptable ImageConfig object
349 self.inst_creator.create(block=True)
351 server = nova_utils.get_server(
352 self.nova, self.neutron, self.keystone,
353 vm_inst_settings=self.inst_creator.instance_settings)
354 derived_image_settings = settings_utils.determine_image_config(
355 self.glance, server, [self.image_creator.image_settings])
356 self.assertIsNotNone(derived_image_settings)
357 self.assertEqual(self.image_creator.image_settings.name,
358 derived_image_settings.name)
361 class SettingsUtilsUnitTests(unittest.TestCase):
363 Exercises the settings_utils.py functions around volumes
366 def test_vol_settings_from_vol(self):
368 name='vol-name', volume_id='vol-id', project_id='proj-id',
369 description='desc', size=99, vol_type='vol-type',
370 availability_zone='zone1', multi_attach=True)
371 settings = settings_utils.create_volume_config(volume)
372 self.assertEqual(volume.name, settings.name)
373 self.assertEqual(volume.description, settings.description)
374 self.assertEqual(volume.size, settings.size)
375 self.assertEqual(volume.type, settings.type_name)
376 self.assertEqual(volume.availability_zone, settings.availability_zone)
377 self.assertEqual(volume.multi_attach, settings.multi_attach)
379 def test_vol_type_settings_from_vol(self):
380 encryption = VolumeTypeEncryption(
381 volume_encryption_id='vol-encrypt-id', volume_type_id='vol-typ-id',
382 control_location='front-end', provider='FooClass', cipher='1',
384 qos_spec = QoSSpec(name='qos-spec-name', spec_id='qos-spec-id',
385 consumer=Consumer.back_end)
386 volume_type = VolumeType(
387 name='vol-type-name', volume_type_id='vol-type-id', public=True,
388 encryption=encryption, qos_spec=qos_spec)
390 settings = settings_utils.create_volume_type_config(volume_type)
391 self.assertEqual(volume_type.name, settings.name)
392 self.assertEqual(volume_type.public, settings.public)
394 encrypt_settings = settings.encryption
395 self.assertIsNotNone(encrypt_settings)
396 self.assertEqual(encryption.control_location,
397 encrypt_settings.control_location.value)
398 self.assertEqual(encryption.cipher, encrypt_settings.cipher)
399 self.assertEqual(encryption.key_size, encrypt_settings.key_size)
401 self.assertEqual(qos_spec.name, settings.qos_spec_name)
403 def test_flavor_settings_from_flavor(self):
405 name='flavor-name', flavor_id='flavor-id', ram=99, disk=101,
406 vcpus=9, ephemeral=3, swap=5, rxtx_factor=7, is_public=False)
407 settings = settings_utils.create_flavor_config(flavor)
408 self.assertEqual(flavor.name, settings.name)
409 self.assertEqual(flavor.id, settings.flavor_id)
410 self.assertEqual(flavor.ram, settings.ram)
411 self.assertEqual(flavor.disk, settings.disk)
412 self.assertEqual(flavor.vcpus, settings.vcpus)
413 self.assertEqual(flavor.ephemeral, settings.ephemeral)
414 self.assertEqual(flavor.swap, settings.swap)
415 self.assertEqual(flavor.rxtx_factor, settings.rxtx_factor)
416 self.assertEqual(flavor.is_public, settings.is_public)