Source code for heat_integrationtests.functional.test_conditions

#    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
  Xian_Zone:
    Fn::Equals:
    - Ref: zone
    - xian
  Xianyang_Zone:
    Fn::Equals:
    - Ref: zone
    - xianyang
  Fujian_Zone:
    Fn::Or:
    - Fn::Equals:
      - Ref: zone
      - fuzhou
    - Fn::Equals:
      - Ref: zone
      - xiamen
  Fujian_Prod:
    Fn::And:
    - Fujian_Zone
    - Prod
  Shannxi_Provice:
    Fn::Or:
    - Xian_Zone
    - Xianyang_Zone
  Not_Shannxi:
    Fn::Not: [Shannxi_Provice]
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
  fujian_prod_res:
    Type: OS::Heat::TestResource
    Condition: Fujian_Prod
    Properties:
      value: fujian_prod_res
  shannxi_res:
    Type: OS::Heat::TestResource
    Condition: Shannxi_Provice
    Properties:
      value: shannxi_res
  not_shannxi_res:
    Type: OS::Heat::TestResource
    Condition: Not_Shannxi
    Properties:
      value: not_shannxi_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
  xian_zone:
    equals:
    - get_param: zone
    - xian
  xianyang_zone:
    equals:
    - get_param: zone
    - xianyang
  fujian_zone:
    or:
    - equals:
      - get_param: zone
      - fuzhou
    - equals:
      - get_param: zone
      - xiamen
  fujian_prod:
    and:
    - fujian_zone
    - prod
  shannxi_provice:
    or:
    - xian_zone
    - xianyang_zone
  not_shannxi:
    not: shannxi_provice
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
  fujian_prod_res:
    type: OS::Heat::TestResource
    condition: fujian_prod
    properties:
      value: fujian_prod_res
  shannxi_res:
    type: OS::Heat::TestResource
    condition: shannxi_provice
    properties:
      value: shannxi_res
  not_shannxi_res:
    type: OS::Heat::TestResource
    condition: not_shannxi
    properties:
      value: not_shannxi_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']}
'''

before_rename_tmpl = '''
heat_template_version: 2016-10-14
parameters:
  env_type:
    default: test
    type: string
conditions:
  cd1: {equals : [{get_param: env_type}, "prod"]}
resources:
  test:
    type: OS::Heat::TestResource
    properties:
      value: {if: [cd1, 'prod', 'test']}
'''

after_rename_tmpl = '''
heat_template_version: 2016-10-14
parameters:
  env_type:
    default: prod
    type: string
conditions:
  cd2: {equals : [{get_param: env_type}, "prod"]}
resources:
  test:
    type: OS::Heat::TestResource
    properties:
      value: {if: [cd2, 'prod', 'test']}
  test2:
    type: OS::Heat::TestResource
    properties:
      value: {if: [cd2, 'prod', 'test']}
'''

fail_rename_tmpl = '''
heat_template_version: 2016-10-14
parameters:
  env_type:
    default: prod
    type: string
conditions:
  cd3: {equals : [{get_param: env_type}, "prod"]}
resources:
  test:
    type: OS::Heat::TestResource
    properties:
      value: {if: [cd3, 'prod', 'test']}
  test2:
    type: OS::Heat::TestResource
    properties:
      value: {if: [cd3, 'prod', 'test']}
  test_fail:
    type: OS::Heat::TestResource
    properties:
      fail: True
    depends_on: [test, test2]
'''

recover_rename_tmpl = '''
heat_template_version: 2016-10-14
parameters:
  env_type:
    default: prod
    type: string
conditions:
  cd3: {equals : [{get_param: env_type}, "prod"]}
resources:
  test2:
    type: OS::Heat::TestResource
    properties:
      value: {if: [cd3, 'prod', 'test']}
  test_fail:
    type: OS::Heat::TestResource
    properties:
      fail: False
    depends_on: [test2]
