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_stack import StackSettings
26 from snaps.openstack.tests import openstack_tests
27 from snaps.openstack.tests.os_source_file_test import OSComponentTestCase
28 from snaps.openstack.utils import heat_utils, neutron_utils
30 __author__ = 'spisarski'
32 logger = logging.getLogger('nova_utils_tests')
35 class HeatSmokeTests(OSComponentTestCase):
37 Tests to ensure that the heat client can communicate with the cloud
40 def test_nova_connect_success(self):
42 Tests to ensure that the proper credentials can connect.
44 heat = heat_utils.heat_client(self.os_creds)
46 # This should not throw an exception
47 stacks = heat.stacks.list()
51 def test_nova_connect_fail(self):
53 Tests to ensure that the improper credentials cannot connect.
55 from snaps.openstack.os_credentials import OSCreds
57 heat = heat_utils.heat_client(
58 OSCreds(username='user', password='pass',
59 auth_url=self.os_creds.auth_url,
60 project_name=self.os_creds.project_name,
61 proxy_settings=self.os_creds.proxy_settings))
62 stacks = heat.stacks.list()
64 # This should throw an exception
65 with self.assertRaises(Exception):
70 class HeatUtilsCreateStackTests(OSComponentTestCase):
72 Test basic Heat functionality
77 Instantiates OpenStack instances that cannot be spawned by Heat
79 guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
80 stack_name1 = guid + '-stack1'
81 stack_name2 = guid + '-stack2'
82 self.network_name = guid + '-net'
83 self.subnet_name = guid + '-subnet'
85 self.image_creator = OpenStackImage(
86 self.os_creds, openstack_tests.cirros_image_settings(
87 name=guid + '-image', image_metadata=self.image_metadata))
88 self.image_creator.create()
91 self.flavor_creator = OpenStackFlavor(
93 FlavorSettings(name=guid + '-flavor', ram=256, disk=10, vcpus=1))
94 self.flavor_creator.create()
96 env_values = {'image_name': self.image_creator.image_settings.name,
97 'flavor_name': self.flavor_creator.flavor_settings.name,
98 'net_name': self.network_name,
99 'subnet_name': self.subnet_name}
100 heat_tmplt_path = pkg_resources.resource_filename(
101 'snaps.openstack.tests.heat', 'test_heat_template.yaml')
102 self.stack_settings1 = StackSettings(
103 name=stack_name1, template_path=heat_tmplt_path,
104 env_values=env_values)
105 self.stack_settings2 = StackSettings(
106 name=stack_name2, template_path=heat_tmplt_path,
107 env_values=env_values)
110 self.heat_client = heat_utils.heat_client(self.os_creds)
114 Cleans the image and downloaded image file
118 heat_utils.delete_stack(self.heat_client, self.stack1)
124 heat_utils.delete_stack(self.heat_client, self.stack2)
128 if self.image_creator:
130 self.image_creator.clean()
134 if self.flavor_creator:
136 self.flavor_creator.clean()
140 def test_create_stack(self):
142 Tests the creation of an OpenStack Heat stack1 that does not exist.
144 self.stack1 = heat_utils.create_stack(self.heat_client,
145 self.stack_settings1)
147 stack_query_1 = heat_utils.get_stack(
148 self.heat_client, stack_settings=self.stack_settings1)
149 self.assertEqual(self.stack1, stack_query_1)
151 stack_query_2 = heat_utils.get_stack(
152 self.heat_client, stack_name=self.stack_settings1.name)
153 self.assertEqual(self.stack1, stack_query_2)
155 stack_query_3 = heat_utils.get_stack_by_id(self.heat_client,
157 self.assertEqual(self.stack1, stack_query_3)
159 outputs = heat_utils.get_stack_outputs(
160 self.heat_client, self.stack1.id)
161 self.assertIsNotNone(outputs)
162 self.assertEqual(0, len(outputs))
164 end_time = time.time() + create_stack.STACK_COMPLETE_TIMEOUT
167 while time.time() < end_time:
168 status = heat_utils.get_stack_status(self.heat_client,
170 if status == create_stack.STATUS_CREATE_COMPLETE:
173 elif status == create_stack.STATUS_CREATE_FAILED:
179 self.assertTrue(is_active)
181 resources = heat_utils.get_resources(self.heat_client, self.stack1)
182 self.assertIsNotNone(resources)
183 self.assertEqual(4, len(resources))
185 neutron = neutron_utils.neutron_client(self.os_creds)
186 networks = heat_utils.get_stack_networks(
187 self.heat_client, neutron, self.stack1)
188 self.assertIsNotNone(networks)
189 self.assertEqual(1, len(networks))
190 self.assertEqual(self.network_name, networks[0].name)
192 subnets = neutron_utils.get_subnets_by_network(neutron, networks[0])
193 self.assertEqual(1, len(subnets))
194 self.assertEqual(self.subnet_name, subnets[0].name)
196 def test_create_stack_x2(self):
198 Tests the creation of an OpenStack keypair that does not exist.
200 self.stack1 = heat_utils.create_stack(self.heat_client,
201 self.stack_settings1)
203 stack1_query_1 = heat_utils.get_stack(
204 self.heat_client, stack_settings=self.stack_settings1)
205 self.assertEqual(self.stack1, stack1_query_1)
207 stack1_query_2 = heat_utils.get_stack(
208 self.heat_client, stack_name=self.stack_settings1.name)
209 self.assertEqual(self.stack1, stack1_query_2)
211 stack1_query_3 = heat_utils.get_stack_by_id(self.heat_client,
213 self.assertEqual(self.stack1, stack1_query_3)
215 outputs = heat_utils.get_stack_outputs(self.heat_client,
217 self.assertIsNotNone(outputs)
218 self.assertEqual(0, len(outputs))
220 end_time = time.time() + create_stack.STACK_COMPLETE_TIMEOUT
223 while time.time() < end_time:
224 status = heat_utils.get_stack_status(self.heat_client,
226 if status == create_stack.STATUS_CREATE_COMPLETE:
229 elif status == create_stack.STATUS_CREATE_FAILED:
235 self.assertTrue(is_active)
237 self.stack2 = heat_utils.create_stack(self.heat_client,
238 self.stack_settings2)
240 stack2_query_1 = heat_utils.get_stack(
241 self.heat_client, stack_settings=self.stack_settings2)
242 self.assertEqual(self.stack2, stack2_query_1)
244 stack2_query_2 = heat_utils.get_stack(
245 self.heat_client, stack_name=self.stack_settings2.name)
246 self.assertEqual(self.stack2, stack2_query_2)
248 stack2_query_3 = heat_utils.get_stack_by_id(self.heat_client,
250 self.assertEqual(self.stack2, stack2_query_3)
252 outputs = heat_utils.get_stack_outputs(self.heat_client,
254 self.assertIsNotNone(outputs)
255 self.assertEqual(0, len(outputs))
257 end_time = time.time() + create_stack.STACK_COMPLETE_TIMEOUT
260 while time.time() < end_time:
261 status = heat_utils.get_stack_status(self.heat_client,
263 if status == create_stack.STATUS_CREATE_COMPLETE:
266 elif status == create_stack.STATUS_CREATE_FAILED:
272 self.assertTrue(is_active)