--- /dev/null
+# 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)
Code Review / multicloud / azure.git / blobdiff