'''


[docs]class CreateUpdateResConditionTest(functional_base.FunctionalTestsBase):
[docs] def res_assert_for_prod(self, resources, bj_prod=True, fj_zone=False, shannxi_provice=False): res_names = [res.resource_name for res in resources] if bj_prod: self.assertEqual(4, len(resources)) self.assertIn('beijing_prod_res', res_names) self.assertIn('not_shannxi_res', res_names) elif fj_zone: self.assertEqual(5, len(resources)) self.assertIn('fujian_res', res_names) self.assertNotIn('beijing_prod_res', res_names) self.assertIn('not_shannxi_res', res_names) self.assertIn('fujian_prod_res', res_names) elif shannxi_provice: self.assertEqual(3, len(resources)) self.assertIn('shannxi_res', res_names) else: self.assertEqual(3, len(resources)) self.assertIn('not_shannxi_res', res_names) self.assertIn('prod_res', res_names) self.assertIn('test_res', res_names)
[docs] def res_assert_for_test(self, resources, fj_zone=False, shannxi_provice=False): res_names = [res.resource_name for res in resources] if fj_zone: self.assertEqual(4, len(resources)) self.assertIn('fujian_res', res_names) self.assertIn('not_shannxi_res', res_names) elif shannxi_provice: self.assertEqual(3, len(resources)) self.assertNotIn('fujian_res', res_names) self.assertIn('shannxi_res', res_names) else: self.assertEqual(3, len(resources)) self.assertIn('not_shannxi_res', res_names) self.assertIn('test_res', res_names) self.assertIn('test_res1', res_names) self.assertNotIn('prod_res', res_names)
[docs] 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'])
[docs] 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'])
[docs] 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 = {'zone': 'xianyang'} self.update_stack(stack_identifier, template=cfn_template, parameters=parms) resources = self.client.resources.list(stack_identifier) self.res_assert_for_test(resources, shannxi_provice=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) parms = {'env_type': 'prod', 'zone': 'xianyang'} 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=False, shannxi_provice=True) self.output_assert_for_prod(stack_identifier, False)
[docs] 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 = {'zone': 'xianyang', '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=False, shannxi_provice=True) self.output_assert_for_prod(stack_identifier, bj_prod=False) parms = {'zone': 'shanghai', '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=False, shannxi_provice=False) 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) parms = {'env_type': 'test', 'zone': 'xianyang'} 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=False, shannxi_provice=True) self.output_assert_for_test(stack_identifier)
[docs] 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 = {'zone': 'xianyang'} self.update_stack(stack_identifier, template=hot_template, parameters=parms) resources = self.client.resources.list(stack_identifier) self.res_assert_for_test(resources, shannxi_provice=True) 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) parms = {'env_type': 'prod', 'zone': 'xianyang'} self.update_stack(stack_identifier, template=hot_template, parameters=parms) resources = self.client.resources.list(stack_identifier) self.res_assert_for_prod(resources, False, shannxi_provice=True) self.output_assert_for_prod(stack_identifier, False)
[docs] 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': 'prod', 'zone': 'xianyang'} self.update_stack(stack_identifier, template=hot_template, parameters=parms) resources = self.client.resources.list(stack_identifier) self.res_assert_for_prod(resources, False, shannxi_provice=True) self.output_assert_for_prod(stack_identifier, False) 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) parms = {'env_type': 'test', 'zone': 'xianyang'} self.update_stack(stack_identifier, template=hot_template, parameters=parms) resources = self.client.resources.list(stack_identifier) self.res_assert_for_test(resources, fj_zone=False, shannxi_provice=True) self.output_assert_for_test(stack_identifier)
[docs] def test_condition_rename(self): stack_identifier = self.stack_create(template=before_rename_tmpl) self.update_stack(stack_identifier, template=after_rename_tmpl) self.update_stack(stack_identifier, template=fail_rename_tmpl, expected_status='UPDATE_FAILED') self.update_stack(stack_identifier, template=recover_rename_tmpl)