azure/aria/aria-extension-cloudify/src/aria/tests/instantiation/test_configuration.py

   1 # Licensed to the Apache Software Foundation (ASF) under one or more
   2 # contributor license agreements.  See the NOTICE file distributed with
   3 # this work for additional information regarding copyright ownership.
   4 # The ASF licenses this file to You under the Apache License, Version 2.0
   5 # (the "License"); you may not use this file except in compliance with
   6 # the License.  You may obtain a copy of the License at
   7 #
   8 #     http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 # Unless required by applicable law or agreed to in writing, software
  11 # distributed under the License is distributed on an "AS IS" BASIS,
  12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 # See the License for the specific language governing permissions and
  14 # limitations under the License.
  15
  16 import pytest
  17
  18 from tests.parser.service_templates import consume_literal
  19 from aria.modeling.utils import parameters_as_values
  20
  21
  22 TEMPLATE = """
  23 tosca_definitions_version: tosca_simple_yaml_1_0
  24
  25 interface_types:
  26   MyInterface:
  27     derived_from: tosca.interfaces.Root
  28     inputs:
  29       interface_string:
  30         type: string
  31         default: value1
  32       interface_integer:
  33         type: integer
  34         default: 1
  35     operation:
  36       implementation: operation.sh
  37       inputs:
  38         operation_string:
  39           type: string
  40           default: value2
  41         operation_integer:
  42           type: integer
  43           default: 2
  44         interface_integer: # will override interface input
  45           type: integer
  46           default: 3
  47
  48 node_types:
  49   LocalNode:
  50     derived_from: tosca.nodes.Root
  51     interfaces:
  52       MyInterface:
  53         type: MyInterface
  54
  55   RemoteNode:
  56     derived_from: tosca.nodes.Compute
  57     interfaces:
  58       MyInterface:
  59         type: MyInterface
  60
  61 topology_template:
  62   node_templates:
  63     local_node:
  64       type: LocalNode
  65
  66     remote_node:
  67       type: RemoteNode
  68 """
  69
  70
  71 BROKEN_TEMPLATE = """
  72 tosca_definitions_version: tosca_simple_yaml_1_0
  73
  74 interface_types:
  75   MyInterface:
  76     derived_from: tosca.interfaces.Root
  77     inputs:
  78       ctx: # reserved name
  79         type: string
  80         default: value1
  81       interface_integer:
  82         type: integer
  83         default: 1
  84     operation:
  85       implementation: operation.sh
  86       inputs:
  87         operation_string:
  88           type: string
  89           default: value2
  90         toolbelt: # reserved name
  91           type: integer
  92           default: 2
  93
  94 node_types:
  95   LocalNode:
  96     derived_from: tosca.nodes.Root
  97     interfaces:
  98       MyInterface:
  99         type: MyInterface
 100
 101 topology_template:
 102   node_templates:
 103     local_node:
 104       type: LocalNode
 105 """
 106
 107
 108 @pytest.fixture
 109 def service():
 110     context, _ = consume_literal(TEMPLATE)
 111     yield context.modeling.instance
 112
 113
 114 @pytest.fixture
 115 def broken_service_issues():
 116     context, _ = consume_literal(BROKEN_TEMPLATE, no_issues=False)
 117     yield context.validation.issues
 118
 119
 120 def test_local(service):
 121     interface = service.nodes['local_node_1'].interfaces['MyInterface']
 122     operation = interface.operations['operation']
 123     assert parameters_as_values(interface.inputs) == {
 124         'interface_string': 'value1',
 125         'interface_integer': 1
 126     }
 127     assert parameters_as_values(operation.inputs) == {
 128         'operation_string': 'value2',
 129         'operation_integer': 2,
 130         'interface_integer': 3
 131     }
 132     assert parameters_as_values(operation.arguments) == {
 133         'process': {},
 134         'script_path': 'operation.sh',
 135         'interface_string': 'value1',
 136         'interface_integer': 3,
 137         'operation_string': 'value2',
 138         'operation_integer': 2
 139     }
 140
 141
 142 def test_remote(service):
 143     interface = service.nodes['remote_node_1'].interfaces['MyInterface']
 144     operation = interface.operations['operation']
 145     assert parameters_as_values(interface.inputs) == {
 146         'interface_string': 'value1',
 147         'interface_integer': 1
 148     }
 149     assert parameters_as_values(operation.inputs) == {
 150         'operation_string': 'value2',
 151         'operation_integer': 2,
 152         'interface_integer': 3
 153     }
 154     assert parameters_as_values(operation.arguments) == {
 155         'process': {},
 156         'use_sudo': False,
 157         'fabric_env': {'user': '', 'password': '', 'key': None, 'key_filename': None},
 158         'script_path': 'operation.sh',
 159         'hide_output': [],
 160         'interface_string': 'value1',
 161         'interface_integer': 3,
 162         'operation_string': 'value2',
 163         'operation_integer': 2
 164     }
 165
 166
 167 def test_reserved_arguments(broken_service_issues):
 168     assert len(broken_service_issues) == 1
 169     message = broken_service_issues[0].message
 170     assert message.startswith('using reserved arguments in operation "operation":')
 171     assert '"ctx"' in message
 172     assert '"toolbelt"' in message