# License for the specific language governing permissions and limitations
# under the License.
-
-import os
import datetime
import copy
import eventlet
import mox
-from heat.common import context
from heat.common import template_format
from heat.engine.resources import autoscaling as asc
from heat.engine.resources import loadbalancer
from heat.engine.resources import instance
from heat.engine.resources import cloud_watch
from heat.engine import clients
-from heat.engine import parser
from heat.engine import scheduler
from heat.engine.resource import Metadata
from heat.openstack.common import timeutils
from heat.tests.v1_1 import fakes
from heat.tests.common import HeatTestCase
from heat.tests.utils import setup_dummy_db
+from heat.tests.utils import parse_stack
+
+as_template = '''
+{
+ "AWSTemplateFormatVersion" : "2010-09-09",
+ "Description" : "AutoScaling Test",
+ "Parameters" : {},
+ "Resources" : {
+ "WebServerGroup" : {
+ "Type" : "AWS::AutoScaling::AutoScalingGroup",
+ "Properties" : {
+ "AvailabilityZones" : ["nova"],
+ "LaunchConfigurationName" : { "Ref" : "LaunchConfig" },
+ "MinSize" : "1",
+ "MaxSize" : "3",
+ "LoadBalancerNames" : [ { "Ref" : "ElasticLoadBalancer" } ]
+ }
+ },
+ "WebServerScaleUpPolicy" : {
+ "Type" : "AWS::AutoScaling::ScalingPolicy",
+ "Properties" : {
+ "AdjustmentType" : "ChangeInCapacity",
+ "AutoScalingGroupName" : { "Ref" : "WebServerGroup" },
+ "Cooldown" : "60",
+ "ScalingAdjustment" : "1"
+ }
+ },
+ "WebServerScaleDownPolicy" : {
+ "Type" : "AWS::AutoScaling::ScalingPolicy",
+ "Properties" : {
+ "AdjustmentType" : "ChangeInCapacity",
+ "AutoScalingGroupName" : { "Ref" : "WebServerGroup" },
+ "Cooldown" : "60",
+ "ScalingAdjustment" : "-1"
+ }
+ },
+ "ElasticLoadBalancer" : {
+ "Type" : "AWS::ElasticLoadBalancing::LoadBalancer",
+ },
+ "LaunchConfig" : {
+ "Type" : "AWS::AutoScaling::LaunchConfiguration",
+ "Properties": {
+ "ImageId" : "foo",
+ "InstanceType" : "bar",
+ }
+ }
+ }
+}
+'''
+
+alarm_template = '''
+{
+ "AWSTemplateFormatVersion" : "2010-09-09",
+ "Description" : "Alarm Test",
+ "Parameters" : {},
+ "Resources" : {
+ "MEMAlarmHigh": {
+ "Type": "AWS::CloudWatch::Alarm",
+ "Properties": {
+ "AlarmDescription": "Scale-up if MEM > 50% for 1 minute",
+ "MetricName": "MemoryUtilization",
+ "Namespace": "system/linux",
+ "Statistic": "Average",
+ "Period": "60",
+ "EvaluationPeriods": "1",
+ "Threshold": "50",
+ "AlarmActions": [],
+ "Dimensions": [],
+ "ComparisonOperator": "GreaterThanThreshold"
+ }
+ }
+ }
+}
+'''
class AutoScalingTest(HeatTestCase):
setup_dummy_db()
self.fc = fakes.FakeClient()
self.m.StubOutWithMock(clients.OpenStackClients, 'nova')
- clients.OpenStackClients.nova().MultipleTimes().AndReturn(self.fc)
-
- def load_template(self):
- self.path = os.path.dirname(os.path.realpath(__file__)).\
- replace('heat/tests', 'templates')
- f = open("%s/AutoScalingMultiAZSample.template" % self.path)
- t = template_format.parse(f.read())
- f.close()
- return t
-
- def parse_stack(self, t):
- self.m.ReplayAll()
- ctx = context.RequestContext.from_dict({
- 'tenant': 'test_tenant',
- 'username': 'test_username',
- 'password': 'password',
- 'auth_url': 'http://localhost:5000/v2.0'})
- template = parser.Template(t)
- params = parser.Parameters('test_stack', template, {'KeyName': 'test'})
- stack = parser.Stack(ctx, 'test_stack', template, params)
-
- return stack
def create_scaling_group(self, t, stack, resource_name):
resource = asc.AutoScalingGroup(resource_name,
Metadata.__set__(mox.IgnoreArg(), expected).AndReturn(None)
def test_scaling_group_update(self):
- t = self.load_template()
- stack = self.parse_stack(t)
+ t = template_format.parse(as_template)
+ stack = parse_stack(t)
self._stub_lb_reload(['WebServerGroup-0'])
now = timeutils.utcnow()
self.m.VerifyAll()
def test_scaling_group_update_ok_maxsize(self):
- t = self.load_template()
+ t = template_format.parse(as_template)
properties = t['Resources']['WebServerGroup']['Properties']
properties['MinSize'] = '1'
properties['MaxSize'] = '3'
- stack = self.parse_stack(t)
+ stack = parse_stack(t)
self._stub_lb_reload(['WebServerGroup-0'])
now = timeutils.utcnow()
self.m.VerifyAll()
def test_scaling_group_update_ok_minsize(self):
- t = self.load_template()
+ t = template_format.parse(as_template)
properties = t['Resources']['WebServerGroup']['Properties']
properties['MinSize'] = '1'
properties['MaxSize'] = '3'
- stack = self.parse_stack(t)
+ stack = parse_stack(t)
self._stub_lb_reload(['WebServerGroup-0'])
now = timeutils.utcnow()
self.m.VerifyAll()
def test_scaling_group_update_ok_desired(self):
- t = self.load_template()
+ t = template_format.parse(as_template)
properties = t['Resources']['WebServerGroup']['Properties']
properties['MinSize'] = '1'
properties['MaxSize'] = '3'
- stack = self.parse_stack(t)
+ stack = parse_stack(t)
self._stub_lb_reload(['WebServerGroup-0'])
now = timeutils.utcnow()
self.m.VerifyAll()
def test_scaling_group_update_ok_desired_remove(self):
- t = self.load_template()
+ t = template_format.parse(as_template)
properties = t['Resources']['WebServerGroup']['Properties']
properties['DesiredCapacity'] = '2'
- stack = self.parse_stack(t)
+ stack = parse_stack(t)
self._stub_lb_reload(['WebServerGroup-0', 'WebServerGroup-1'])
now = timeutils.utcnow()
self.m.VerifyAll()
def test_scaling_group_update_ok_cooldown(self):
- t = self.load_template()
+ t = template_format.parse(as_template)
properties = t['Resources']['WebServerGroup']['Properties']
properties['Cooldown'] = '60'
- stack = self.parse_stack(t)
+ stack = parse_stack(t)
self._stub_lb_reload(['WebServerGroup-0'])
now = timeutils.utcnow()
Make sure that we can change the update-able properties
without replacing the Alarm resource.
'''
- t = self.load_template()
+ t = template_format.parse(alarm_template)
#short circuit the alarm's references
properties = t['Resources']['MEMAlarmHigh']['Properties']
properties['AlarmActions'] = ['a']
properties['Dimensions'] = [{'a': 'v'}]
- stack = self.parse_stack(t)
+ stack = parse_stack(t)
# the watch rule needs a valid stack_id
stack.store()
Make sure that the Alarm resource IS replaced when non-update-able
properties are changed.
'''
- t = self.load_template()
+ t = template_format.parse(alarm_template)
#short circuit the alarm's references
properties = t['Resources']['MEMAlarmHigh']['Properties']
properties['AlarmActions'] = ['a']
properties['Dimensions'] = [{'a': 'v'}]
- stack = self.parse_stack(t)
+ stack = parse_stack(t)
# the watch rule needs a valid stack_id
stack.store()
self.m.VerifyAll()
def test_scaling_group_adjust(self):
- t = self.load_template()
- stack = self.parse_stack(t)
+ t = template_format.parse(as_template)
+ stack = parse_stack(t)
# start with 3
properties = t['Resources']['WebServerGroup']['Properties']
self.m.VerifyAll()
def test_scaling_group_nochange(self):
- t = self.load_template()
- stack = self.parse_stack(t)
+ t = template_format.parse(as_template)
+ stack = parse_stack(t)
# Create initial group, 2 instances
properties = t['Resources']['WebServerGroup']['Properties']
self.m.VerifyAll()
def test_scaling_group_percent(self):
- t = self.load_template()
- stack = self.parse_stack(t)
+ t = template_format.parse(as_template)
+ stack = parse_stack(t)
# Create initial group, 2 instances
properties = t['Resources']['WebServerGroup']['Properties']
resource.delete()
def test_scaling_group_cooldown_toosoon(self):
- t = self.load_template()
- stack = self.parse_stack(t)
+ t = template_format.parse(as_template)
+ stack = parse_stack(t)
# Create initial group, 2 instances, Cooldown 60s
properties = t['Resources']['WebServerGroup']['Properties']
resource.delete()
def test_scaling_group_cooldown_ok(self):
- t = self.load_template()
- stack = self.parse_stack(t)
+ t = template_format.parse(as_template)
+ stack = parse_stack(t)
# Create initial group, 2 instances, Cooldown 60s
properties = t['Resources']['WebServerGroup']['Properties']
resource.delete()
def test_scaling_group_cooldown_zero(self):
- t = self.load_template()
- stack = self.parse_stack(t)
+ t = template_format.parse(as_template)
+ stack = parse_stack(t)
# Create initial group, 2 instances, Cooldown 0
properties = t['Resources']['WebServerGroup']['Properties']
self.m.VerifyAll()
def test_scaling_policy_up(self):
- t = self.load_template()
- stack = self.parse_stack(t)
+ t = template_format.parse(as_template)
+ stack = parse_stack(t)
# Create initial group
self._stub_lb_reload(['WebServerGroup-0'])
self.m.VerifyAll()
def test_scaling_policy_down(self):
- t = self.load_template()
- stack = self.parse_stack(t)
+ t = template_format.parse(as_template)
+ stack = parse_stack(t)
# Create initial group, 2 instances
properties = t['Resources']['WebServerGroup']['Properties']
self.m.VerifyAll()
def test_scaling_policy_cooldown_toosoon(self):
- t = self.load_template()
- stack = self.parse_stack(t)
+ t = template_format.parse(as_template)
+ stack = parse_stack(t)
# Create initial group
self._stub_lb_reload(['WebServerGroup-0'])
self.m.VerifyAll()
def test_scaling_policy_cooldown_ok(self):
- t = self.load_template()
- stack = self.parse_stack(t)
+ t = template_format.parse(as_template)
+ stack = parse_stack(t)
# Create initial group
self._stub_lb_reload(['WebServerGroup-0'])
self.m.VerifyAll()
def test_scaling_policy_cooldown_zero(self):
- t = self.load_template()
- stack = self.parse_stack(t)
+ t = template_format.parse(as_template)
+ stack = parse_stack(t)
# Create initial group
self._stub_lb_reload(['WebServerGroup-0'])
self.m.VerifyAll()
def test_scaling_policy_cooldown_none(self):
- t = self.load_template()
- stack = self.parse_stack(t)
+ t = template_format.parse(as_template)
+ stack = parse_stack(t)
# Create initial group
self._stub_lb_reload(['WebServerGroup-0'])
# under the License.
-import os
-
import mox
-from heat.common import context
from heat.common import exception
from heat.common import template_format
-from heat.engine import parser
from heat.engine import scheduler
from heat.engine.resources import dbinstance as dbi
from heat.tests.common import HeatTestCase
from heat.tests.utils import setup_dummy_db
+from heat.tests.utils import parse_stack
+
+
+rds_template = '''
+{
+ "AWSTemplateFormatVersion" : "2010-09-09",
+ "Description" : "RDS Test",
+ "Parameters" : {
+ "KeyName" : {
+ "Description" : "KeyName",
+ "Type" : "String",
+ "Default" : "test"
+ }
+ },
+ "Resources" : {
+ "DatabaseServer": {
+ "Type": "AWS::RDS::DBInstance",
+ "Properties": {
+ "DBName" : "wordpress",
+ "Engine" : "MySQL",
+ "MasterUsername" : "admin",
+ "DBInstanceClass" : "db.m1.small",
+ "DBSecurityGroups" : [],
+ "AllocatedStorage" : "5",
+ "MasterUserPassword": "admin"
+ }
+ }
+ }
+}
+'''
class DBInstanceTest(HeatTestCase):
self.m.StubOutWithMock(dbi.DBInstance, 'create_with_template')
self.m.StubOutWithMock(dbi.DBInstance, 'nested')
- def load_template(self):
- self.path = os.path.dirname(os.path.realpath(__file__)).\
- replace('heat/tests', 'templates')
- f = open("%s/WordPress_With_RDS.template" % self.path)
- t = template_format.parse(f.read())
- f.close()
- return t
-
- def parse_stack(self, t):
- ctx = context.RequestContext.from_dict({
- 'tenant': 'test_tenant',
- 'tenant_id': '1234abcd',
- 'username': 'test_username',
- 'password': 'password',
- 'auth_url': 'http://localhost:5000/v2.0'})
- template = parser.Template(t)
- params = parser.Parameters('test_stack', template, {'KeyName': 'test'})
- stack = parser.Stack(ctx, 'test_stack', template, params)
-
- return stack
-
def create_dbinstance(self, t, stack, resource_name):
resource = dbi.DBInstance(resource_name,
t['Resources'][resource_name],
class FakeNested:
resources = {'DatabaseInstance': FakeDatabaseInstance()}
- params = {
- 'AllocatedStorage': u'5',
- 'DBInstanceClass': u'db.m1.small',
- 'DBName': u'wordpress',
- 'DBSecurityGroups': '',
- 'KeyName': 'test',
- 'MasterUserPassword': u'admin',
- 'MasterUsername': u'admin',
- 'Port': '3306'
- }
+ params = {'DBSecurityGroups': '',
+ 'MasterUsername': u'admin',
+ 'MasterUserPassword': u'admin',
+ 'DBName': u'wordpress',
+ 'KeyName': u'test',
+ 'AllocatedStorage': u'5',
+ 'DBInstanceClass': u'db.m1.small',
+ 'Port': '3306'}
dbi.DBInstance.create_with_template(mox.IgnoreArg(),
params).AndReturn(None)
dbi.DBInstance.nested().MultipleTimes().AndReturn(fn)
self.m.ReplayAll()
- t = self.load_template()
- s = self.parse_stack(t)
+ t = template_format.parse(rds_template)
+ s = parse_stack(t)
resource = self.create_dbinstance(t, s, 'DatabaseServer')
self.assertEqual('0.0.0.0', resource.FnGetAtt('Endpoint.Address'))
# under the License.
-import os
-
-from heat.common import context
from heat.common import template_format
from heat.engine.resources import eip
-from heat.engine import parser
from heat.engine import scheduler
from heat.tests.common import HeatTestCase
from heat.tests.v1_1 import fakes
from heat.tests.utils import setup_dummy_db
+from heat.tests.utils import parse_stack
+
+
+eip_template = '''
+{
+ "AWSTemplateFormatVersion" : "2010-09-09",
+ "Description" : "EIP Test",
+ "Parameters" : {},
+ "Resources" : {
+ "IPAddress" : {
+ "Type" : "AWS::EC2::EIP"
+ },
+ "IPAssoc" : {
+ "Type" : "AWS::EC2::EIPAssociation",
+ "Properties" : {
+ "InstanceId" : { "Ref" : "WebServer" },
+ "EIP" : { "Ref" : "IPAddress" }
+ }
+ },
+ "WebServer": {
+ "Type": "AWS::EC2::Instance",
+ }
+ }
+}
+'''
class EIPTest(HeatTestCase):
self.m.StubOutWithMock(self.fc.servers, 'get')
setup_dummy_db()
- def load_template(self):
- self.path = os.path.dirname(os.path.realpath(__file__)).\
- replace('heat/tests', 'templates')
- f = open("%s/WordPress_Single_Instance_With_EIP.template" % self.path)
- t = template_format.parse(f.read())
- f.close()
- return t
-
- def parse_stack(self, t):
- ctx = context.RequestContext.from_dict({
- 'tenant': 'test_tenant',
- 'username': 'test_username',
- 'password': 'password',
- 'auth_url': 'http://localhost:5000/v2.0'})
- template = parser.Template(t)
- params = parser.Parameters('test_stack', template, {'KeyName': 'test'})
- stack = parser.Stack(ctx, 'test_stack', template, params)
-
- return stack
-
def create_eip(self, t, stack, resource_name):
resource = eip.ElasticIp(resource_name,
t['Resources'][resource_name],
self.m.ReplayAll()
- t = self.load_template()
- stack = self.parse_stack(t)
+ t = template_format.parse(eip_template)
+ stack = parse_stack(t)
+
resource = self.create_eip(t, stack, 'IPAddress')
try:
self.m.ReplayAll()
- t = self.load_template()
- stack = self.parse_stack(t)
+ t = template_format.parse(eip_template)
+ stack = parse_stack(t)
resource = self.create_eip(t, stack, 'IPAddress')
association = self.create_association(t, stack, 'IPAssoc')
# under the License.
-import os
import json
import mox
from heat.tests.utils import setup_dummy_db
-tests_dir = os.path.dirname(os.path.realpath(__file__))
-templates_dir = os.path.normpath(os.path.join(tests_dir,
- os.path.pardir, os.path.pardir,
- 'templates'))
+wp_template = '''
+{
+ "AWSTemplateFormatVersion" : "2010-09-09",
+ "Description" : "WordPress",
+ "Parameters" : {
+ "KeyName" : {
+ "Description" : "KeyName",
+ "Type" : "String",
+ "Default" : "test"
+ }
+ },
+ "Resources" : {
+ "WebServer": {
+ "Type": "AWS::EC2::Instance",
+ "Properties": {
+ "ImageId" : "F17-x86_64-gold",
+ "InstanceType" : "m1.large",
+ "KeyName" : "test",
+ "UserData" : "wordpress"
+ }
+ }
+ }
+}
+'''
def create_context(mocks, user='stacks_test_user',
def get_wordpress_stack(stack_name, ctx):
- tmpl_path = os.path.join(templates_dir,
- 'WordPress_Single_Instance_gold.template')
- with open(tmpl_path) as f:
- t = template_format.parse(f.read())
-
+ t = template_format.parse(wp_template)
template = parser.Template(t)
parameters = parser.Parameters(stack_name, template,
{'KeyName': 'test'})
stack = parser.Stack(ctx, stack_name, template, parameters)
-
return stack
# under the License.
-import os
import copy
import mox
from heat.tests.utils import setup_dummy_db
+wp_template = '''
+{
+ "AWSTemplateFormatVersion" : "2010-09-09",
+ "Description" : "WordPress",
+ "Parameters" : {
+ "KeyName" : {
+ "Description" : "KeyName",
+ "Type" : "String",
+ "Default" : "test"
+ }
+ },
+ "Resources" : {
+ "WebServer": {
+ "Type": "AWS::EC2::Instance",
+ "Properties": {
+ "ImageId" : "F17-x86_64-gold",
+ "InstanceType" : "m1.large",
+ "KeyName" : "test",
+ "UserData" : "wordpress"
+ }
+ }
+ }
+}
+'''
+
+
class instancesTest(HeatTestCase):
def setUp(self):
super(instancesTest, self).setUp()
self.fc = fakes.FakeClient()
- self.path = os.path.dirname(os.path.realpath(__file__)).\
- replace('heat/tests', 'templates')
setup_dummy_db()
def _create_test_instance(self, return_server, name):
- f = open("%s/WordPress_Single_Instance_gold.template" % self.path)
- t = template_format.parse(f.read())
- f.close()
-
stack_name = '%s_stack' % name
+ t = template_format.parse(wp_template)
template = parser.Template(t)
params = parser.Parameters(stack_name, template, {'KeyName': 'test'})
stack = parser.Stack(None, stack_name, template, params,
# under the License.
import copy
-import os
import eventlet
import mox
-from heat.tests.v1_1 import fakes
-from heat.common import context
from heat.common import exception
from heat.common import template_format
from heat.engine.resources import autoscaling as asc
from heat.engine.resources import instance
-from heat.engine.resources import loadbalancer
-from heat.engine import clients
-from heat.engine import parser
from heat.engine import scheduler
from heat.tests.common import HeatTestCase
from heat.tests.utils import setup_dummy_db
+from heat.tests.utils import parse_stack
+
+ig_template = '''
+{
+ "AWSTemplateFormatVersion" : "2010-09-09",
+ "Description" : "Template to create multiple instances.",
+ "Parameters" : {},
+ "Resources" : {
+ "JobServerGroup" : {
+ "Type" : "OS::Heat::InstanceGroup",
+ "Properties" : {
+ "LaunchConfigurationName" : { "Ref" : "JobServerConfig" },
+ "Size" : "1",
+ "AvailabilityZones" : ["nova"]
+ }
+ },
+
+ "JobServerConfig" : {
+ "Type" : "AWS::AutoScaling::LaunchConfiguration",
+ "Properties": {
+ "ImageId" : "foo",
+ "InstanceType" : "m1.large",
+ "KeyName" : "test",
+ "UserData" : "jsconfig data"
+ }
+ }
+ }
+}
+'''
class InstanceGroupTest(HeatTestCase):
def setUp(self):
super(InstanceGroupTest, self).setUp()
- self.fc = fakes.FakeClient()
- self.m.StubOutWithMock(loadbalancer.LoadBalancer, 'reload')
setup_dummy_db()
- self.m.StubOutWithMock(clients.OpenStackClients, 'nova')
- clients.OpenStackClients.nova().MultipleTimes().AndReturn(self.fc)
-
- def load_template(self):
- self.path = os.path.dirname(os.path.realpath(__file__)).\
- replace('heat/tests', 'templates')
- f = open("%s/InstanceGroup.template" % self.path)
- t = template_format.parse(f.read())
- f.close()
- return t
-
- def parse_stack(self, t):
- self.m.ReplayAll()
- ctx = context.RequestContext.from_dict({
- 'tenant': 'test_tenant',
- 'username': 'test_username',
- 'password': 'password',
- 'auth_url': 'http://localhost:5000/v2.0'})
- template = parser.Template(t)
- params = parser.Parameters('test_stack', template, {'KeyName': 'test'})
- stack = parser.Stack(ctx, 'test_stack', template, params)
-
- return stack
def _stub_create(self, num):
self.m.StubOutWithMock(eventlet, 'sleep')
def test_instance_group(self):
- t = self.load_template()
- stack = self.parse_stack(t)
+ t = template_format.parse(ig_template)
+ stack = parse_stack(t)
# start with min then delete
self._stub_create(1)
def test_missing_image(self):
- t = self.load_template()
- stack = self.parse_stack(t)
+ t = template_format.parse(ig_template)
+ stack = parse_stack(t)
resource = asc.InstanceGroup('JobServerGroup',
t['Resources']['JobServerGroup'],
self.m.VerifyAll()
def test_update_size(self):
- t = self.load_template()
+ t = template_format.parse(ig_template)
properties = t['Resources']['JobServerGroup']['Properties']
properties['Size'] = '2'
- stack = self.parse_stack(t)
+ stack = parse_stack(t)
self._stub_create(2)
self.m.ReplayAll()
import mox
import re
-import os
from oslo.config import cfg
from heat.common import exception
from heat.common import config
-from heat.common import context
from heat.common import template_format
from heat.engine import clients
-from heat.engine import parser
from heat.engine import scheduler
from heat.engine.resources import instance
from heat.engine.resources import user
from heat.engine.resource import Metadata
from heat.tests.common import HeatTestCase
from heat.tests.utils import setup_dummy_db
+from heat.tests.utils import parse_stack
from heat.tests.v1_1 import fakes
from heat.tests import fakes as test_fakes
-def create_context(mocks, user='lb_test_user',
- tenant='test_tenant', ctx=None):
- ctx = ctx or context.get_admin_context()
- mocks.StubOutWithMock(ctx, 'username')
- mocks.StubOutWithMock(ctx, 'tenant_id')
- ctx.username = user
- ctx.tenant_id = tenant
- return ctx
+lb_template = '''
+{
+ "AWSTemplateFormatVersion": "2010-09-09",
+ "Description": "LB Template",
+ "Parameters" : {
+ "KeyName" : {
+ "Description" : "KeyName",
+ "Type" : "String",
+ "Default" : "test"
+ }
+ },
+ "Resources": {
+ "WikiServerOne": {
+ "Type": "AWS::EC2::Instance",
+ "Properties": {
+ "ImageId": "F17-x86_64-gold",
+ "InstanceType" : "m1.large",
+ "KeyName" : "test",
+ "UserData" : "some data"
+ }
+ },
+ "LoadBalancer" : {
+ "Type" : "AWS::ElasticLoadBalancing::LoadBalancer",
+ "Properties" : {
+ "AvailabilityZones" : ["nova"],
+ "Instances" : [{"Ref": "WikiServerOne"}],
+ "Listeners" : [ {
+ "LoadBalancerPort" : "80",
+ "InstancePort" : "80",
+ "Protocol" : "HTTP"
+ }]
+ }
+ }
+ }
+}
+'''
class LoadBalancerTest(HeatTestCase):
'http://127.0.0.1:8000/v1/waitcondition')
setup_dummy_db()
- def load_template(self):
- self.path = os.path.dirname(os.path.realpath(__file__)).\
- replace('heat/tests', 'templates')
- f = open("%s/WordPress_With_LB.template" % self.path)
- t = template_format.parse(f.read())
- f.close()
- return t
-
- def parse_stack(self, t):
- template = parser.Template(t)
- params = parser.Parameters('test_stack', template, {'KeyName': 'test'})
- stack = parser.Stack(create_context(self.m), 'test_stack', template,
- params, stack_id=None, disable_rollback=True)
- stack.store()
-
- return stack
-
def create_loadbalancer(self, t, stack, resource_name):
resource = lb.LoadBalancer(resource_name,
t['Resources'][resource_name],
def test_loadbalancer(self):
self.m.StubOutWithMock(user.User, 'keystone')
- user.User.keystone().MultipleTimes().AndReturn(self.fkc)
+ user.User.keystone().AndReturn(self.fkc)
self.m.StubOutWithMock(user.AccessKey, 'keystone')
- user.AccessKey.keystone().MultipleTimes().AndReturn(self.fkc)
+ user.AccessKey.keystone().AndReturn(self.fkc)
self.m.StubOutWithMock(wc.WaitConditionHandle, 'keystone')
wc.WaitConditionHandle.keystone().MultipleTimes().AndReturn(self.fkc)
clients.OpenStackClients.nova(
"compute").MultipleTimes().AndReturn(self.fc)
- clients.OpenStackClients.nova().MultipleTimes().AndReturn(self.fc)
self.fc.servers.create(
flavor=2, image=745, key_name='test',
meta=None, nics=None, name=u'test_stack.LoadBalancer.LB_instance',
security_groups=None, availability_zone=None).AndReturn(
self.fc.servers.list()[1])
Metadata.__set__(mox.IgnoreArg(),
- mox.IgnoreArg()).MultipleTimes().AndReturn(None)
+ mox.IgnoreArg()).AndReturn(None)
self.m.StubOutWithMock(wc.WaitConditionHandle, 'get_status')
wc.WaitConditionHandle.get_status().AndReturn(['SUCCESS'])
self.m.ReplayAll()
- t = self.load_template()
- s = self.parse_stack(t)
+ t = template_format.parse(lb_template)
+ s = parse_stack(t)
+ s.store()
resource = self.create_loadbalancer(t, s, 'LoadBalancer')
# under the License.
-import os
-
from oslo.config import cfg
from heat.common import config
-from heat.common import context
from heat.common import exception
from heat.common import template_format
-from heat.engine import parser
from heat.engine import scheduler
from heat.engine.resources import user
from heat.tests.common import HeatTestCase
from heat.tests import fakes
from heat.tests.utils import setup_dummy_db
+from heat.tests.utils import parse_stack
import keystoneclient.exceptions
+user_template = '''
+{
+ "AWSTemplateFormatVersion" : "2010-09-09",
+ "Description" : "Just a User",
+ "Parameters" : {},
+ "Resources" : {
+ "CfnUser" : {
+ "Type" : "AWS::IAM::User"
+ }
+ }
+}
+'''
+
+user_accesskey_template = '''
+{
+ "AWSTemplateFormatVersion" : "2010-09-09",
+ "Description" : "Just a User",
+ "Parameters" : {},
+ "Resources" : {
+ "CfnUser" : {
+ "Type" : "AWS::IAM::User"
+ },
+
+ "HostKeys" : {
+ "Type" : "AWS::IAM::AccessKey",
+ "Properties" : {
+ "UserName" : {"Ref": "CfnUser"}
+ }
+ }
+ }
+}
+'''
+
+
+user_policy_template = '''
+{
+ "AWSTemplateFormatVersion" : "2010-09-09",
+ "Description" : "Just a User",
+ "Parameters" : {},
+ "Resources" : {
+ "CfnUser" : {
+ "Type" : "AWS::IAM::User",
+ "Properties" : {
+ "Policies" : [ { "Ref": "WebServerAccessPolicy"} ]
+ }
+ },
+ "WebServerAccessPolicy" : {
+ "Type" : "OS::Heat::AccessPolicy",
+ "Properties" : {
+ "AllowedResources" : [ "WikiDatabase" ]
+ }
+ },
+ "WikiDatabase" : {
+ "Type" : "AWS::EC2::Instance",
+ }
+ }
+}
+'''
+
class UserPolicyTestCase(HeatTestCase):
def setUp(self):
class UserTest(UserPolicyTestCase):
- def load_template(self, template_name='Rails_Single_Instance.template'):
- self.path = os.path.dirname(os.path.realpath(__file__)).\
- replace('heat/tests', 'templates')
- f = open("%s/%s" % (self.path, template_name))
- t = template_format.parse(f.read())
- f.close()
- return t
-
- def parse_stack(self, t):
- ctx = context.RequestContext.from_dict({
- 'tenant_id': 'test_tenant',
- 'username': 'test_username',
- 'password': 'password',
- 'auth_url': 'http://localhost:5000/v2.0'})
- template = parser.Template(t)
- params = parser.Parameters('test_stack',
- template,
- {'KeyName': 'test',
- 'DBRootPassword': 'test',
- 'DBUsername': 'test',
- 'DBPassword': 'test'})
- stack = parser.Stack(ctx, 'test_stack', template, params)
-
- return stack
-
def create_user(self, t, stack, resource_name):
resource = user.User(resource_name,
t['Resources'][resource_name],
self.m.ReplayAll()
- t = self.load_template()
- stack = self.parse_stack(t)
+ t = template_format.parse(user_template)
+ stack = parse_stack(t)
resource = self.create_user(t, stack, 'CfnUser')
self.assertEqual(self.fc.user_id, resource.resource_id)
self.m.ReplayAll()
- tmpl = 'WordPress_Single_Instance_With_HA_AccessPolicy.template'
- t = self.load_template(template_name=tmpl)
- stack = self.parse_stack(t)
+ t = template_format.parse(user_policy_template)
+ stack = parse_stack(t)
resource = self.create_user(t, stack, 'CfnUser')
self.assertEqual(self.fc.user_id, resource.resource_id)
def test_user_create_bad_policies(self):
self.m.ReplayAll()
- tmpl = 'WordPress_Single_Instance_With_HA_AccessPolicy.template'
- t = self.load_template(template_name=tmpl)
+ t = template_format.parse(user_policy_template)
t['Resources']['CfnUser']['Properties']['Policies'] = ['NoExistBad']
- stack = self.parse_stack(t)
+ stack = parse_stack(t)
resource_name = 'CfnUser'
resource = user.User(resource_name,
t['Resources'][resource_name],
self.m.ReplayAll()
- tmpl = 'WordPress_Single_Instance_With_HA_AccessPolicy.template'
- t = self.load_template(template_name=tmpl)
- stack = self.parse_stack(t)
+ t = template_format.parse(user_policy_template)
+ stack = parse_stack(t)
resource = self.create_user(t, stack, 'CfnUser')
self.assertEqual(self.fc.user_id, resource.resource_id)
self.m.ReplayAll()
- tmpl = 'WordPress_Single_Instance_With_HA_AccessPolicy.template'
- t = self.load_template(template_name=tmpl)
+ t = template_format.parse(user_policy_template)
t['Resources']['CfnUser']['Properties']['Policies'] = [
'WebServerAccessPolicy', {'an_ignored': 'policy'}]
- stack = self.parse_stack(t)
+ stack = parse_stack(t)
resource = self.create_user(t, stack, 'CfnUser')
self.assertEqual(self.fc.user_id, resource.resource_id)
class AccessKeyTest(UserPolicyTestCase):
- def load_template(self):
- self.path = os.path.dirname(os.path.realpath(__file__)).\
- replace('heat/tests', 'templates')
- f = open("%s/Rails_Single_Instance.template" % self.path)
- t = template_format.parse(f.read())
- f.close()
- return t
-
- def parse_stack(self, t):
- ctx = context.RequestContext.from_dict({
- 'tenant_id': 'test_tenant',
- 'username': 'test_username',
- 'password': 'password',
- 'auth_url': 'http://localhost:5000/v2.0'})
- template = parser.Template(t)
- params = parser.Parameters('test_stack',
- template,
- {'KeyName': 'test',
- 'DBRootPassword': 'test',
- 'DBUsername': 'test',
- 'DBPassword': 'test'})
- stack = parser.Stack(ctx, 'test_stack', template, params)
-
- return stack
-
def create_access_key(self, t, stack, resource_name):
resource = user.AccessKey(resource_name,
t['Resources'][resource_name],
self.m.ReplayAll()
- t = self.load_template()
+ t = template_format.parse(user_accesskey_template)
# Override the Ref for UserName with a hard-coded name,
# so we don't need to create the User resource
t['Resources']['HostKeys']['Properties']['UserName'] =\
'test_stack.CfnUser'
- stack = self.parse_stack(t)
+ stack = parse_stack(t)
stack.resources['CfnUser'].resource_id = self.fc.user_id
stack.resources['CfnUser'].state = 'CREATE_COMPLETE'
def test_access_key_no_user(self):
self.m.ReplayAll()
- t = self.load_template()
+ t = template_format.parse(user_accesskey_template)
# Set the resource properties UserName to an unknown user
t['Resources']['HostKeys']['Properties']['UserName'] =\
'test_stack.NoExist'
- stack = self.parse_stack(t)
+ stack = parse_stack(t)
stack.resources['CfnUser'].resource_id = self.fc.user_id
resource = user.AccessKey('HostKeys',
class AccessPolicyTest(UserPolicyTestCase):
- def load_template(self):
- template_name =\
- 'WordPress_Single_Instance_With_HA_AccessPolicy.template'
- self.path = os.path.dirname(os.path.realpath(__file__)).\
- replace('heat/tests', 'templates')
- f = open("%s/%s" % (self.path, template_name))
- t = template_format.parse(f.read())
- f.close()
- return t
-
- def parse_stack(self, t):
- ctx = context.RequestContext.from_dict({
- 'tenant_id': 'test_tenant',
- 'username': 'test_username',
- 'password': 'password',
- 'auth_url': 'http://localhost:5000/v2.0'})
- template = parser.Template(t)
- params = parser.Parameters('test_stack',
- template,
- {'KeyName': 'test',
- 'DBRootPassword': 'test',
- 'DBUsername': 'test',
- 'DBPassword': 'test'})
- stack = parser.Stack(ctx, 'test_stack', template, params)
-
- return stack
-
def test_accesspolicy_create_ok(self):
- t = self.load_template()
- stack = self.parse_stack(t)
+ t = template_format.parse(user_policy_template)
+ stack = parse_stack(t)
resource_name = 'WebServerAccessPolicy'
resource = user.AccessPolicy(resource_name,
self.assertEqual(user.User.CREATE_COMPLETE, resource.state)
def test_accesspolicy_create_ok_empty(self):
- t = self.load_template()
+ t = template_format.parse(user_policy_template)
resource_name = 'WebServerAccessPolicy'
t['Resources'][resource_name]['Properties']['AllowedResources'] = []
- stack = self.parse_stack(t)
+ stack = parse_stack(t)
resource = user.AccessPolicy(resource_name,
t['Resources'][resource_name],
self.assertEqual(user.User.CREATE_COMPLETE, resource.state)
def test_accesspolicy_create_err_notfound(self):
- t = self.load_template()
+ t = template_format.parse(user_policy_template)
resource_name = 'WebServerAccessPolicy'
t['Resources'][resource_name]['Properties']['AllowedResources'] = [
'NoExistResource']
- stack = self.parse_stack(t)
+ stack = parse_stack(t)
resource = user.AccessPolicy(resource_name,
t['Resources'][resource_name],
self.assertRaises(exception.ResourceNotFound, resource.handle_create)
def test_accesspolicy_update(self):
- t = self.load_template()
+ t = template_format.parse(user_policy_template)
resource_name = 'WebServerAccessPolicy'
- stack = self.parse_stack(t)
+ stack = parse_stack(t)
resource = user.AccessPolicy(resource_name,
t['Resources'][resource_name],
resource.handle_update({}))
def test_accesspolicy_access_allowed(self):
- t = self.load_template()
+ t = template_format.parse(user_policy_template)
resource_name = 'WebServerAccessPolicy'
- stack = self.parse_stack(t)
+ stack = parse_stack(t)
resource = user.AccessPolicy(resource_name,
t['Resources'][resource_name],
Code Review / openstack-build / heat-build.git / commitdiff