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
  29
  30 __author__ = 'spisarski'
  31
  32 logger = logging.getLogger('nova_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_nova_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_nova_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 = self.__class__.__name__ + '-' + str(guid) + '-stack1'
  81         stack_name2 = self.__class__.__name__ + '-' + str(guid) + '-stack2'
  82
  83         self.image_creator = OpenStackImage(
  84             self.os_creds, openstack_tests.cirros_image_settings(
  85                 name=self.__class__.__name__ + '-' + str(guid) + '-image',
  86                 image_metadata=self.image_metadata))
  87         self.image_creator.create()
  88
  89         # Create Flavor
  90         self.flavor_creator = OpenStackFlavor(
  91             self.os_creds,
  92             FlavorSettings(name=guid + '-flavor', ram=256, disk=10, vcpus=1))
  93         self.flavor_creator.create()
  94
  95         env_values = {'image_name': self.image_creator.image_settings.name,
  96                       'flavor_name': self.flavor_creator.flavor_settings.name}
  97         heat_tmplt_path = pkg_resources.resource_filename(
  98             'snaps.openstack.tests.heat', 'test_heat_template.yaml')
  99         self.stack_settings1 = StackSettings(
 100             name=stack_name1, template_path=heat_tmplt_path,
 101             env_values=env_values)
 102         self.stack_settings2 = StackSettings(
 103             name=stack_name2, template_path=heat_tmplt_path,
 104             env_values=env_values)
 105         self.stack1 = None
 106         self.stack2 = None
 107         self.heat_client = heat_utils.heat_client(self.os_creds)
 108
 109     def tearDown(self):
 110         """
 111         Cleans the image and downloaded image file
 112         """
 113         if self.stack1:
 114             try:
 115                 heat_utils.delete_stack(self.heat_client, self.stack1)
 116             except:
 117                 pass
 118
 119         if self.stack2:
 120             try:
 121                 heat_utils.delete_stack(self.heat_client, self.stack2)
 122             except:
 123                 pass
 124
 125         if self.image_creator:
 126             try:
 127                 self.image_creator.clean()
 128             except:
 129                 pass
 130
 131         if self.flavor_creator:
 132             try:
 133                 self.flavor_creator.clean()
 134             except:
 135                 pass
 136
 137     def test_create_stack(self):
 138         """
 139         Tests the creation of an OpenStack Heat stack1 that does not exist.
 140         """
 141         self.stack1 = heat_utils.create_stack(self.heat_client,
 142                                               self.stack_settings1)
 143
 144         stack_query_1 = heat_utils.get_stack(
 145             self.heat_client, stack_settings=self.stack_settings1)
 146         self.assertEqual(self.stack1, stack_query_1)
 147
 148         stack_query_2 = heat_utils.get_stack(
 149             self.heat_client, stack_name=self.stack_settings1.name)
 150         self.assertEqual(self.stack1, stack_query_2)
 151
 152         stack_query_3 = heat_utils.get_stack_by_id(self.heat_client,
 153                                                    self.stack1.id)
 154         self.assertEqual(self.stack1, stack_query_3)
 155
 156         outputs = heat_utils.get_stack_outputs(
 157             self.heat_client, self.stack1.id)
 158         self.assertIsNotNone(outputs)
 159         self.assertEqual(0, len(outputs))
 160
 161         end_time = time.time() + create_stack.STACK_COMPLETE_TIMEOUT
 162
 163         is_active = False
 164         while time.time() < end_time:
 165             status = heat_utils.get_stack_status(self.heat_client,
 166                                                  self.stack1.id)
 167             if status == create_stack.STATUS_CREATE_COMPLETE:
 168                 is_active = True
 169                 break
 170             elif status == create_stack.STATUS_CREATE_FAILED:
 171                 is_active = False
 172                 break
 173
 174             time.sleep(3)
 175
 176         self.assertTrue(is_active)
 177
 178     def test_create_stack_x2(self):
 179         """
 180         Tests the creation of an OpenStack keypair that does not exist.
 181         """
 182         self.stack1 = heat_utils.create_stack(self.heat_client,
 183                                               self.stack_settings1)
 184
 185         stack1_query_1 = heat_utils.get_stack(
 186             self.heat_client, stack_settings=self.stack_settings1)
 187         self.assertEqual(self.stack1, stack1_query_1)
 188
 189         stack1_query_2 = heat_utils.get_stack(
 190             self.heat_client, stack_name=self.stack_settings1.name)
 191         self.assertEqual(self.stack1, stack1_query_2)
 192
 193         stack1_query_3 = heat_utils.get_stack_by_id(self.heat_client,
 194                                                     self.stack1.id)
 195         self.assertEqual(self.stack1, stack1_query_3)
 196
 197         outputs = heat_utils.get_stack_outputs(self.heat_client,
 198                                                self.stack1.id)
 199         self.assertIsNotNone(outputs)
 200         self.assertEqual(0, len(outputs))
 201
 202         end_time = time.time() + create_stack.STACK_COMPLETE_TIMEOUT
 203
 204         is_active = False
 205         while time.time() < end_time:
 206             status = heat_utils.get_stack_status(self.heat_client,
 207                                                  self.stack1.id)
 208             if status == create_stack.STATUS_CREATE_COMPLETE:
 209                 is_active = True
 210                 break
 211             elif status == create_stack.STATUS_CREATE_FAILED:
 212                 is_active = False
 213                 break
 214
 215             time.sleep(3)
 216
 217         self.assertTrue(is_active)
 218
 219         self.stack2 = heat_utils.create_stack(self.heat_client,
 220                                               self.stack_settings2)
 221
 222         stack2_query_1 = heat_utils.get_stack(
 223             self.heat_client, stack_settings=self.stack_settings2)
 224         self.assertEqual(self.stack2, stack2_query_1)
 225
 226         stack2_query_2 = heat_utils.get_stack(
 227             self.heat_client, stack_name=self.stack_settings2.name)
 228         self.assertEqual(self.stack2, stack2_query_2)
 229
 230         stack2_query_3 = heat_utils.get_stack_by_id(self.heat_client,
 231                                                     self.stack2.id)
 232         self.assertEqual(self.stack2, stack2_query_3)
 233
 234         outputs = heat_utils.get_stack_outputs(self.heat_client,
 235                                                self.stack2.id)
 236         self.assertIsNotNone(outputs)
 237         self.assertEqual(0, len(outputs))
 238
 239         end_time = time.time() + create_stack.STACK_COMPLETE_TIMEOUT
 240
 241         is_active = False
 242         while time.time() < end_time:
 243             status = heat_utils.get_stack_status(self.heat_client,
 244                                                  self.stack2.id)
 245             if status == create_stack.STATUS_CREATE_COMPLETE:
 246                 is_active = True
 247                 break
 248             elif status == create_stack.STATUS_CREATE_FAILED:
 249                 is_active = False
 250                 break
 251
 252             time.sleep(3)
 253
 254         self.assertTrue(is_active)