1 # Copyright (c) 2015 Intel Research and Development Ireland Ltd.
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
7 # http://www.apache.org/licenses/LICENSE-2.0
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
15 from __future__ import print_function
17 from __future__ import absolute_import
21 import experimental_framework.common as common
23 from experimental_framework import heat_manager, APEX_LAKE_ROOT
26 __author__ = 'gpetralx'
29 def get_mock_heat(version, *args, **kwargs):
33 class MockStacks(object):
35 def __init__(self, stacks):
40 for stack in self.stacks:
41 list_name.append(stack.stack_name)
45 def validate(self, template=None):
49 for stack in self.stacks:
51 return self.stacks.remove(stack)
53 def create(self, stack_name=None, files=None, template=None,
56 self.stacks.append(MockStack(stack_name))
59 class MockStacks_2(object):
61 def __init__(self, stacks):
68 class MockStack(object):
70 def __init__(self, stack_name):
71 self.name = stack_name
74 def stack_status(self):
75 return self.stack_name + '_status'
85 def __eq__(self, other):
86 return self.name == other
89 class MockHeat(object):
92 stacks = [MockStack('stack_1'), MockStack('stack_2')]
93 self.stacks_list = MockStacks(stacks)
97 return self.stacks_list
100 class MockHeat_2(MockHeat):
103 stacks = [MockStack('stack_1'), MockStack('stack_2')]
104 self.stacks_list = MockStacks_2(stacks)
107 class HeatManagerMock(heat_manager.HeatManager):
110 if self.heat is None:
111 self.heat = MockHeat()
114 class HeatManagerMock_2(heat_manager.HeatManager):
117 if self.heat is None:
118 self.heat = MockHeat_2()
121 class TestHeatManager(unittest.TestCase):
125 credentials['ip_controller'] = '1.1.1.1'
126 credentials['heat_url'] = 'http://heat_url'
127 credentials['user'] = 'user'
128 credentials['password'] = 'password'
129 credentials['auth_uri'] = 'auth_uri'
130 credentials['project'] = 'project'
131 self.heat_manager = HeatManagerMock(credentials)
132 self.heat_manager.init_heat()
137 def test_is_stack_deployed_for_success(self):
138 self.assertTrue(self.heat_manager.is_stack_deployed('stack_1'))
139 self.assertFalse(self.heat_manager.is_stack_deployed('stack_n'))
141 def test_check_status_for_success(self):
142 self.assertEqual('stack_1_status',
143 self.heat_manager.check_stack_status('stack_1'))
144 self.assertEqual('NOT_FOUND',
145 self.heat_manager.check_stack_status('stack_x'))
147 def test_validate_template_for_success(self):
148 template_file = os.path.join(
150 'tests/data/test_templates/VTC_base_single_vm_wait_1.yaml')
151 with self.assertRaises(ValueError):
152 self.heat_manager.validate_heat_template(template_file)
154 def test_delete_stack_for_success(self):
155 self.assertTrue(self.heat_manager.delete_stack('stack_1'))
156 self.assertFalse(self.heat_manager.delete_stack('stack_x'))
158 def test_delete_stack_for_success_2(self):
159 self.assertTrue(self.heat_manager.delete_stack('stack_1'))
161 @mock.patch('experimental_framework.common.LOG')
162 @mock.patch('heatclient.common.template_utils.get_template_contents')
163 @mock.patch('heatclient.client.Client')
164 # @mock.patch('heatclient.client.Client', side_effect=DummyHeatClient)
165 def test_create_stack_for_success(self, mock_stack_create,
166 mock_get_template_contents,
168 return_value = ({'template': 'template'}, 'template')
169 mock_get_template_contents.return_value = return_value
170 self.heat_manager.create_stack('template', 'stack_n', 'parameters')
171 self.assertTrue(self.heat_manager.is_stack_deployed('stack_n'))
174 class TestHeatManager_2(unittest.TestCase):
178 credentials['ip_controller'] = '1.1.1.1'
179 credentials['heat_url'] = 'http://heat_url'
180 credentials['user'] = 'user'
181 credentials['password'] = 'password'
182 credentials['auth_uri'] = 'auth_uri'
183 credentials['project'] = 'project'
184 self.heat_manager = HeatManagerMock_2(credentials)
189 def test_delete_stack_for_success_2(self):
190 common.LOG = logging.getLogger()
191 self.assertFalse(self.heat_manager.delete_stack('stack_1'))
194 class ServiceCatalog():
196 def url_for(self, service_type):
197 return 'http://heat_url'
200 class KeystoneMock(object):
203 def auth_token(self):
206 service_catalog = ServiceCatalog()
209 class TestHeatInit(unittest.TestCase):
213 credentials['ip_controller'] = '1.1.1.1'
214 credentials['heat_url'] = 'http://heat_url'
215 credentials['user'] = 'user'
216 credentials['password'] = 'password'
217 credentials['auth_uri'] = 'auth_uri'
218 credentials['project'] = 'project'
219 self.heat_manager = heat_manager.HeatManager(credentials)
224 @mock.patch('heatclient.client.Client')
225 @mock.patch('keystoneclient.v2_0.client.Client')
226 def test_heat_init_for_sanity(self, keystone_client, heat_client):
227 keystone_client.return_value = KeystoneMock()
228 heat_client.return_value = MockHeat()
229 self.heat_manager.init_heat()
230 keystone_client.assert_called_once_with(username='user',
231 tenant_name='project',
234 heat_client.assert_called_once_with('1', endpoint='http://heat_url',