snaps/openstack/utils/tests/heat_utils_tests.py

   1 # Copyright (c) 2017 Cable Television Laboratories, Inc. ("CableLabs")
   2 #                    and others.  All rights reserved.
   3 #
   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:
   7 #
   8 #     http://www.apache.org/licenses/LICENSE-2.0
   9 #
  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.
  15 import logging
  16 import pkg_resources
  17 import uuid
  18
  19 import time
  20
  21 from snaps.openstack import create_stack
  22 from snaps.openstack.create_flavor import OpenStackFlavor, FlavorSettings
  23
  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
  29
  30 __author__ = 'spisarski'
  31
  32 logger = logging.getLogger('heat_utils_tests')
  33
  34
  35 class HeatSmokeTests(OSComponentTestCase):
  36     """
  37     Tests to ensure that the heat client can communicate with the cloud
  38     """
  39
  40     def test_heat_connect_success(self):
  41         """
  42         Tests to ensure that the proper credentials can connect.
  43         """
  44         heat = heat_utils.heat_client(self.os_creds)
  45
  46         # This should not throw an exception
  47         stacks = heat.stacks.list()
  48         for stack in stacks:
  49             print stack
  50
  51     def test_heat_connect_fail(self):
  52         """
  53         Tests to ensure that the improper credentials cannot connect.
  54         """
  55         from snaps.openstack.os_credentials import OSCreds
  56
  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()
  63
  64         # This should throw an exception
  65         with self.assertRaises(Exception):
  66             for stack in stacks:
  67                 print stack
  68
  69
  70 class HeatUtilsCreateStackTests(OSComponentTestCase):
  71     """
  72     Test basic Heat functionality
  73     """
  74
  75     def setUp(self):
  76         """
  77         Instantiates OpenStack instances that cannot be spawned by Heat
  78         """
  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'
  84
  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()
  89
  90         # Create Flavor
  91         self.flavor_creator = OpenStackFlavor(
  92             self.os_creds,
  93             FlavorSettings(name=guid + '-flavor', ram=256, disk=10, vcpus=1))
  94         self.flavor_creator.create()
  95
  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)
 108         self.stack1 = None
 109         self.stack2 = None
 110         self.heat_client = heat_utils.heat_client(self.os_creds)
 111
 112     def tearDown(self):
 113         """
 114         Cleans the image and downloaded image file
 115         """
 116         if self.stack1:
 117             try:
 118                 heat_utils.delete_stack(self.heat_client, self.stack1)
 119             except:
 120                 pass
 121
 122         if self.stack2:
 123             try:
 124                 heat_utils.delete_stack(self.heat_client, self.stack2)
 125             except:
 126                 pass
 127
 128         if self.image_creator:
 129             try:
 130                 self.image_creator.clean()
 131             except:
 132                 pass
 133
 134         if self.flavor_creator:
 135             try:
 136                 self.flavor_creator.clean()
 137             except:
 138                 pass
 139
 140     def test_create_stack(self):
 141         """
 142         Tests the creation of an OpenStack Heat stack1 that does not exist.
 143         """
 144         self.stack1 = heat_utils.create_stack(self.heat_client,
 145                                               self.stack_settings1)
 146
 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)
 150
 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)
 154
 155         stack_query_3 = heat_utils.get_stack_by_id(self.heat_client,
 156                                                    self.stack1.id)
 157         self.assertEqual(self.stack1, stack_query_3)
 158
 159         outputs = heat_utils.get_stack_outputs(
 160             self.heat_client, self.stack1.id)
 161         self.assertIsNotNone(outputs)
 162         self.assertEqual(0, len(outputs))
 163
 164         end_time = time.time() + create_stack.STACK_COMPLETE_TIMEOUT
 165
 166         is_active = False
 167         while time.time() < end_time:
 168             status = heat_utils.get_stack_status(self.heat_client,
 169                                                  self.stack1.id)
 170             if status == create_stack.STATUS_CREATE_COMPLETE:
 171                 is_active = True
 172                 break
 173             elif status == create_stack.STATUS_CREATE_FAILED:
 174                 is_active = False
 175                 break
 176
 177             time.sleep(3)
 178
 179         self.assertTrue(is_active)
 180
 181         resources = heat_utils.get_resources(self.heat_client, self.stack1)
 182         self.assertIsNotNone(resources)
 183         self.assertEqual(4, len(resources))
 184
 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)
 191
 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)
 195
 196     def test_create_stack_x2(self):
 197         """
 198         Tests the creation of an OpenStack keypair that does not exist.
 199         """
 200         self.stack1 = heat_utils.create_stack(self.heat_client,
 201                                               self.stack_settings1)
 202
 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)
 206
 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)
 210
 211         stack1_query_3 = heat_utils.get_stack_by_id(self.heat_client,
 212                                                     self.stack1.id)
 213         self.assertEqual(self.stack1, stack1_query_3)
 214
 215         outputs = heat_utils.get_stack_outputs(self.heat_client,
 216                                                self.stack1.id)
 217         self.assertIsNotNone(outputs)
 218         self.assertEqual(0, len(outputs))
 219
 220         end_time = time.time() + create_stack.STACK_COMPLETE_TIMEOUT
 221
 222         is_active = False
 223         while time.time() < end_time:
 224             status = heat_utils.get_stack_status(self.heat_client,
 225                                                  self.stack1.id)
 226             if status == create_stack.STATUS_CREATE_COMPLETE:
 227                 is_active = True
 228                 break
 229             elif status == create_stack.STATUS_CREATE_FAILED:
 230                 is_active = False
 231                 break
 232
 233             time.sleep(3)
 234
 235         self.assertTrue(is_active)
 236
 237         self.stack2 = heat_utils.create_stack(self.heat_client,
 238                                               self.stack_settings2)
 239
 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)
 243
 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)
 247
 248         stack2_query_3 = heat_utils.get_stack_by_id(self.heat_client,
 249                                                     self.stack2.id)
 250         self.assertEqual(self.stack2, stack2_query_3)
 251
 252         outputs = heat_utils.get_stack_outputs(self.heat_client,
 253                                                self.stack2.id)
 254         self.assertIsNotNone(outputs)
 255         self.assertEqual(0, len(outputs))
 256
 257         end_time = time.time() + create_stack.STACK_COMPLETE_TIMEOUT
 258
 259         is_active = False
 260         while time.time() < end_time:
 261             status = heat_utils.get_stack_status(self.heat_client,
 262                                                  self.stack2.id)
 263             if status == create_stack.STATUS_CREATE_COMPLETE:
 264                 is_active = True
 265                 break
 266             elif status == create_stack.STATUS_CREATE_FAILED:
 267                 is_active = False
 268                 break
 269
 270             time.sleep(3)
 271
 272         self.assertTrue(is_active)