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.
23 from snaps.openstack import create_stack
24 from snaps.openstack.create_flavor import OpenStackFlavor, FlavorSettings
26 from snaps.openstack.create_image import OpenStackImage
27 from snaps.openstack.create_instance import OpenStackVmInstance
28 from snaps.openstack.create_stack import StackSettings
29 from snaps.openstack.tests import openstack_tests
30 from snaps.openstack.tests.os_source_file_test import OSComponentTestCase
31 from snaps.openstack.utils import (
32 heat_utils, neutron_utils, nova_utils, settings_utils, glance_utils,
35 __author__ = 'spisarski'
37 logger = logging.getLogger('heat_utils_tests')
40 class HeatSmokeTests(OSComponentTestCase):
42 Tests to ensure that the heat client can communicate with the cloud
45 def test_heat_connect_success(self):
47 Tests to ensure that the proper credentials can connect.
49 heat = heat_utils.heat_client(self.os_creds)
51 # This should not throw an exception
52 stacks = heat.stacks.list()
56 def test_heat_connect_fail(self):
58 Tests to ensure that the improper credentials cannot connect.
60 from snaps.openstack.os_credentials import OSCreds
62 heat = heat_utils.heat_client(
63 OSCreds(username='user', password='pass',
64 auth_url=self.os_creds.auth_url,
65 project_name=self.os_creds.project_name,
66 proxy_settings=self.os_creds.proxy_settings))
67 stacks = heat.stacks.list()
69 # This should throw an exception
70 with self.assertRaises(Exception):
75 class HeatUtilsCreateSimpleStackTests(OSComponentTestCase):
77 Test basic Heat functionality
82 Instantiates OpenStack instances that cannot be spawned by Heat
84 guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
85 stack_name1 = guid + '-stack1'
86 stack_name2 = guid + '-stack2'
87 self.network_name = guid + '-net'
88 self.subnet_name = guid + '-subnet'
89 self.vm_inst_name = guid + '-inst'
91 self.image_creator = OpenStackImage(
92 self.os_creds, openstack_tests.cirros_image_settings(
93 name=guid + '-image', image_metadata=self.image_metadata))
94 self.image_creator.create()
97 self.flavor_creator = OpenStackFlavor(
99 FlavorSettings(name=guid + '-flavor', ram=256, disk=10, vcpus=1))
100 self.flavor_creator.create()
102 env_values = {'image_name': self.image_creator.image_settings.name,
103 'flavor_name': self.flavor_creator.flavor_settings.name,
104 'net_name': self.network_name,
105 'subnet_name': self.subnet_name,
106 'inst_name': self.vm_inst_name}
107 heat_tmplt_path = pkg_resources.resource_filename(
108 'snaps.openstack.tests.heat', 'test_heat_template.yaml')
109 self.stack_settings1 = StackSettings(
110 name=stack_name1, template_path=heat_tmplt_path,
111 env_values=env_values)
112 self.stack_settings2 = StackSettings(
113 name=stack_name2, template_path=heat_tmplt_path,
114 env_values=env_values)
117 self.heat_client = heat_utils.heat_client(self.os_creds)
121 Cleans the stack and image
125 heat_utils.delete_stack(self.heat_client, self.stack1)
131 heat_utils.delete_stack(self.heat_client, self.stack2)
135 if self.image_creator:
137 self.image_creator.clean()
141 if self.flavor_creator:
143 self.flavor_creator.clean()
147 def test_create_stack(self):
149 Tests the creation of an OpenStack Heat stack1 that does not exist.
151 self.stack1 = heat_utils.create_stack(self.heat_client,
152 self.stack_settings1)
154 stack_query_1 = heat_utils.get_stack(
155 self.heat_client, stack_settings=self.stack_settings1)
156 self.assertEqual(self.stack1, stack_query_1)
158 stack_query_2 = heat_utils.get_stack(
159 self.heat_client, stack_name=self.stack_settings1.name)
160 self.assertEqual(self.stack1, stack_query_2)
162 stack_query_3 = heat_utils.get_stack_by_id(self.heat_client,
164 self.assertEqual(self.stack1, stack_query_3)
166 resources = heat_utils.get_resources(self.heat_client, self.stack1)
167 self.assertIsNotNone(resources)
168 self.assertEqual(4, len(resources))
170 outputs = heat_utils.get_outputs(self.heat_client, self.stack1)
171 self.assertIsNotNone(outputs)
172 self.assertEqual(0, len(outputs))
174 self.assertTrue(stack_active(self.heat_client, self.stack1))
176 neutron = neutron_utils.neutron_client(self.os_creds)
177 networks = heat_utils.get_stack_networks(
178 self.heat_client, neutron, self.stack1)
179 self.assertIsNotNone(networks)
180 self.assertEqual(1, len(networks))
181 self.assertEqual(self.network_name, networks[0].name)
183 subnets = neutron_utils.get_subnets_by_network(neutron, networks[0])
184 self.assertEqual(1, len(subnets))
185 self.assertEqual(self.subnet_name, subnets[0].name)
187 nova = nova_utils.nova_client(self.os_creds)
188 servers = heat_utils.get_stack_servers(
189 self.heat_client, nova, self.stack1)
190 self.assertIsNotNone(servers)
191 self.assertEqual(1, len(servers))
192 self.assertEqual(self.vm_inst_name, servers[0].name)
194 def test_create_stack_x2(self):
196 Tests the creation of an OpenStack keypair that does not exist.
198 self.stack1 = heat_utils.create_stack(self.heat_client,
199 self.stack_settings1)
201 stack1_query_1 = heat_utils.get_stack(
202 self.heat_client, stack_settings=self.stack_settings1)
203 self.assertEqual(self.stack1, stack1_query_1)
205 stack1_query_2 = heat_utils.get_stack(
206 self.heat_client, stack_name=self.stack_settings1.name)
207 self.assertEqual(self.stack1, stack1_query_2)
209 stack1_query_3 = heat_utils.get_stack_by_id(self.heat_client,
211 self.assertEqual(self.stack1, stack1_query_3)
213 self.assertTrue(stack_active(self.heat_client, self.stack1))
215 self.stack2 = heat_utils.create_stack(self.heat_client,
216 self.stack_settings2)
218 stack2_query_1 = heat_utils.get_stack(
219 self.heat_client, stack_settings=self.stack_settings2)
220 self.assertEqual(self.stack2, stack2_query_1)
222 stack2_query_2 = heat_utils.get_stack(
223 self.heat_client, stack_name=self.stack_settings2.name)
224 self.assertEqual(self.stack2, stack2_query_2)
226 stack2_query_3 = heat_utils.get_stack_by_id(self.heat_client,
228 self.assertEqual(self.stack2, stack2_query_3)
230 self.assertTrue(stack_active(self.heat_client, self.stack2))
233 class HeatUtilsCreateComplexStackTests(OSComponentTestCase):
235 Test basic Heat functionality
240 Instantiates OpenStack instances that cannot be spawned by Heat
242 guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
243 stack_name = guid + '-stack'
244 self.network_name = guid + '-net'
245 self.subnet_name = guid + '-subnet'
246 self.vm_inst1_name = guid + '-inst1'
247 self.vm_inst2_name = guid + '-inst2'
248 self.flavor1_name = guid + '-flavor1'
249 self.flavor2_name = guid + '-flavor2'
250 self.keypair_name = guid + '-keypair'
252 self.image_creator1 = OpenStackImage(
253 self.os_creds, openstack_tests.cirros_image_settings(
254 name=guid + '-image1', image_metadata=self.image_metadata))
255 self.image_creator1.create()
257 self.image_creator2 = OpenStackImage(
258 self.os_creds, openstack_tests.cirros_image_settings(
259 name=guid + '-image2', image_metadata=self.image_metadata))
260 self.image_creator2.create()
262 env_values = {'image1_name': self.image_creator1.image_settings.name,
263 'image2_name': self.image_creator2.image_settings.name,
264 'flavor1_name': self.flavor1_name,
265 'flavor2_name': self.flavor2_name,
266 'net_name': self.network_name,
267 'subnet_name': self.subnet_name,
268 'keypair_name': self.keypair_name,
269 'inst1_name': self.vm_inst1_name,
270 'inst2_name': self.vm_inst2_name,
271 'external_net_name': self.ext_net_name}
272 heat_tmplt_path = pkg_resources.resource_filename(
273 'snaps.openstack.tests.heat', 'floating_ip_heat_template.yaml')
274 stack_settings = StackSettings(
275 name=stack_name, template_path=heat_tmplt_path,
276 env_values=env_values)
277 self.heat_client = heat_utils.heat_client(self.os_creds)
278 self.stack = heat_utils.create_stack(self.heat_client, stack_settings)
280 self.assertTrue(stack_active(self.heat_client, self.stack))
282 self.keypair1_settings = None
283 self.keypair2_settings = None
287 Cleans the stack and image
291 heat_utils.delete_stack(self.heat_client, self.stack)
292 # Wait until stack deployment has completed
293 end_time = time.time() + create_stack.STACK_COMPLETE_TIMEOUT
295 while time.time() < end_time:
296 status = heat_utils.get_stack_status(self.heat_client,
298 if status == create_stack.STATUS_DELETE_COMPLETE:
301 elif status == create_stack.STATUS_DELETE_FAILED:
308 nova = nova_utils.nova_client(self.os_creds)
309 neutron = neutron_utils.neutron_client(self.os_creds)
310 glance = glance_utils.glance_client(self.os_creds)
311 servers = heat_utils.get_stack_servers(
312 self.heat_client, nova, self.stack)
313 for server in servers:
314 vm_settings = settings_utils.create_vm_inst_settings(
315 nova, neutron, server)
316 img_settings = settings_utils.determine_image_config(
318 [self.image_creator1.image_settings,
319 self.image_creator2.image_settings])
320 vm_creator = OpenStackVmInstance(
321 self.os_creds, vm_settings, img_settings)
322 vm_creator.initialize()
324 vm_creator.vm_deleted(block=True)
326 heat_utils.delete_stack(self.heat_client, self.stack)
331 if self.image_creator1:
333 self.image_creator1.clean()
337 if self.image_creator2:
339 self.image_creator2.clean()
343 if self.keypair1_settings:
344 expanded_path = os.path.expanduser(
345 self.keypair1_settings.private_filepath)
346 os.chmod(expanded_path, 0o755)
347 os.remove(expanded_path)
349 if self.keypair2_settings:
350 expanded_path = os.path.expanduser(
351 self.keypair2_settings.private_filepath)
352 os.chmod(expanded_path, 0o755)
353 os.remove(expanded_path)
355 def test_get_settings_from_stack(self):
357 Tests that a heat template with floating IPs and can have the proper
358 settings derived from settings_utils.py.
360 resources = heat_utils.get_resources(self.heat_client, self.stack)
361 self.assertIsNotNone(resources)
362 self.assertEqual(12, len(resources))
364 options = heat_utils.get_outputs(self.heat_client, self.stack)
365 self.assertIsNotNone(options)
366 self.assertEqual(1, len(options))
368 neutron = neutron_utils.neutron_client(self.os_creds)
369 networks = heat_utils.get_stack_networks(
370 self.heat_client, neutron, self.stack)
371 self.assertIsNotNone(networks)
372 self.assertEqual(1, len(networks))
373 self.assertEqual(self.network_name, networks[0].name)
375 network_settings = settings_utils.create_network_settings(
376 neutron, networks[0])
377 self.assertIsNotNone(network_settings)
378 self.assertEqual(self.network_name, network_settings.name)
380 nova = nova_utils.nova_client(self.os_creds)
381 glance = glance_utils.glance_client(self.os_creds)
383 servers = heat_utils.get_stack_servers(
384 self.heat_client, nova, self.stack)
385 self.assertIsNotNone(servers)
386 self.assertEqual(2, len(servers))
388 image_settings = settings_utils.determine_image_config(
390 [self.image_creator1.image_settings,
391 self.image_creator2.image_settings])
393 self.assertIsNotNone(image_settings)
394 if image_settings.name.endswith('1'):
396 self.image_creator1.image_settings.name, image_settings.name)
399 self.image_creator2.image_settings.name, image_settings.name)
401 image_settings = settings_utils.determine_image_config(
403 [self.image_creator1.image_settings,
404 self.image_creator2.image_settings])
405 if image_settings.name.endswith('1'):
407 self.image_creator1.image_settings.name, image_settings.name)
410 self.image_creator2.image_settings.name, image_settings.name)
412 self.keypair1_settings = settings_utils.determine_keypair_settings(
413 self.heat_client, self.stack, servers[0],
414 priv_key_key='private_key')
415 self.assertIsNotNone(self.keypair1_settings)
416 self.assertEqual(self.keypair_name, self.keypair1_settings.name)
418 self.keypair2_settings = settings_utils.determine_keypair_settings(
419 self.heat_client, self.stack, servers[1],
420 priv_key_key='private_key')
421 self.assertIsNotNone(self.keypair2_settings)
422 self.assertEqual(self.keypair_name, self.keypair2_settings.name)
425 class HeatUtilsRouterTests(OSComponentTestCase):
427 Test Heat volume functionality
432 Instantiates OpenStack instances that cannot be spawned by Heat
434 guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
435 stack_name = guid + '-stack'
437 self.net_name = guid + '-net'
438 self.subnet_name = guid + '-subnet'
439 self.router_name = guid + '-router'
442 'net_name': self.net_name,
443 'subnet_name': self.subnet_name,
444 'router_name': self.router_name,
445 'external_net_name': self.ext_net_name}
447 heat_tmplt_path = pkg_resources.resource_filename(
448 'snaps.openstack.tests.heat', 'router_heat_template.yaml')
449 self.stack_settings = StackSettings(
450 name=stack_name, template_path=heat_tmplt_path,
451 env_values=env_values)
453 self.heat_client = heat_utils.heat_client(self.os_creds)
454 self.neutron = neutron_utils.neutron_client(self.os_creds)
458 Cleans the image and downloaded image file
462 heat_utils.delete_stack(self.heat_client, self.stack)
466 def test_create_router_with_stack(self):
468 Tests the creation of an OpenStack router with Heat and the retrieval
469 of the Router Domain objects from heat_utils#get_stack_routers().
471 self.stack = heat_utils.create_stack(
472 self.heat_client, self.stack_settings)
474 # Wait until stack deployment has completed
475 end_time = time.time() + create_stack.STACK_COMPLETE_TIMEOUT
477 while time.time() < end_time:
478 status = heat_utils.get_stack_status(self.heat_client,
480 if status == create_stack.STATUS_CREATE_COMPLETE:
483 elif status == create_stack.STATUS_CREATE_FAILED:
489 self.assertTrue(is_active)
491 routers = heat_utils.get_stack_routers(
492 self.heat_client, self.neutron, self.stack)
494 self.assertEqual(1, len(routers))
497 self.assertEqual(self.router_name, router.name)
499 ext_net = neutron_utils.get_network(
500 self.neutron, network_name=self.ext_net_name)
501 self.assertEqual(ext_net.id, router.external_network_id)
504 class HeatUtilsVolumeTests(OSComponentTestCase):
506 Test Heat volume functionality
511 Instantiates OpenStack instances that cannot be spawned by Heat
513 guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
514 stack_name = guid + '-stack'
515 self.volume_name = guid + '-vol'
516 self.volume_type_name = guid + '-vol-type'
519 'volume_name': self.volume_name,
520 'volume_type_name': self.volume_type_name}
522 heat_tmplt_path = pkg_resources.resource_filename(
523 'snaps.openstack.tests.heat', 'volume_heat_template.yaml')
524 self.stack_settings = StackSettings(
525 name=stack_name, template_path=heat_tmplt_path,
526 env_values=env_values)
528 self.heat_client = heat_utils.heat_client(self.os_creds)
529 self.cinder = cinder_utils.cinder_client(self.os_creds)
537 heat_utils.delete_stack(self.heat_client, self.stack)
541 def test_create_vol_with_stack(self):
543 Tests the creation of an OpenStack volume with Heat.
545 self.stack = heat_utils.create_stack(
546 self.heat_client, self.stack_settings)
547 self.assertTrue(stack_active(self.heat_client, self.stack))
549 volumes = heat_utils.get_stack_volumes(
550 self.heat_client, self.cinder, self.stack)
552 self.assertEqual(1, len(volumes))
555 self.assertEqual(self.volume_name, volume.name)
556 self.assertEqual(self.volume_type_name, volume.type)
557 self.assertEqual(1, volume.size)
558 self.assertEqual(False, volume.multi_attach)
560 def test_create_vol_types_with_stack(self):
562 Tests the creation of an OpenStack volume with Heat.
564 self.stack = heat_utils.create_stack(
565 self.heat_client, self.stack_settings)
566 self.assertTrue(stack_active(self.heat_client, self.stack))
568 volume_types = heat_utils.get_stack_volume_types(
569 self.heat_client, self.cinder, self.stack)
571 self.assertEqual(1, len(volume_types))
573 volume_type = volume_types[0]
575 self.assertEqual(self.volume_type_name, volume_type.name)
576 self.assertTrue(volume_type.public)
577 self.assertIsNone(volume_type.qos_spec)
579 encryption = volume_type.encryption
580 self.assertIsNotNone(encryption)
581 self.assertIsNone(encryption.cipher)
582 self.assertEqual('front-end', encryption.control_location)
583 self.assertIsNone(encryption.key_size)
584 self.assertEqual(u'nova.volume.encryptors.luks.LuksEncryptor',
586 self.assertEqual(volume_type.id, encryption.volume_type_id)
589 class HeatUtilsFlavorTests(OSComponentTestCase):
591 Test Heat volume functionality
596 Instantiates OpenStack instances that cannot be spawned by Heat
598 guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
599 self.name_prefix = guid
600 stack_name = guid + '-stack'
602 heat_tmplt_path = pkg_resources.resource_filename(
603 'snaps.openstack.tests.heat', 'flavor_heat_template.yaml')
604 self.stack_settings = StackSettings(
605 name=stack_name, template_path=heat_tmplt_path)
607 self.heat_client = heat_utils.heat_client(self.os_creds)
608 self.nova = nova_utils.nova_client(self.os_creds)
616 heat_utils.delete_stack(self.heat_client, self.stack)
620 def test_create_flavor_with_stack(self):
622 Tests the creation of an OpenStack volume with Heat.
624 self.stack = heat_utils.create_stack(
625 self.heat_client, self.stack_settings)
627 self.assertTrue(stack_active(self.heat_client, self.stack))
629 flavors = heat_utils.get_stack_flavors(
630 self.heat_client, self.nova, self.stack)
632 self.assertEqual(1, len(flavors))
635 self.assertTrue(flavor.name.startswith(self.name_prefix))
636 self.assertEqual(1024, flavor.ram)
637 self.assertEqual(200, flavor.disk)
638 self.assertEqual(8, flavor.vcpus)
639 self.assertEqual(0, flavor.ephemeral)
640 self.assertIsNone(flavor.swap)
641 self.assertEqual(1.0, flavor.rxtx_factor)
642 self.assertTrue(flavor.is_public)
645 class HeatUtilsKeypairTests(OSComponentTestCase):
647 Test Heat volume functionality
652 Instantiates OpenStack instances that cannot be spawned by Heat
654 guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
655 stack_name = guid + '-stack'
656 self.keypair_name = guid + '-kp'
658 env_values = {'keypair_name': self.keypair_name}
660 heat_tmplt_path = pkg_resources.resource_filename(
661 'snaps.openstack.tests.heat', 'keypair_heat_template.yaml')
662 self.stack_settings = StackSettings(
663 name=stack_name, template_path=heat_tmplt_path,
664 env_values=env_values)
666 self.heat_client = heat_utils.heat_client(self.os_creds)
667 self.nova = nova_utils.nova_client(self.os_creds)
675 heat_utils.delete_stack(self.heat_client, self.stack)
679 def test_create_keypair_with_stack(self):
681 Tests the creation of an OpenStack keypair with Heat.
683 self.stack = heat_utils.create_stack(
684 self.heat_client, self.stack_settings)
685 self.assertTrue(stack_active(self.heat_client, self.stack))
687 keypairs = heat_utils.get_stack_keypairs(
688 self.heat_client, self.nova, self.stack)
690 self.assertEqual(1, len(keypairs))
691 keypair = keypairs[0]
693 self.assertEqual(self.keypair_name, keypair.name)
695 outputs = heat_utils.get_outputs(self.heat_client, self.stack)
697 for output in outputs:
698 if output.key == 'private_key':
699 self.assertTrue(output.value.startswith(
700 '-----BEGIN RSA PRIVATE KEY-----'))
702 keypair = nova_utils.get_keypair_by_id(self.nova, keypair.id)
703 self.assertIsNotNone(keypair)
705 self.assertEqual(self.keypair_name, keypair.name)
708 class HeatUtilsSecurityGroupTests(OSComponentTestCase):
710 Test Heat volume functionality
715 Instantiates OpenStack instances that cannot be spawned by Heat
717 guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
718 stack_name = guid + '-stack'
719 self.sec_grp_name = guid + '-sec-grp'
721 env_values = {'security_group_name': self.sec_grp_name}
723 heat_tmplt_path = pkg_resources.resource_filename(
724 'snaps.openstack.tests.heat', 'security_group_heat_template.yaml')
725 self.stack_settings = StackSettings(
726 name=stack_name, template_path=heat_tmplt_path,
727 env_values=env_values)
729 self.heat_client = heat_utils.heat_client(self.os_creds)
730 self.neutron = neutron_utils.neutron_client(self.os_creds)
738 heat_utils.delete_stack(self.heat_client, self.stack)
742 def test_create_security_group_with_stack(self):
744 Tests the creation of an OpenStack SecurityGroup with Heat.
746 self.stack = heat_utils.create_stack(
747 self.heat_client, self.stack_settings)
748 self.assertTrue(stack_active(self.heat_client, self.stack))
750 sec_grp = heat_utils.get_stack_security_groups(
751 self.heat_client, self.neutron, self.stack)[0]
753 self.assertEqual(self.sec_grp_name, sec_grp.name)
754 self.assertEqual('Test description', sec_grp.description)
755 self.assertEqual(2, len(sec_grp.rules))
758 has_icmp_rule = False
760 for rule in sec_grp.rules:
761 if (rule.security_group_id == sec_grp.id
762 and rule.direction == 'egress'
763 and rule.ethertype == 'IPv4'
764 and rule.port_range_min == 22
765 and rule.port_range_max == 22
766 and rule.protocol == 'tcp'
767 and rule.remote_group_id is None
768 and rule.remote_ip_prefix == '0.0.0.0/0'):
770 if (rule.security_group_id == sec_grp.id
771 and rule.direction == 'ingress'
772 and rule.ethertype == 'IPv4'
773 and rule.port_range_min is None
774 and rule.port_range_max is None
775 and rule.protocol == 'icmp'
776 and rule.remote_group_id is None
777 and rule.remote_ip_prefix == '0.0.0.0/0'):
780 self.assertTrue(has_ssh_rule)
781 self.assertTrue(has_icmp_rule)
784 def stack_active(heat_cli, stack):
786 Blocks until stack application has successfully completed or failed
787 :param heat_cli: the Heat client
788 :param stack: the Stack domain object
791 # Wait until stack deployment has completed
792 end_time = time.time() + create_stack.STACK_COMPLETE_TIMEOUT
794 while time.time() < end_time:
795 status = heat_utils.get_stack_status(heat_cli, stack.id)
796 if status == create_stack.STATUS_CREATE_COMPLETE:
799 elif status == create_stack.STATUS_CREATE_FAILED: