from heat.engine import parser
from heat.engine import resource
from heat.engine import scheduler
+from heat.engine import template as tmpl
from heat.openstack.common import log as logging
cfg.CONF.max_nested_stack_depth
raise exception.RequestLimitExceeded(message=msg)
template = parser.Template(child_template)
+ if ((len(template[tmpl.RESOURCES]) +
+ self.stack.root_stack.total_resources() >
+ cfg.CONF.max_resources_per_stack)):
+ raise exception.StackResourceLimitExceeded()
self._outputs_to_attribs(child_template)
# Note we disable rollback for nested stacks, since they
import copy
+from oslo.config import cfg
+
+cfg.CONF.import_opt('max_resources_per_stack', 'heat.common.config')
+
from heat.common import exception
from heat.common import template_format
from heat.common import urlfetch
self.m.VerifyAll()
+ def test_nested_stack_create_exceeds_resource_limit(self):
+ cfg.CONF.set_override('max_resources_per_stack', 1)
+ resource._register_class('GenericResource',
+ generic_rsrc.GenericResource)
+ urlfetch.get('https://server.test/the.template').MultipleTimes().\
+ AndReturn('''
+HeatTemplateFormatVersion: '2012-12-12'
+Parameters:
+ KeyName:
+ Type: String
+Resources:
+ NestedResource:
+ Type: GenericResource
+Outputs:
+ Foo:
+ Value: bar
+''')
+ self.m.ReplayAll()
+
+ t = template_format.parse(self.test_template)
+ stack = self.parse_stack(t)
+ stack.create()
+ self.assertEquals(stack.state, (stack.CREATE, stack.FAILED))
+ self.assertIn('Maximum resources per stack exceeded',
+ stack.status_reason)
+
+ self.m.VerifyAll()
+
+ def test_nested_stack_create_equals_resource_limit(self):
+ cfg.CONF.set_override('max_resources_per_stack', 2)
+ resource._register_class('GenericResource',
+ generic_rsrc.GenericResource)
+ urlfetch.get('https://server.test/the.template').MultipleTimes().\
+ AndReturn('''
+HeatTemplateFormatVersion: '2012-12-12'
+Parameters:
+ KeyName:
+ Type: String
+Resources:
+ NestedResource:
+ Type: GenericResource
+Outputs:
+ Foo:
+ Value: bar
+''')
+ self.m.ReplayAll()
+
+ t = template_format.parse(self.test_template)
+ stack = self.parse_stack(t)
+ stack.create()
+ self.assertEquals(stack.state, (stack.CREATE, stack.COMPLETE))
+ self.assertIn('NestedResource',
+ stack.resources['the_nested'].nested().resources)
+
+ self.m.VerifyAll()
+
def test_nested_stack_update(self):
urlfetch.get('https://server.test/the.template').MultipleTimes().\
AndReturn(self.nested_template)