366 lines
12 KiB
Python
366 lines
12 KiB
Python
# 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
|
|
|
|
|
|
cfn_template = '''
|
|
AWSTemplateFormatVersion: 2010-09-09
|
|
Parameters:
|
|
env_type:
|
|
Default: test
|
|
Type: String
|
|
AllowedValues: [prod, test]
|
|
zone:
|
|
Type: String
|
|
Default: beijing
|
|
Conditions:
|
|
Prod: {"Fn::Equals" : [{Ref: env_type}, "prod"]}
|
|
Test:
|
|
Fn::Not:
|
|
- Fn::Equals:
|
|
- Ref: env_type
|
|
- prod
|
|
Beijing_Prod:
|
|
Fn::And:
|
|
- Fn::Equals:
|
|
- Ref: env_type
|
|
- prod
|
|
- Fn::Equals:
|
|
- Ref: zone
|
|
- beijing
|
|
Fujian_Zone:
|
|
Fn::Or:
|
|
- Fn::Equals:
|
|
- Ref: zone
|
|
- fuzhou
|
|
- Fn::Equals:
|
|
- Ref: zone
|
|
- xiamen
|
|
Resources:
|
|
test_res:
|
|
Type: OS::Heat::TestResource
|
|
Properties:
|
|
value: {"Fn::If": ["Prod", "env_is_prod", "env_is_test"]}
|
|
prod_res:
|
|
Type: OS::Heat::TestResource
|
|
Properties:
|
|
value: prod_res
|
|
Condition: Prod
|
|
test_res1:
|
|
Type: OS::Heat::TestResource
|
|
Properties:
|
|
value: just in test env
|
|
Condition: Test
|
|
beijing_prod_res:
|
|
Type: OS::Heat::TestResource
|
|
Properties:
|
|
value: beijing_prod_res
|
|
Condition: Beijing_Prod
|
|
fujian_res:
|
|
Type: OS::Heat::TestResource
|
|
Condition: Fujian_Zone
|
|
Properties:
|
|
value: fujian_res
|
|
Outputs:
|
|
res_value:
|
|
Value: {"Fn::GetAtt": [prod_res, output]}
|
|
Condition: Prod
|
|
test_res_value:
|
|
Value: {"Fn::GetAtt": [test_res, output]}
|
|
prod_resource:
|
|
Value: {"Fn::If": [Prod, {Ref: prod_res}, 'no_prod_res']}
|
|
test_res1_value:
|
|
Value: {"Fn::If": [Test, {"Fn::GetAtt": [test_res1, output]},
|
|
'no_test_res1']}
|
|
beijing_prod_res:
|
|
Value: {"Fn::If": [Beijing_Prod, {Ref: beijing_prod_res}, 'no_prod_res']}
|
|
'''
|
|
|
|
hot_template = '''
|
|
heat_template_version: 2016-10-14
|
|
parameters:
|
|
env_type:
|
|
default: test
|
|
type: string
|
|
constraints:
|
|
- allowed_values: [prod, test]
|
|
zone:
|
|
type: string
|
|
default: beijing
|
|
conditions:
|
|
prod: {equals : [{get_param: env_type}, "prod"]}
|
|
test:
|
|
not:
|
|
equals:
|
|
- get_param: env_type
|
|
- prod
|
|
beijing_prod:
|
|
and:
|
|
- equals:
|
|
- get_param: zone
|
|
- beijing
|
|
- equals:
|
|
- get_param: env_type
|
|
- prod
|
|
fujian_zone:
|
|
or:
|
|
- equals:
|
|
- get_param: zone
|
|
- fuzhou
|
|
- equals:
|
|
- get_param: zone
|
|
- xiamen
|
|
resources:
|
|
test_res:
|
|
type: OS::Heat::TestResource
|
|
properties:
|
|
value: {if: ["prod", "env_is_prod", "env_is_test"]}
|
|
prod_res:
|
|
type: OS::Heat::TestResource
|
|
properties:
|
|
value: prod_res
|
|
condition: prod
|
|
test_res1:
|
|
type: OS::Heat::TestResource
|
|
properties:
|
|
value: just in test env
|
|
condition: test
|
|
beijing_prod_res:
|
|
type: OS::Heat::TestResource
|
|
properties:
|
|
value: beijing_prod_res
|
|
condition: beijing_prod
|
|
fujian_res:
|
|
type: OS::Heat::TestResource
|
|
condition: fujian_zone
|
|
properties:
|
|
value: fujian_res
|
|
outputs:
|
|
res_value:
|
|
value: {get_attr: [prod_res, output]}
|
|
condition: prod
|
|
test_res_value:
|
|
value: {get_attr: [test_res, output]}
|
|
prod_resource:
|
|
value: {if: [prod, {get_resource: prod_res}, 'no_prod_res']}
|
|
test_res1_value:
|
|
value: {if: [test, {get_attr: [test_res1, output]}, 'no_test_res1']}
|
|
beijing_prod_res:
|
|
value: {if: [beijing_prod, {get_resource: beijing_prod_res},
|
|
'no_prod_res']}
|
|
'''
|
|
|
|
|
|
class CreateUpdateResConditionTest(functional_base.FunctionalTestsBase):
|
|
|
|
def setUp(self):
|
|
super(CreateUpdateResConditionTest, self).setUp()
|
|
|
|
def res_assert_for_prod(self, resources, bj_prod=True, fj_zone=False):
|
|
res_names = [res.resource_name for res in resources]
|
|
if bj_prod:
|
|
self.assertEqual(3, len(resources))
|
|
self.assertIn('beijing_prod_res', res_names)
|
|
elif fj_zone:
|
|
self.assertEqual(3, len(resources))
|
|
self.assertIn('fujian_res', res_names)
|
|
self.assertNotIn('beijing_prod_res', res_names)
|
|
else:
|
|
self.assertEqual(2, len(resources))
|
|
self.assertIn('prod_res', res_names)
|
|
self.assertIn('test_res', res_names)
|
|
|
|
def res_assert_for_test(self, resources, fj_zone=False):
|
|
res_names = [res.resource_name for res in resources]
|
|
|
|
if fj_zone:
|
|
self.assertEqual(3, len(resources))
|
|
self.assertIn('fujian_res', res_names)
|
|
else:
|
|
self.assertEqual(2, len(resources))
|
|
self.assertNotIn('fujian_res', res_names)
|
|
|
|
self.assertIn('test_res', res_names)
|
|
self.assertIn('test_res1', res_names)
|
|
self.assertNotIn('prod_res', res_names)
|
|
|
|
def output_assert_for_prod(self, stack_id, bj_prod=True):
|
|
output = self.client.stacks.output_show(stack_id,
|
|
'res_value')['output']
|
|
self.assertEqual('prod_res', output['output_value'])
|
|
|
|
test_res_value = self.client.stacks.output_show(
|
|
stack_id, 'test_res_value')['output']
|
|
self.assertEqual('env_is_prod', test_res_value['output_value'])
|
|
|
|
prod_resource = self.client.stacks.output_show(
|
|
stack_id, 'prod_resource')['output']
|
|
self.assertNotEqual('no_prod_res', prod_resource['output_value'])
|
|
|
|
test_res_output = self.client.stacks.output_show(
|
|
stack_id, 'test_res1_value')['output']
|
|
self.assertEqual('no_test_res1', test_res_output['output_value'])
|
|
|
|
beijing_prod_res = self.client.stacks.output_show(
|
|
stack_id, 'beijing_prod_res')['output']
|
|
if bj_prod:
|
|
self.assertNotEqual('no_prod_res',
|
|
beijing_prod_res['output_value'])
|
|
else:
|
|
self.assertEqual('no_prod_res', beijing_prod_res['output_value'])
|
|
|
|
def output_assert_for_test(self, stack_id):
|
|
output = self.client.stacks.output_show(stack_id,
|
|
'res_value')['output']
|
|
self.assertIsNone(output['output_value'])
|
|
|
|
test_res_value = self.client.stacks.output_show(
|
|
stack_id, 'test_res_value')['output']
|
|
self.assertEqual('env_is_test', test_res_value['output_value'])
|
|
|
|
prod_resource = self.client.stacks.output_show(
|
|
stack_id, 'prod_resource')['output']
|
|
self.assertEqual('no_prod_res', prod_resource['output_value'])
|
|
|
|
test_res_output = self.client.stacks.output_show(
|
|
stack_id, 'test_res1_value')['output']
|
|
self.assertEqual('just in test env',
|
|
test_res_output['output_value'])
|
|
|
|
beijing_prod_res = self.client.stacks.output_show(
|
|
stack_id, 'beijing_prod_res')['output']
|
|
self.assertEqual('no_prod_res', beijing_prod_res['output_value'])
|
|
|
|
def test_stack_create_update_cfn_template_test_to_prod(self):
|
|
stack_identifier = self.stack_create(template=cfn_template)
|
|
resources = self.client.resources.list(stack_identifier)
|
|
self.res_assert_for_test(resources)
|
|
self.output_assert_for_test(stack_identifier)
|
|
|
|
parms = {'zone': 'fuzhou'}
|
|
self.update_stack(stack_identifier,
|
|
template=cfn_template,
|
|
parameters=parms)
|
|
|
|
resources = self.client.resources.list(stack_identifier)
|
|
self.res_assert_for_test(resources, fj_zone=True)
|
|
self.output_assert_for_test(stack_identifier)
|
|
|
|
parms = {'env_type': 'prod'}
|
|
self.update_stack(stack_identifier,
|
|
template=cfn_template,
|
|
parameters=parms)
|
|
|
|
resources = self.client.resources.list(stack_identifier)
|
|
self.res_assert_for_prod(resources)
|
|
self.output_assert_for_prod(stack_identifier)
|
|
|
|
parms = {'env_type': 'prod',
|
|
'zone': 'shanghai'}
|
|
self.update_stack(stack_identifier,
|
|
template=cfn_template,
|
|
parameters=parms)
|
|
|
|
resources = self.client.resources.list(stack_identifier)
|
|
self.res_assert_for_prod(resources, False)
|
|
self.output_assert_for_prod(stack_identifier, False)
|
|
|
|
parms = {'env_type': 'prod',
|
|
'zone': 'xiamen'}
|
|
self.update_stack(stack_identifier,
|
|
template=cfn_template,
|
|
parameters=parms)
|
|
|
|
resources = self.client.resources.list(stack_identifier)
|
|
self.res_assert_for_prod(resources, bj_prod=False, fj_zone=True)
|
|
self.output_assert_for_prod(stack_identifier, False)
|
|
|
|
def test_stack_create_update_cfn_template_prod_to_test(self):
|
|
parms = {'env_type': 'prod'}
|
|
stack_identifier = self.stack_create(template=cfn_template,
|
|
parameters=parms)
|
|
resources = self.client.resources.list(stack_identifier)
|
|
self.res_assert_for_prod(resources)
|
|
self.output_assert_for_prod(stack_identifier)
|
|
|
|
parms = {'zone': 'xiamen',
|
|
'env_type': 'prod'}
|
|
self.update_stack(stack_identifier,
|
|
template=cfn_template,
|
|
parameters=parms)
|
|
|
|
resources = self.client.resources.list(stack_identifier)
|
|
self.res_assert_for_prod(resources, bj_prod=False, fj_zone=True)
|
|
self.output_assert_for_prod(stack_identifier, bj_prod=False)
|
|
|
|
parms = {'env_type': 'test'}
|
|
self.update_stack(stack_identifier,
|
|
template=cfn_template,
|
|
parameters=parms)
|
|
|
|
resources = self.client.resources.list(stack_identifier)
|
|
self.res_assert_for_test(resources)
|
|
self.output_assert_for_test(stack_identifier)
|
|
|
|
parms = {'env_type': 'test',
|
|
'zone': 'fuzhou'}
|
|
self.update_stack(stack_identifier,
|
|
template=cfn_template,
|
|
parameters=parms)
|
|
|
|
resources = self.client.resources.list(stack_identifier)
|
|
self.res_assert_for_test(resources, fj_zone=True)
|
|
self.output_assert_for_test(stack_identifier)
|
|
|
|
def test_stack_create_update_hot_template_test_to_prod(self):
|
|
stack_identifier = self.stack_create(template=hot_template)
|
|
resources = self.client.resources.list(stack_identifier)
|
|
self.res_assert_for_test(resources)
|
|
self.output_assert_for_test(stack_identifier)
|
|
|
|
parms = {'env_type': 'prod'}
|
|
self.update_stack(stack_identifier,
|
|
template=hot_template,
|
|
parameters=parms)
|
|
|
|
resources = self.client.resources.list(stack_identifier)
|
|
self.res_assert_for_prod(resources)
|
|
self.output_assert_for_prod(stack_identifier)
|
|
|
|
parms = {'env_type': 'prod',
|
|
'zone': 'shanghai'}
|
|
self.update_stack(stack_identifier,
|
|
template=hot_template,
|
|
parameters=parms)
|
|
|
|
resources = self.client.resources.list(stack_identifier)
|
|
self.res_assert_for_prod(resources, False)
|
|
self.output_assert_for_prod(stack_identifier, False)
|
|
|
|
def test_stack_create_update_hot_template_prod_to_test(self):
|
|
parms = {'env_type': 'prod'}
|
|
stack_identifier = self.stack_create(template=hot_template,
|
|
parameters=parms)
|
|
resources = self.client.resources.list(stack_identifier)
|
|
self.res_assert_for_prod(resources)
|
|
self.output_assert_for_prod(stack_identifier)
|
|
|
|
parms = {'env_type': 'test'}
|
|
self.update_stack(stack_identifier,
|
|
template=hot_template,
|
|
parameters=parms)
|
|
|
|
resources = self.client.resources.list(stack_identifier)
|
|
self.res_assert_for_test(resources)
|
|
self.output_assert_for_test(stack_identifier)
|