# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from heat_integrationtests.functional import functional_base
from heatclient import exc
import six
[docs]class StackPreviewTest(functional_base.FunctionalTestsBase):
template = '''
heat_template_version: 2015-04-30
parameters:
incomming:
type: string
resources:
one:
type: OS::Heat::TestResource
properties:
value: fred
two:
type: OS::Heat::TestResource
properties:
value: {get_param: incomming}
depends_on: one
outputs:
main_out:
value: {get_attr: [two, output]}
'''
env = '''
parameters:
incomming: abc
'''
[docs] def setUp(self):
super(StackPreviewTest, self).setUp()
self.client = self.orchestration_client
self.project_id = self.identity_client.project_id
def _assert_resource(self, res, stack_name):
self.assertEqual(stack_name, res['stack_name'])
self.assertEqual('INIT', res['resource_action'])
self.assertEqual('COMPLETE', res['resource_status'])
for field in ('resource_status_reason', 'physical_resource_id',
'description'):
self.assertIn(field, res)
self.assertEqual('', res[field])
# 'creation_time' and 'updated_time' are None when preview
for field in ('creation_time', 'updated_time'):
self.assertIn(field, res)
self.assertIsNone(res[field])
self.assertIn('output', res['attributes'])
# resource_identity
self.assertEqual(stack_name,
res['resource_identity']['stack_name'])
self.assertEqual('None', res['resource_identity']['stack_id'])
self.assertEqual(self.project_id,
res['resource_identity']['tenant'])
self.assertEqual('/resources/%s' % res['resource_name'],
res['resource_identity']['path'])
# stack_identity
self.assertEqual(stack_name,
res['stack_identity']['stack_name'])
self.assertEqual('None', res['stack_identity']['stack_id'])
self.assertEqual(self.project_id,
res['stack_identity']['tenant'])
self.assertEqual('', res['stack_identity']['path'])
def _assert_results(self, result, stack_name):
# global stuff.
self.assertEqual(stack_name, result['stack_name'])
self.assertTrue(result['disable_rollback'])
self.assertEqual('None', result['id'])
self.assertIsNone(result['parent'])
self.assertEqual('No description', result['template_description'])
# parameters
self.assertEqual('None', result['parameters']['OS::stack_id'])
self.assertEqual(stack_name, result['parameters']['OS::stack_name'])
self.assertEqual('abc', result['parameters']['incomming'])
[docs] def test_basic_pass(self):
stack_name = self._stack_rand_name()
result = self.client.stacks.preview(
template=self.template,
stack_name=stack_name,
disable_rollback=True,
environment=self.env).to_dict()
self._assert_results(result, stack_name)
for res in result['resources']:
self._assert_resource(res, stack_name)
self.assertEqual('OS::Heat::TestResource',
res['resource_type'])
# common properties
self.assertFalse(res['properties']['fail'])
self.assertEqual(0, res['properties']['wait_secs'])
self.assertFalse(res['properties']['update_replace'])
if res['resource_name'] == 'one':
self.assertEqual('fred', res['properties']['value'])
self.assertEqual(['two'], res['required_by'])
if res['resource_name'] == 'two':
self.assertEqual('abc', res['properties']['value'])
self.assertEqual([], res['required_by'])
[docs] def test_basic_fail(self):
stack_name = self._stack_rand_name()
# break the template so it fails validation.
wont_work = self.template.replace('get_param: incomming',
'get_param: missing')
excp = self.assertRaises(exc.HTTPBadRequest,
self.client.stacks.preview,
template=wont_work,
stack_name=stack_name,
disable_rollback=True,
environment=self.env)
self.assertIn('Property error: : resources.two.properties.value: '
': The Parameter (missing) was not provided.',
six.text_type(excp))
[docs] def test_nested_pass(self):
"""Nested stacks need to recurse down the stacks."""
main_template = '''
heat_template_version: 2015-04-30
parameters:
incomming:
type: string
resources:
main:
type: nested.yaml
properties:
value: {get_param: incomming}
outputs:
main_out:
value: {get_attr: [main, output]}
'''
nested_template = '''
heat_template_version: 2015-04-30
parameters:
value:
type: string
resources:
nested:
type: OS::Heat::TestResource
properties:
value: {get_param: value}
outputs:
output:
value: {get_attr: [nested, output]}
'''
stack_name = self._stack_rand_name()
result = self.client.stacks.preview(
disable_rollback=True,
stack_name=stack_name,
template=main_template,
files={'nested.yaml': nested_template},
environment=self.env).to_dict()
self._assert_results(result, stack_name)
# nested resources return a list of their resources.
res = result['resources'][0][0]
nested_stack_name = '%s-%s' % (stack_name,
res['parent_resource'])
self._assert_resource(res, nested_stack_name)
self.assertEqual('OS::Heat::TestResource',
res['resource_type'])
self.assertFalse(res['properties']['fail'])
self.assertEqual(0, res['properties']['wait_secs'])
self.assertFalse(res['properties']['update_replace'])
self.assertEqual('abc', res['properties']['value'])
self.assertEqual([], res['required_by'])
[docs] def test_res_group_with_nested_template(self):
main_template = '''
heat_template_version: 2015-04-30
resources:
fixed_network:
type: "OS::Neutron::Net"
rg:
type: "OS::Heat::ResourceGroup"
properties:
count: 1
resource_def:
type: nested.yaml
properties:
fixed_network_id: {get_resource: fixed_network}
'''
nested_template = '''
heat_template_version: 2015-04-30
parameters:
fixed_network_id:
type: string
resources:
port:
type: "OS::Neutron::Port"
properties:
network_id:
get_param: fixed_network_id
'''
stack_name = self._stack_rand_name()
result = self.client.stacks.preview(
disable_rollback=True,
stack_name=stack_name,
template=main_template,
files={'nested.yaml': nested_template}).to_dict()
resource_names = []
def get_resource_names(resources):
for item in resources:
if isinstance(item, dict):
resource_names.append(item['resource_name'])
else:
get_resource_names(item)
get_resource_names(result['resources'])
# ensure that fixed network and port here
self.assertIn('fixed_network', resource_names)
self.assertIn('port', resource_names)