OpenStack Orchestration (Heat)
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
heat/heat/tests/test_nested_stack.py

381 lines
12 KiB

#
# 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.
import mock
from oslo_config import cfg
from requests import exceptions
import six
import yaml
from heat.common import exception
from heat.common import identifier
from heat.common import template_format
from heat.common import urlfetch
from heat.engine import resource
from heat.engine.resources.aws.cfn import stack as stack_res
from heat.engine import rsrc_defn
from heat.engine import stack as parser
from heat.engine import template
from heat.objects import resource_data as resource_data_object
from heat.tests import common
from heat.tests import generic_resource as generic_rsrc
from heat.tests import utils
class NestedStackTest(common.HeatTestCase):
test_template = '''
HeatTemplateFormatVersion: '2012-12-12'
Resources:
the_nested:
Type: AWS::CloudFormation::Stack
Properties:
TemplateURL: https://server.test/the.template
Parameters:
KeyName: foo
'''
nested_template = '''
HeatTemplateFormatVersion: '2012-12-12'
Parameters:
KeyName:
Type: String
Outputs:
Foo:
Value: bar
'''
update_template = '''
HeatTemplateFormatVersion: '2012-12-12'
Parameters:
KeyName:
Type: String
Outputs:
Bar:
Value: foo
'''
def setUp(self):
super(NestedStackTest, self).setUp()
self.patchobject(urlfetch, 'get')
def validate_stack(self, template):
t = template_format.parse(template)
stack = self.parse_stack(t)
res = stack.validate()
self.assertIsNone(res)
return stack
def parse_stack(self, t, data=None):
ctx = utils.dummy_context('test_username', 'aaaa', 'password')
stack_name = 'test_stack'
tmpl = template.Template(t)
stack = parser.Stack(ctx, stack_name, tmpl, adopt_stack_data=data)
stack.store()
return stack
@mock.patch.object(parser.Stack, 'root_stack_id')
@mock.patch.object(parser.Stack, 'total_resources')
def test_nested_stack_three_deep(self, tr, rsi):
root_template = '''
HeatTemplateFormatVersion: 2012-12-12
Resources:
Nested:
Type: AWS::CloudFormation::Stack
Properties:
TemplateURL: 'https://server.test/depth1.template'
'''
depth1_template = '''
HeatTemplateFormatVersion: 2012-12-12
Resources:
Nested:
Type: AWS::CloudFormation::Stack
Properties:
TemplateURL: 'https://server.test/depth2.template'
'''
depth2_template = '''
HeatTemplateFormatVersion: 2012-12-12
Resources:
Nested:
Type: AWS::CloudFormation::Stack
Properties:
TemplateURL: 'https://server.test/depth3.template'
Parameters:
KeyName: foo
'''
urlfetch.get.side_effect = [
depth1_template,
depth2_template,
self.nested_template]
rsi.return_value = '1234'
tr.return_value = 2
self.validate_stack(root_template)
calls = [mock.call('https://server.test/depth1.template'),
mock.call('https://server.test/depth2.template'),
mock.call('https://server.test/depth3.template')]
urlfetch.get.assert_has_calls(calls)
tr.assert_called_with('1234')
@mock.patch.object(parser.Stack, 'root_stack_id')
@mock.patch.object(parser.Stack, 'total_resources')
def test_nested_stack_six_deep(self, tr, rsi):
tmpl = '''
HeatTemplateFormatVersion: 2012-12-12
Resources:
Nested:
Type: AWS::CloudFormation::Stack
Properties:
TemplateURL: 'https://server.test/depth%i.template'
'''
root_template = tmpl % 1
depth1_template = tmpl % 2
depth2_template = tmpl % 3
depth3_template = tmpl % 4
depth4_template = tmpl % 5
depth5_template = tmpl % 6
depth5_template += '''
Parameters:
KeyName: foo
'''
urlfetch.get.side_effect = [
depth1_template,
depth2_template,
depth3_template,
depth4_template,
depth5_template,
self.nested_template]
rsi.return_value = '1234'
tr.return_value = 5
t = template_format.parse(root_template)
stack = self.parse_stack(t)
res = self.assertRaises(exception.StackValidationFailed,
stack.validate)
self.assertIn('Recursion depth exceeds', six.text_type(res))
calls = [mock.call('https://server.test/depth1.template'),
mock.call('https://server.test/depth2.template'),
mock.call('https://server.test/depth3.template'),
mock.call('https://server.test/depth4.template'),
mock.call('https://server.test/depth5.template'),
mock.call('https://server.test/depth6.template')]
urlfetch.get.assert_has_calls(calls)
def test_nested_stack_four_wide(self):
root_template = '''
HeatTemplateFormatVersion: 2012-12-12
Resources:
Nested:
Type: AWS::CloudFormation::Stack
Properties:
TemplateURL: 'https://server.test/depth1.template'
Parameters:
KeyName: foo
Nested2:
Type: AWS::CloudFormation::Stack
Properties:
TemplateURL: 'https://server.test/depth2.template'
Parameters:
KeyName: foo
Nested3:
Type: AWS::CloudFormation::Stack
Properties:
TemplateURL: 'https://server.test/depth3.template'
Parameters:
KeyName: foo
Nested4:
Type: AWS::CloudFormation::Stack
Properties:
TemplateURL: 'https://server.test/depth4.template'
Parameters:
KeyName: foo
'''
urlfetch.get.return_value = self.nested_template
self.validate_stack(root_template)
calls = [mock.call('https://server.test/depth1.template'),
mock.call('https://server.test/depth2.template'),
mock.call('https://server.test/depth3.template'),
mock.call('https://server.test/depth4.template')]
urlfetch.get.assert_has_calls(calls, any_order=True)
@mock.patch.object(parser.Stack, 'root_stack_id')
@mock.patch.object(parser.Stack, 'total_resources')
def test_nested_stack_infinite_recursion(self, tr, rsi):
tmpl = '''
HeatTemplateFormatVersion: 2012-12-12
Resources:
Nested:
Type: AWS::CloudFormation::Stack
Properties:
TemplateURL: 'https://server.test/the.template'
'''
urlfetch.get.return_value = tmpl
t = template_format.parse(tmpl)
stack = self.parse_stack(t)
rsi.return_value = '1234'
tr.return_value = 2
res = self.assertRaises(exception.StackValidationFailed,
stack.validate)
self.assertIn('Recursion depth exceeds', six.text_type(res))
expected_count = cfg.CONF.get('max_nested_stack_depth') + 1
self.assertEqual(expected_count, urlfetch.get.call_count)
def test_child_params(self):
t = template_format.parse(self.test_template)
stack = self.parse_stack(t)
nested_stack = stack['the_nested']
nested_stack.properties.data[nested_stack.PARAMETERS] = {'foo': 'bar'}
self.assertEqual({'foo': 'bar'}, nested_stack.child_params())
def test_child_template_when_file_is_fetched(self):
urlfetch.get.return_value = 'template_file'
t = template_format.parse(self.test_template)
stack = self.parse_stack(t)
nested_stack = stack['the_nested']
with mock.patch('heat.common.template_format.parse') as mock_parse:
mock_parse.return_value = 'child_template'
self.assertEqual('child_template', nested_stack.child_template())
mock_parse.assert_called_once_with('template_file')
def test_child_template_when_fetching_file_fails(self):
urlfetch.get.side_effect = exceptions.RequestException()
t = template_format.parse(self.test_template)
stack = self.parse_stack(t)
nested_stack = stack['the_nested']
self.assertRaises(ValueError, nested_stack.child_template)
def test_child_template_when_io_error(self):
msg = 'Failed to retrieve template'
urlfetch.get.side_effect = urlfetch.URLFetchError(msg)
t = template_format.parse(self.test_template)
stack = self.parse_stack(t)
nested_stack = stack['the_nested']
self.assertRaises(ValueError, nested_stack.child_template)
class ResDataResource(generic_rsrc.GenericResource):
def handle_create(self):
self.data_set("test", 'A secret value', True)
class ResDataStackTest(common.HeatTestCase):
tmpl = '''
HeatTemplateFormatVersion: "2012-12-12"
Parameters:
KeyName:
Type: String
Resources:
res:
Type: "res.data.resource"
Outputs:
Foo:
Value: bar
'''
def setUp(self):
super(ResDataStackTest, self).setUp()
resource._register_class("res.data.resource", ResDataResource)
def create_stack(self, template):
t = template_format.parse(template)
stack = utils.parse_stack(t)
stack.create()
self.assertEqual((stack.CREATE, stack.COMPLETE), stack.state)
return stack
def test_res_data_delete(self):
stack = self.create_stack(self.tmpl)
res = stack['res']
stack.delete()
self.assertEqual((stack.DELETE, stack.COMPLETE), stack.state)
self.assertRaises(
exception.NotFound,
resource_data_object.ResourceData.get_val, res, 'test')
class NestedStackCrudTest(common.HeatTestCase):
nested_template = '''
HeatTemplateFormatVersion: '2012-12-12'
Parameters:
KeyName:
Type: String
Outputs:
Foo:
Value: bar
'''
def setUp(self):
super(NestedStackCrudTest, self).setUp()
self.ctx = utils.dummy_context('test_username', 'aaaa', 'password')
empty_template = {"HeatTemplateFormatVersion": "2012-12-12"}
self.stack = parser.Stack(self.ctx, 'test',
template.Template(empty_template))
self.stack.store()
self.patchobject(urlfetch, 'get', return_value=self.nested_template)
self.nested_parsed = yaml.load(self.nested_template)
self.nested_params = {"KeyName": "foo"}
self.defn = rsrc_defn.ResourceDefinition(
'test_t_res',
'AWS::CloudFormation::Stack',
{"TemplateURL": "https://server.test/the.template",
"Parameters": self.nested_params})
self.res = stack_res.NestedStack('test_t_res',
self.defn, self.stack)
self.assertIsNone(self.res.validate())
self.res._store()
def test_handle_create(self):
self.res.create_with_template = mock.Mock(return_value=None)
self.res.handle_create()
self.res.create_with_template.assert_called_once_with(
self.nested_parsed, self.nested_params, None, adopt_data=None)
def test_handle_adopt(self):
self.res.create_with_template = mock.Mock(return_value=None)
self.res.handle_adopt(resource_data={'resource_id': 'fred'})
self.res.create_with_template.assert_called_once_with(
self.nested_parsed, self.nested_params, None,
adopt_data={'resource_id': 'fred'})
def test_handle_update(self):
self.res.update_with_template = mock.Mock(return_value=None)
self.res.handle_update(self.defn, None, None)
self.res.update_with_template.assert_called_once_with(
self.nested_parsed, self.nested_params, None)
def test_handle_delete(self):
self.res.rpc_client = mock.MagicMock()
self.res.action = self.res.CREATE
stack_identity = identifier.HeatIdentifier(
self.ctx.tenant_id,
self.res.physical_resource_name(),
self.res.resource_id)
self.res.handle_delete()
self.res.rpc_client.return_value.delete_stack.assert_called_once_with(
self.ctx, stack_identity)