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.openstack import create_stack
22 from snaps.openstack.create_flavor import OpenStackFlavor, FlavorSettings
24 from snaps.openstack.create_image import OpenStackImage
25 from snaps.openstack.create_instance import OpenStackVmInstance
26 from snaps.openstack.create_stack import StackSettings
27 from snaps.openstack.tests import openstack_tests
28 from snaps.openstack.tests.os_source_file_test import OSComponentTestCase
29 from snaps.openstack.utils import (
30 heat_utils, neutron_utils, nova_utils, settings_utils, glance_utils)
32 __author__ = 'spisarski'
34 logger = logging.getLogger('heat_utils_tests')
37 class HeatSmokeTests(OSComponentTestCase):
39 Tests to ensure that the heat client can communicate with the cloud
42 def test_heat_connect_success(self):
44 Tests to ensure that the proper credentials can connect.
46 heat = heat_utils.heat_client(self.os_creds)
48 # This should not throw an exception
49 stacks = heat.stacks.list()
53 def test_heat_connect_fail(self):
55 Tests to ensure that the improper credentials cannot connect.
57 from snaps.openstack.os_credentials import OSCreds
59 heat = heat_utils.heat_client(
60 OSCreds(username='user', password='pass',
61 auth_url=self.os_creds.auth_url,
62 project_name=self.os_creds.project_name,
63 proxy_settings=self.os_creds.proxy_settings))
64 stacks = heat.stacks.list()
66 # This should throw an exception
67 with self.assertRaises(Exception):
72 class HeatUtilsCreateSimpleStackTests(OSComponentTestCase):
74 Test basic Heat functionality
79 Instantiates OpenStack instances that cannot be spawned by Heat
81 guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
82 stack_name1 = guid + '-stack1'
83 stack_name2 = guid + '-stack2'
84 self.network_name = guid + '-net'
85 self.subnet_name = guid + '-subnet'
86 self.vm_inst_name = guid + '-inst'
88 self.image_creator = OpenStackImage(
89 self.os_creds, openstack_tests.cirros_image_settings(
90 name=guid + '-image', image_metadata=self.image_metadata))
91 self.image_creator.create()
94 self.flavor_creator = OpenStackFlavor(
96 FlavorSettings(name=guid + '-flavor', ram=256, disk=10, vcpus=1))
97 self.flavor_creator.create()
99 env_values = {'image_name': self.image_creator.image_settings.name,
100 'flavor_name': self.flavor_creator.flavor_settings.name,
101 'net_name': self.network_name,
102 'subnet_name': self.subnet_name,
103 'inst_name': self.vm_inst_name}
104 heat_tmplt_path = pkg_resources.resource_filename(
105 'snaps.openstack.tests.heat', 'test_heat_template.yaml')
106 self.stack_settings1 = StackSettings(
107 name=stack_name1, template_path=heat_tmplt_path,
108 env_values=env_values)
109 self.stack_settings2 = StackSettings(
110 name=stack_name2, template_path=heat_tmplt_path,
111 env_values=env_values)
114 self.heat_client = heat_utils.heat_client(self.os_creds)
118 Cleans the image and downloaded image file
122 heat_utils.delete_stack(self.heat_client, self.stack1)
128 heat_utils.delete_stack(self.heat_client, self.stack2)
132 if self.image_creator:
134 self.image_creator.clean()
138 if self.flavor_creator:
140 self.flavor_creator.clean()
144 def test_create_stack(self):
146 Tests the creation of an OpenStack Heat stack1 that does not exist.
148 self.stack1 = heat_utils.create_stack(self.heat_client,
149 self.stack_settings1)
151 stack_query_1 = heat_utils.get_stack(
152 self.heat_client, stack_settings=self.stack_settings1)
153 self.assertEqual(self.stack1, stack_query_1)
155 stack_query_2 = heat_utils.get_stack(
156 self.heat_client, stack_name=self.stack_settings1.name)
157 self.assertEqual(self.stack1, stack_query_2)
159 stack_query_3 = heat_utils.get_stack_by_id(self.heat_client,
161 self.assertEqual(self.stack1, stack_query_3)
163 resources = heat_utils.get_resources(self.heat_client, self.stack1)
164 self.assertIsNotNone(resources)
165 self.assertEqual(4, len(resources))
167 outputs = heat_utils.get_outputs(self.heat_client, self.stack1)
168 self.assertIsNotNone(outputs)
169 self.assertEqual(0, len(outputs))
171 # Wait until stack deployment has completed
172 end_time = time.time() + create_stack.STACK_COMPLETE_TIMEOUT
174 while time.time() < end_time:
175 status = heat_utils.get_stack_status(self.heat_client,
177 if status == create_stack.STATUS_CREATE_COMPLETE:
180 elif status == create_stack.STATUS_CREATE_FAILED:
186 self.assertTrue(is_active)
188 neutron = neutron_utils.neutron_client(self.os_creds)
189 networks = heat_utils.get_stack_networks(
190 self.heat_client, neutron, self.stack1)
191 self.assertIsNotNone(networks)
192 self.assertEqual(1, len(networks))
193 self.assertEqual(self.network_name, networks[0].name)
195 subnets = neutron_utils.get_subnets_by_network(neutron, networks[0])
196 self.assertEqual(1, len(subnets))
197 self.assertEqual(self.subnet_name, subnets[0].name)
199 nova = nova_utils.nova_client(self.os_creds)
200 servers = heat_utils.get_stack_servers(
201 self.heat_client, nova, self.stack1)
202 self.assertIsNotNone(servers)
203 self.assertEqual(1, len(servers))
204 self.assertEqual(self.vm_inst_name, servers[0].name)
206 def test_create_stack_x2(self):
208 Tests the creation of an OpenStack keypair that does not exist.
210 self.stack1 = heat_utils.create_stack(self.heat_client,
211 self.stack_settings1)
213 stack1_query_1 = heat_utils.get_stack(
214 self.heat_client, stack_settings=self.stack_settings1)
215 self.assertEqual(self.stack1, stack1_query_1)
217 stack1_query_2 = heat_utils.get_stack(
218 self.heat_client, stack_name=self.stack_settings1.name)
219 self.assertEqual(self.stack1, stack1_query_2)
221 stack1_query_3 = heat_utils.get_stack_by_id(self.heat_client,
223 self.assertEqual(self.stack1, stack1_query_3)
225 end_time = time.time() + create_stack.STACK_COMPLETE_TIMEOUT
227 while time.time() < end_time:
228 status = heat_utils.get_stack_status(self.heat_client,
230 if status == create_stack.STATUS_CREATE_COMPLETE:
233 elif status == create_stack.STATUS_CREATE_FAILED:
239 self.assertTrue(is_active)
241 self.stack2 = heat_utils.create_stack(self.heat_client,
242 self.stack_settings2)
244 stack2_query_1 = heat_utils.get_stack(
245 self.heat_client, stack_settings=self.stack_settings2)
246 self.assertEqual(self.stack2, stack2_query_1)
248 stack2_query_2 = heat_utils.get_stack(
249 self.heat_client, stack_name=self.stack_settings2.name)
250 self.assertEqual(self.stack2, stack2_query_2)
252 stack2_query_3 = heat_utils.get_stack_by_id(self.heat_client,
254 self.assertEqual(self.stack2, stack2_query_3)
256 end_time = time.time() + create_stack.STACK_COMPLETE_TIMEOUT
259 while time.time() < end_time:
260 status = heat_utils.get_stack_status(self.heat_client,
262 if status == create_stack.STATUS_CREATE_COMPLETE:
265 elif status == create_stack.STATUS_CREATE_FAILED:
271 self.assertTrue(is_active)
274 class HeatUtilsCreateComplexStackTests(OSComponentTestCase):
276 Test basic Heat functionality
281 Instantiates OpenStack instances that cannot be spawned by Heat
283 guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
284 stack_name = guid + '-stack'
285 self.network_name = guid + '-net'
286 self.subnet_name = guid + '-subnet'
287 self.vm_inst1_name = guid + '-inst1'
288 self.vm_inst2_name = guid + '-inst2'
289 self.flavor1_name = guid + '-flavor1'
290 self.flavor2_name = guid + '-flavor2'
291 self.keypair_name = guid + '-keypair'
293 self.image_creator1 = OpenStackImage(
294 self.os_creds, openstack_tests.cirros_image_settings(
295 name=guid + '-image1', image_metadata=self.image_metadata))
296 self.image_creator1.create()
298 self.image_creator2 = OpenStackImage(
299 self.os_creds, openstack_tests.cirros_image_settings(
300 name=guid + '-image2', image_metadata=self.image_metadata))
301 self.image_creator2.create()
303 env_values = {'image1_name': self.image_creator1.image_settings.name,
304 'image2_name': self.image_creator2.image_settings.name,
305 'flavor1_name': self.flavor1_name,
306 'flavor2_name': self.flavor2_name,
307 'net_name': self.network_name,
308 'subnet_name': self.subnet_name,
309 'keypair_name': self.keypair_name,
310 'inst1_name': self.vm_inst1_name,
311 'inst2_name': self.vm_inst2_name,
312 'external_net_name': self.ext_net_name}
313 heat_tmplt_path = pkg_resources.resource_filename(
314 'snaps.openstack.tests.heat', 'floating_ip_heat_template.yaml')
315 stack_settings = StackSettings(
316 name=stack_name, template_path=heat_tmplt_path,
317 env_values=env_values)
318 self.heat_client = heat_utils.heat_client(self.os_creds)
319 self.stack = heat_utils.create_stack(self.heat_client, stack_settings)
321 # Wait until stack deployment has completed
322 end_time = time.time() + create_stack.STACK_COMPLETE_TIMEOUT
324 while time.time() < end_time:
325 status = heat_utils.get_stack_status(self.heat_client,
327 if status == create_stack.STATUS_CREATE_COMPLETE:
330 elif status == create_stack.STATUS_CREATE_FAILED:
335 self.assertTrue(is_active)
339 Cleans the image and downloaded image file
343 heat_utils.delete_stack(self.heat_client, self.stack)
344 # Wait until stack deployment has completed
345 end_time = time.time() + create_stack.STACK_COMPLETE_TIMEOUT
347 while time.time() < end_time:
348 status = heat_utils.get_stack_status(self.heat_client,
350 if status == create_stack.STATUS_DELETE_COMPLETE:
353 elif status == create_stack.STATUS_DELETE_FAILED:
360 nova = nova_utils.nova_client(self.os_creds)
361 neutron = neutron_utils.neutron_client(self.os_creds)
362 glance = glance_utils.glance_client(self.os_creds)
363 servers = heat_utils.get_stack_servers(
364 self.heat_client, nova, self.stack)
365 for server in servers:
366 vm_settings = settings_utils.create_vm_inst_settings(
367 nova, neutron, server)
368 img_settings = settings_utils.determine_image_settings(
370 [self.image_creator1.image_settings,
371 self.image_creator2.image_settings])
372 vm_creator = OpenStackVmInstance(
373 self.os_creds, vm_settings, img_settings)
374 vm_creator.initialize()
376 vm_creator.vm_deleted(block=True)
378 heat_utils.delete_stack(self.heat_client, self.stack)
383 if self.image_creator1:
385 self.image_creator1.clean()
389 if self.image_creator2:
391 self.image_creator2.clean()
395 def test_get_settings_from_stack(self):
397 Tests that a heat template with floating IPs and can have the proper
398 settings derived from settings_utils.py.
400 resources = heat_utils.get_resources(self.heat_client, self.stack)
401 self.assertIsNotNone(resources)
402 self.assertEqual(12, len(resources))
404 options = heat_utils.get_outputs(self.heat_client, self.stack)
405 self.assertIsNotNone(options)
406 self.assertEqual(1, len(options))
408 neutron = neutron_utils.neutron_client(self.os_creds)
409 networks = heat_utils.get_stack_networks(
410 self.heat_client, neutron, self.stack)
411 self.assertIsNotNone(networks)
412 self.assertEqual(1, len(networks))
413 self.assertEqual(self.network_name, networks[0].name)
415 network_settings = settings_utils.create_network_settings(
416 neutron, networks[0])
417 self.assertIsNotNone(network_settings)
418 self.assertEqual(self.network_name, network_settings.name)
420 nova = nova_utils.nova_client(self.os_creds)
421 glance = glance_utils.glance_client(self.os_creds)
423 servers = heat_utils.get_stack_servers(
424 self.heat_client, nova, self.stack)
425 self.assertIsNotNone(servers)
426 self.assertEqual(2, len(servers))
428 image_settings = settings_utils.determine_image_settings(
430 [self.image_creator1.image_settings,
431 self.image_creator2.image_settings])
433 self.assertIsNotNone(image_settings)
434 if image_settings.name.endswith('1'):
436 self.image_creator1.image_settings.name, image_settings.name)
439 self.image_creator2.image_settings.name, image_settings.name)
441 image_settings = settings_utils.determine_image_settings(
443 [self.image_creator1.image_settings,
444 self.image_creator2.image_settings])
445 if image_settings.name.endswith('1'):
447 self.image_creator1.image_settings.name, image_settings.name)
450 self.image_creator2.image_settings.name, image_settings.name)
452 keypair1_settings = settings_utils.determine_keypair_settings(
453 self.heat_client, self.stack, servers[0],
454 priv_key_key='private_key')
455 self.assertIsNotNone(keypair1_settings)
456 self.assertEqual(self.keypair_name, keypair1_settings.name)
458 keypair2_settings = settings_utils.determine_keypair_settings(
459 self.heat_client, self.stack, servers[1],
460 priv_key_key='private_key')
461 self.assertIsNotNone(keypair2_settings)
462 self.assertEqual(self.keypair_name, keypair2_settings.name)