diff options
Diffstat (limited to 'azure/aria/aria-extension-cloudify/src/aria/extensions/aria_extension_tosca/simple_v1_0/modeling/interfaces.py')
| -rw-r--r-- | azure/aria/aria-extension-cloudify/src/aria/extensions/aria_extension_tosca/simple_v1_0/modeling/interfaces.py | 530 |
1 files changed, 530 insertions, 0 deletions
diff --git a/azure/aria/aria-extension-cloudify/src/aria/extensions/aria_extension_tosca/simple_v1_0/modeling/interfaces.py b/azure/aria/aria-extension-cloudify/src/aria/extensions/aria_extension_tosca/simple_v1_0/modeling/interfaces.py new file mode 100644 index 0000000..23a03b7 --- /dev/null +++ b/azure/aria/aria-extension-cloudify/src/aria/extensions/aria_extension_tosca/simple_v1_0/modeling/interfaces.py @@ -0,0 +1,530 @@ +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You 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 aria.utils.collections import (merge, deepcopy_with_locators, OrderedDict) +from aria.parser.presentation import get_locator +from aria.parser.validation import Issue + +from .parameters import (coerce_parameter_value, convert_parameter_definitions_to_values) + + +# +# InterfaceType +# + +def get_inherited_operations(context, presentation): + """ + Returns our operation definitions added on top of those of our parent, if we have one + (recursively). + + Allows overriding all aspects of parent operations except input data types. + """ + + # Get operations from parent + parent = presentation._get_parent(context) + operations = get_inherited_operations(context, parent) if parent is not None else OrderedDict() + + # Add/merge our operations + our_operations = presentation.operations # OperationDefinition + merge_operation_definitions(context, operations, our_operations, presentation._name, + presentation, 'type') + + for operation in operations.itervalues(): + operation._reset_method_cache() + + return operations + + +# +# InterfaceDefinition +# + +def get_and_override_input_definitions_from_type(context, presentation): + """ + Returns our input definitions added on top of those of the interface type, if specified. + + Allows overriding all aspects of parent interface type inputs except data types. + """ + + inputs = OrderedDict() + + # Get inputs from type + the_type = presentation._get_type(context) # InterfaceType + type_inputs = the_type._get_inputs(context) if the_type is not None else None + if type_inputs: + for input_name, type_input in type_inputs.iteritems(): + inputs[input_name] = type_input._clone(presentation) + + # Add/merge our inputs + our_inputs = presentation.inputs # PropertyDefinition + if our_inputs: + merge_input_definitions(context, inputs, our_inputs, presentation._name, None, presentation, + 'definition') + + return inputs + + +def get_and_override_operation_definitions_from_type(context, presentation): + """ + Returns our operation definitions added on top of those of the interface type, if specified. + + Allows overriding all aspects of parent interface type inputs except data types. + """ + + operations = OrderedDict() + + # Get operations from type + the_type = presentation._get_type(context) # InterfaceType + type_operations = the_type._get_operations(context) if the_type is not None else None + if type_operations: + for operations_name, type_operation in type_operations.iteritems(): + operations[operations_name] = type_operation._clone(presentation) + + # Add/merge our operations + our_operations = presentation.operations # OperationDefinition + merge_operation_definitions(context, operations, our_operations, presentation._name, + presentation, 'definition') + + return operations + + +# +# NodeType, RelationshipType, GroupType +# + +def get_inherited_interface_definitions(context, presentation, type_name, for_presentation=None): + """ + Returns our interface definitions added on top of those of our parent, if we have one + (recursively). + + Allows overriding all aspects of parent interfaces except interface and operation input data + types. + """ + + if for_presentation is None: + for_presentation = presentation + + # Get interfaces from parent + parent = presentation._get_parent(context) + interfaces = get_inherited_interface_definitions(context, parent, type_name, for_presentation) \ + if parent is not None else OrderedDict() + + # Add/merge interfaces from their types + merge_interface_definitions_from_their_types(context, interfaces, presentation) + + # Add/merge our interfaces + our_interfaces = presentation.interfaces + merge_interface_definitions(context, interfaces, our_interfaces, presentation, for_presentation) + + return interfaces + + +# +# NodeTemplate, RelationshipTemplate, GroupTemplate +# + +def get_template_interfaces(context, presentation, type_name): + """ + Returns the assigned interface_template values while making sure they are defined in the type. + This includes the interfaces themselves, their operations, and inputs for interfaces and + operations. + + Interface and operation inputs' default values, if available, will be used if we did not assign + them. + + Makes sure that required inputs indeed end up with a value. + + This code is especially complex due to the many levels of nesting involved. + """ + + template_interfaces = OrderedDict() + + the_type = presentation._get_type(context) # NodeType, RelationshipType, GroupType + # InterfaceDefinition (or InterfaceAssignment in the case of RelationshipTemplate): + interface_definitions = the_type._get_interfaces(context) if the_type is not None else None + + # Copy over interfaces from the type (will initialize inputs with default values) + if interface_definitions is not None: + for interface_name, interface_definition in interface_definitions.iteritems(): + # Note that in the case of a RelationshipTemplate, we will already have the values as + # InterfaceAssignment. It will not be converted, just cloned. + template_interfaces[interface_name] = \ + convert_interface_definition_from_type_to_template(context, interface_definition, + presentation) + + # Fill in our interfaces + our_interface_assignments = presentation.interfaces + if our_interface_assignments: + # InterfaceAssignment: + for interface_name, our_interface_assignment in our_interface_assignments.iteritems(): + if interface_name in template_interfaces: + interface_assignment = template_interfaces[interface_name] # InterfaceAssignment + # InterfaceDefinition (or InterfaceAssignment in the case of RelationshipTemplate): + interface_definition = interface_definitions[interface_name] + merge_interface(context, presentation, interface_assignment, + our_interface_assignment, interface_definition, interface_name) + else: + context.validation.report( + 'interface definition "%s" not declared at %s "%s" in "%s"' + % (interface_name, type_name, presentation.type, presentation._fullname), + locator=our_interface_assignment._locator, level=Issue.BETWEEN_TYPES) + + # Check that there are no required inputs that we haven't assigned + for interface_name, interface_template in template_interfaces.iteritems(): + if interface_name in interface_definitions: + # InterfaceDefinition (or InterfaceAssignment in the case of RelationshipTemplate): + interface_definition = interface_definitions[interface_name] + our_interface_assignment = our_interface_assignments.get(interface_name) \ + if our_interface_assignments is not None else None + validate_required_inputs(context, presentation, interface_template, + interface_definition, our_interface_assignment, interface_name) + + return template_interfaces + + +# +# Utils +# + +def convert_interface_definition_from_type_to_template(context, presentation, container): + from ..assignments import InterfaceAssignment + + if isinstance(presentation, InterfaceAssignment): + # Nothing to convert, so just clone + return presentation._clone(container) + + raw = convert_interface_definition_from_type_to_raw_template(context, presentation) + return InterfaceAssignment(name=presentation._name, raw=raw, container=container) + + +def convert_interface_definition_from_type_to_raw_template(context, presentation): # pylint: disable=invalid-name + raw = OrderedDict() + + # Copy default values for inputs + interface_inputs = presentation._get_inputs(context) + if interface_inputs is not None: + raw['inputs'] = convert_parameter_definitions_to_values(context, interface_inputs) + + # Copy operations + operations = presentation._get_operations(context) + if operations: + for operation_name, operation in operations.iteritems(): + raw[operation_name] = OrderedDict() + description = operation.description + if description is not None: + raw[operation_name]['description'] = deepcopy_with_locators(description._raw) + implementation = operation.implementation + if implementation is not None: + raw[operation_name]['implementation'] = deepcopy_with_locators(implementation._raw) + inputs = operation.inputs + if inputs is not None: + raw[operation_name]['inputs'] = convert_parameter_definitions_to_values(context, + inputs) + + return raw + + +def convert_requirement_interface_definitions_from_type_to_raw_template(context, raw_requirement, # pylint: disable=invalid-name + interface_definitions): + if not interface_definitions: + return + if 'interfaces' not in raw_requirement: + raw_requirement['interfaces'] = OrderedDict() + for interface_name, interface_definition in interface_definitions.iteritems(): + raw_interface = convert_interface_definition_from_type_to_raw_template(context, + interface_definition) + if interface_name in raw_requirement['interfaces']: + merge(raw_requirement['interfaces'][interface_name], raw_interface) + else: + raw_requirement['interfaces'][interface_name] = raw_interface + + +def merge_interface(context, presentation, interface_assignment, our_interface_assignment, + interface_definition, interface_name): + # Assign/merge interface inputs + assign_raw_inputs(context, interface_assignment._raw, our_interface_assignment.inputs, + interface_definition._get_inputs(context), interface_name, None, presentation) + + # Assign operation implementations and inputs + our_operation_templates = our_interface_assignment.operations # OperationAssignment + # OperationDefinition or OperationAssignment: + operation_definitions = interface_definition._get_operations(context) \ + if hasattr(interface_definition, '_get_operations') else interface_definition.operations + if our_operation_templates: + # OperationAssignment: + for operation_name, our_operation_template in our_operation_templates.iteritems(): + operation_definition = operation_definitions.get(operation_name) # OperationDefinition + + our_input_assignments = our_operation_template.inputs + our_implementation = our_operation_template.implementation + + if operation_definition is None: + context.validation.report( + 'interface definition "%s" refers to an unknown operation "%s" in "%s"' + % (interface_name, operation_name, presentation._fullname), + locator=our_operation_template._locator, level=Issue.BETWEEN_TYPES) + + if (our_input_assignments is not None) or (our_implementation is not None): + # Make sure we have the dict + if (operation_name not in interface_assignment._raw) \ + or (interface_assignment._raw[operation_name] is None): + interface_assignment._raw[operation_name] = OrderedDict() + + if our_implementation is not None: + interface_assignment._raw[operation_name]['implementation'] = \ + deepcopy_with_locators(our_implementation._raw) + + # Assign/merge operation inputs + input_definitions = operation_definition.inputs \ + if operation_definition is not None else None + assign_raw_inputs(context, interface_assignment._raw[operation_name], + our_input_assignments, input_definitions, interface_name, + operation_name, presentation) + + +def merge_raw_input_definition(context, the_raw_input, our_input, interface_name, operation_name, + presentation, type_name): + # Check if we changed the type + # TODO: allow a sub-type? + input_type1 = the_raw_input.get('type') + input_type2 = our_input.type + if input_type1 != input_type2: + if operation_name is not None: + context.validation.report( + 'interface %s "%s" changes operation input "%s.%s" type from "%s" to "%s" in "%s"' + % (type_name, interface_name, operation_name, our_input._name, input_type1, + input_type2, presentation._fullname), + locator=input_type2._locator, level=Issue.BETWEEN_TYPES) + else: + context.validation.report( + 'interface %s "%s" changes input "%s" type from "%s" to "%s" in "%s"' + % (type_name, interface_name, our_input._name, input_type1, input_type2, + presentation._fullname), + locator=input_type2._locator, level=Issue.BETWEEN_TYPES) + + # Merge + merge(the_raw_input, our_input._raw) + + +def merge_input_definitions(context, inputs, our_inputs, interface_name, operation_name, + presentation, type_name): + for input_name, our_input in our_inputs.iteritems(): + if input_name in inputs: + merge_raw_input_definition(context, inputs[input_name]._raw, our_input, interface_name, + operation_name, presentation, type_name) + else: + inputs[input_name] = our_input._clone(presentation) + + +def merge_raw_input_definitions(context, raw_inputs, our_inputs, interface_name, operation_name, + presentation, type_name): + for input_name, our_input in our_inputs.iteritems(): + if input_name in raw_inputs: + merge_raw_input_definition(context, raw_inputs[input_name], our_input, interface_name, + operation_name, presentation, type_name) + else: + raw_inputs[input_name] = deepcopy_with_locators(our_input._raw) + + +def merge_raw_operation_definition(context, raw_operation, our_operation, interface_name, + presentation, type_name): + if not isinstance(our_operation._raw, dict): + # Convert short form to long form + raw_operation['implementation'] = deepcopy_with_locators(our_operation._raw) + return + + # Add/merge inputs + our_operation_inputs = our_operation.inputs + if our_operation_inputs: + # Make sure we have the dict + if ('inputs' not in raw_operation) or (raw_operation.get('inputs') is None): + raw_operation['inputs'] = OrderedDict() + + merge_raw_input_definitions(context, raw_operation['inputs'], our_operation_inputs, + interface_name, our_operation._name, presentation, type_name) + + # Override the description + if our_operation._raw.get('description') is not None: + raw_operation['description'] = deepcopy_with_locators(our_operation._raw['description']) + + # Add/merge implementation + if our_operation._raw.get('implementation') is not None: + if raw_operation.get('implementation') is not None: + merge(raw_operation['implementation'], + deepcopy_with_locators(our_operation._raw['implementation'])) + else: + raw_operation['implementation'] = \ + deepcopy_with_locators(our_operation._raw['implementation']) + + +def merge_operation_definitions(context, operations, our_operations, interface_name, presentation, + type_name): + if not our_operations: + return + for operation_name, our_operation in our_operations.iteritems(): + if operation_name in operations: + merge_raw_operation_definition(context, operations[operation_name]._raw, our_operation, + interface_name, presentation, type_name) + else: + operations[operation_name] = our_operation._clone(presentation) + + +def merge_raw_operation_definitions(context, raw_operations, our_operations, interface_name, + presentation, type_name): + for operation_name, our_operation in our_operations.iteritems(): + if operation_name in raw_operations: + raw_operation = raw_operations[operation_name] + if isinstance(raw_operation, basestring): + # Convert short form to long form + raw_operations[operation_name] = OrderedDict((('implementation', raw_operation),)) + raw_operation = raw_operations[operation_name] + merge_raw_operation_definition(context, raw_operation, our_operation, interface_name, + presentation, type_name) + else: + raw_operations[operation_name] = deepcopy_with_locators(our_operation._raw) + + +# From either an InterfaceType or an InterfaceDefinition: +def merge_interface_definition(context, interface, our_source, presentation, type_name): + if hasattr(our_source, 'type'): + # Check if we changed the interface type + type1 = interface._get_type(context) + type2 = our_source._get_type(context) + + if (type2 is not None) and not type1._is_descendant(context, type2): + context.validation.report( + 'interface definition type "{0}" is not a descendant of overridden ' + 'interface definition type "{1}"' \ + .format(type1._name, type2._name), + locator=our_source._locator, level=Issue.BETWEEN_TYPES) + + # Add/merge inputs + our_interface_inputs = our_source._get_inputs(context) \ + if hasattr(our_source, '_get_inputs') else our_source.inputs + if our_interface_inputs: + # Make sure we have the dict + if ('inputs' not in interface._raw) or (interface._raw.get('inputs') is None): + interface._raw['inputs'] = OrderedDict() + + merge_raw_input_definitions(context, interface._raw['inputs'], our_interface_inputs, + our_source._name, None, presentation, type_name) + + # Add/merge operations + our_operations = our_source._get_operations(context) \ + if hasattr(our_source, '_get_operations') else our_source.operations + if our_operations is not None: + merge_raw_operation_definitions(context, interface._raw, our_operations, our_source._name, + presentation, type_name) + + +def merge_interface_definitions(context, interfaces, our_interfaces, presentation, + for_presentation=None): + if not our_interfaces: + return + for name, our_interface in our_interfaces.iteritems(): + if name in interfaces: + merge_interface_definition(context, interfaces[name], our_interface, presentation, + 'definition') + else: + interfaces[name] = our_interface._clone(for_presentation) + + +def merge_interface_definitions_from_their_types(context, interfaces, presentation): + for interface in interfaces.itervalues(): + the_type = interface._get_type(context) # InterfaceType + if the_type is not None: + merge_interface_definition(context, interface, the_type, presentation, 'type') + + +def assign_raw_inputs(context, values, assignments, definitions, interface_name, operation_name, + presentation): + if not assignments: + return + + # Make sure we have the dict + if ('inputs' not in values) or (values['inputs'] is None): + values['inputs'] = OrderedDict() + + # Assign inputs + for input_name, assignment in assignments.iteritems(): + if (definitions is not None) and (input_name not in definitions): + if operation_name is not None: + context.validation.report( + 'interface definition "%s" assigns a value to an unknown operation input' + ' "%s.%s" in "%s"' + % (interface_name, operation_name, input_name, presentation._fullname), + locator=assignment._locator, level=Issue.BETWEEN_TYPES) + else: + context.validation.report( + 'interface definition "%s" assigns a value to an unknown input "%s" in "%s"' + % (interface_name, input_name, presentation._fullname), + locator=assignment._locator, level=Issue.BETWEEN_TYPES) + + definition = definitions.get(input_name) if definitions is not None else None + + # Note: default value has already been assigned + + # Coerce value + values['inputs'][input_name] = coerce_parameter_value(context, assignment, definition, + assignment.value) + + +def validate_required_inputs(context, presentation, assignment, definition, original_assignment, + interface_name, operation_name=None): + # The validation of the `required` field of inputs that belong to operations and interfaces + # (as opposed to topology template and workflow inputs) is done only in the parsing stage. + # This reasoning follows the TOSCA spirit, where anything that is declared as required in the + # type, must be assigned in the corresponding template. + input_definitions = definition.inputs + if input_definitions: + for input_name, input_definition in input_definitions.iteritems(): + if input_definition.required: + prop = assignment.inputs.get(input_name) \ + if ((assignment is not None) and (assignment.inputs is not None)) else None + value = prop.value if prop is not None else None + value = value.value if value is not None else None + if value is None: + if operation_name is not None: + context.validation.report( + 'interface definition "%s" does not assign a value to a required' + ' operation input "%s.%s" in "%s"' + % (interface_name, operation_name, input_name, presentation._fullname), + locator=get_locator(original_assignment, presentation._locator), + level=Issue.BETWEEN_TYPES) + else: + context.validation.report( + 'interface definition "%s" does not assign a value to a required input' + ' "%s" in "%s"' + % (interface_name, input_name, presentation._fullname), + locator=get_locator(original_assignment, presentation._locator), + level=Issue.BETWEEN_TYPES) + + if operation_name is not None: + return + + assignment_operations = assignment.operations + operation_definitions = definition._get_operations(context) + if operation_definitions: + for operation_name, operation_definition in operation_definitions.iteritems(): + assignment_operation = assignment_operations.get(operation_name) \ + if assignment_operations is not None else None + original_operation = \ + original_assignment.operations.get(operation_name, original_assignment) \ + if (original_assignment is not None) \ + and (original_assignment.operations is not None) \ + else original_assignment + validate_required_inputs(context, presentation, assignment_operation, + operation_definition, original_operation, interface_name, + operation_name) |