| # 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.parser.validation import Issue |
| from aria.utils.collections import (deepcopy_with_locators, OrderedDict) |
| |
| from .parameters import (convert_parameter_definitions_to_values, validate_required_values, |
| coerce_parameter_value) |
| from .interfaces import (convert_requirement_interface_definitions_from_type_to_raw_template, |
| merge_interface_definitions, merge_interface, validate_required_inputs) |
| |
| |
| # |
| # NodeType |
| # |
| |
| def get_inherited_requirement_definitions(context, presentation): |
| """ |
| Returns our requirement definitions added on top of those of our parent, if we have one |
| (recursively). |
| |
| Allows overriding requirement definitions if they have the same name. |
| """ |
| |
| parent = presentation._get_parent(context) |
| requirement_definitions = get_inherited_requirement_definitions(context, parent) \ |
| if parent is not None else [] |
| |
| our_requirement_definitions = presentation.requirements |
| if our_requirement_definitions: |
| for requirement_name, our_requirement_definition in our_requirement_definitions: |
| # Remove existing requirement definitions of this name if they exist |
| for name, requirement_definition in requirement_definitions: |
| if name == requirement_name: |
| requirement_definitions.remove((name, requirement_definition)) |
| |
| requirement_definitions.append((requirement_name, our_requirement_definition)) |
| |
| return requirement_definitions |
| |
| |
| # |
| # NodeTemplate |
| # |
| |
| def get_template_requirements(context, presentation): |
| """ |
| Returns our requirements added on top of those of the node type if they exist there. |
| |
| If the requirement has a relationship, the relationship properties and interfaces are assigned. |
| |
| Returns the assigned property, interface input, and interface operation input values while |
| making sure they are defined in our type. Default values, if available, will be used if we did |
| not assign them. Also makes sure that required properties and inputs indeed end up with a value. |
| """ |
| |
| requirement_assignments = [] |
| |
| the_type = presentation._get_type(context) # NodeType |
| requirement_definitions = the_type._get_requirements(context) if the_type is not None else None |
| |
| # Add our requirement assignments |
| our_requirement_assignments = presentation.requirements |
| if our_requirement_assignments: |
| add_requirement_assignments(context, presentation, requirement_assignments, |
| requirement_definitions, our_requirement_assignments) |
| |
| # Validate occurrences |
| if requirement_definitions: |
| for requirement_name, requirement_definition in requirement_definitions: |
| # Allowed occurrences |
| allowed_occurrences = requirement_definition.occurrences |
| allowed_occurrences = allowed_occurrences if allowed_occurrences is not None else None |
| |
| # Count actual occurrences |
| actual_occurrences = 0 |
| for name, _ in requirement_assignments: |
| if name == requirement_name: |
| actual_occurrences += 1 |
| |
| if allowed_occurrences is None: |
| # If not specified, we interpret this to mean that exactly 1 occurrence is required |
| if actual_occurrences == 0: |
| # If it's not there, we will automatically add it (this behavior is not in the |
| # TOSCA spec, but seems implied) |
| requirement_assignment, \ |
| relationship_property_definitions, \ |
| relationship_interface_definitions = \ |
| convert_requirement_from_definition_to_assignment(context, |
| requirement_definition, |
| None, presentation) |
| validate_requirement_assignment(context, presentation, requirement_assignment, |
| relationship_property_definitions, |
| relationship_interface_definitions) |
| requirement_assignments.append((requirement_name, requirement_assignment)) |
| elif actual_occurrences > 1: |
| context.validation.report( |
| 'requirement "%s" is allowed only one occurrence in "%s": %d' |
| % (requirement_name, presentation._fullname, actual_occurrences), |
| locator=presentation._locator, level=Issue.BETWEEN_TYPES) |
| else: |
| if not allowed_occurrences.is_in(actual_occurrences): |
| if allowed_occurrences.value[1] == 'UNBOUNDED': |
| context.validation.report( |
| 'requirement "%s" does not have at least %d occurrences in "%s": has %d' |
| % (requirement_name, allowed_occurrences.value[0], |
| presentation._fullname, actual_occurrences), |
| locator=presentation._locator, level=Issue.BETWEEN_TYPES) |
| else: |
| context.validation.report( |
| 'requirement "%s" is allowed between %d and %d occurrences in "%s":' |
| ' has %d' |
| % (requirement_name, allowed_occurrences.value[0], |
| allowed_occurrences.value[1], presentation._fullname, |
| actual_occurrences), |
| locator=presentation._locator, level=Issue.BETWEEN_TYPES) |
| |
| return requirement_assignments |
| |
| |
| # |
| # Utils |
| # |
| |
| def convert_requirement_from_definition_to_assignment(context, requirement_definition, # pylint: disable=too-many-branches |
| our_requirement_assignment, container): |
| from ..assignments import RequirementAssignment |
| |
| raw = OrderedDict() |
| |
| # Capability type name: |
| raw['capability'] = deepcopy_with_locators(requirement_definition.capability) |
| |
| node_type = requirement_definition._get_node_type(context) |
| if node_type is not None: |
| raw['node'] = deepcopy_with_locators(node_type._name) |
| |
| relationship_type = None |
| relationship_template = None |
| relationship_property_definitions = None |
| relationship_interface_definitions = None |
| |
| # First try to find the relationship if we declared it |
| # RelationshipAssignment: |
| our_relationship = our_requirement_assignment.relationship \ |
| if our_requirement_assignment is not None else None |
| if our_relationship is not None: |
| relationship_type, relationship_type_variant = our_relationship._get_type(context) |
| if relationship_type_variant == 'relationship_template': |
| relationship_template = relationship_type |
| relationship_type = relationship_template._get_type(context) |
| |
| definition_relationship_type = None |
| relationship_definition = requirement_definition.relationship # RelationshipDefinition |
| if relationship_definition is not None: |
| definition_relationship_type = relationship_definition._get_type(context) |
| |
| # If not exists, try at the node type |
| if relationship_type is None: |
| relationship_type = definition_relationship_type |
| else: |
| # Make sure the type is derived |
| if not definition_relationship_type._is_descendant(context, relationship_type): |
| context.validation.report( |
| 'assigned relationship type "%s" is not a descendant of declared relationship type' |
| ' "%s"' \ |
| % (relationship_type._name, definition_relationship_type._name), |
| locator=container._locator, level=Issue.BETWEEN_TYPES) |
| |
| if relationship_type is not None: |
| raw['relationship'] = OrderedDict() |
| |
| type_name = our_relationship.type if our_relationship is not None else None |
| if type_name is None: |
| type_name = relationship_type._name |
| |
| raw['relationship']['type'] = deepcopy_with_locators(type_name) |
| |
| # These are our property definitions |
| relationship_property_definitions = relationship_type._get_properties(context) |
| |
| if relationship_template is not None: |
| # Property values from template |
| raw['relationship']['properties'] = relationship_template._get_property_values(context) |
| else: |
| if relationship_property_definitions: |
| # Convert property definitions to values |
| raw['relationship']['properties'] = \ |
| convert_parameter_definitions_to_values(context, |
| relationship_property_definitions) |
| |
| # These are our interface definitions |
| # InterfaceDefinition: |
| relationship_interface_definitions = OrderedDict(relationship_type._get_interfaces(context)) |
| |
| # Convert interface definitions to templates |
| convert_requirement_interface_definitions_from_type_to_raw_template( |
| context, |
| raw['relationship'], |
| relationship_interface_definitions) |
| |
| if relationship_definition: |
| # Merge extra interface definitions |
| # InterfaceDefinition: |
| definition_interface_definitions = relationship_definition.interfaces |
| merge_interface_definitions(context, relationship_interface_definitions, |
| definition_interface_definitions, requirement_definition, |
| container) |
| |
| if relationship_template is not None: |
| # Interfaces from template |
| interfaces = relationship_template._get_interfaces(context) |
| if interfaces: |
| raw['relationship']['interfaces'] = OrderedDict() |
| for interface_name, interface in interfaces.iteritems(): |
| raw['relationship']['interfaces'][interface_name] = interface._raw |
| |
| return \ |
| RequirementAssignment(name=requirement_definition._name, raw=raw, container=container), \ |
| relationship_property_definitions, \ |
| relationship_interface_definitions |
| |
| |
| def add_requirement_assignments(context, presentation, requirement_assignments, |
| requirement_definitions, our_requirement_assignments): |
| for requirement_name, our_requirement_assignment in our_requirement_assignments: |
| requirement_definition = get_first_requirement(requirement_definitions, requirement_name) |
| if requirement_definition is not None: |
| requirement_assignment, \ |
| relationship_property_definitions, \ |
| relationship_interface_definitions = \ |
| convert_requirement_from_definition_to_assignment(context, requirement_definition, |
| our_requirement_assignment, |
| presentation) |
| merge_requirement_assignment(context, |
| relationship_property_definitions, |
| relationship_interface_definitions, |
| requirement_assignment, our_requirement_assignment) |
| validate_requirement_assignment(context, |
| our_requirement_assignment.relationship \ |
| or our_requirement_assignment, |
| requirement_assignment, |
| relationship_property_definitions, |
| relationship_interface_definitions) |
| requirement_assignments.append((requirement_name, requirement_assignment)) |
| else: |
| context.validation.report('requirement "%s" not declared at node type "%s" in "%s"' |
| % (requirement_name, presentation.type, |
| presentation._fullname), |
| locator=our_requirement_assignment._locator, |
| level=Issue.BETWEEN_TYPES) |
| |
| |
| def merge_requirement_assignment(context, relationship_property_definitions, |
| relationship_interface_definitions, requirement, our_requirement): |
| our_capability = our_requirement.capability |
| if our_capability is not None: |
| requirement._raw['capability'] = deepcopy_with_locators(our_capability) |
| |
| our_node = our_requirement.node |
| if our_node is not None: |
| requirement._raw['node'] = deepcopy_with_locators(our_node) |
| |
| our_node_filter = our_requirement.node_filter |
| if our_node_filter is not None: |
| requirement._raw['node_filter'] = deepcopy_with_locators(our_node_filter._raw) |
| |
| our_relationship = our_requirement.relationship # RelationshipAssignment |
| if (our_relationship is not None) and (our_relationship.type is None): |
| # Make sure we have a dict |
| if 'relationship' not in requirement._raw: |
| requirement._raw['relationship'] = OrderedDict() |
| |
| merge_requirement_assignment_relationship(context, our_relationship, |
| relationship_property_definitions, |
| relationship_interface_definitions, |
| requirement, our_relationship) |
| |
| |
| def merge_requirement_assignment_relationship(context, presentation, property_definitions, |
| interface_definitions, requirement, our_relationship): |
| our_relationship_properties = our_relationship._raw.get('properties') |
| if our_relationship_properties: |
| # Make sure we have a dict |
| if 'properties' not in requirement._raw['relationship']: |
| requirement._raw['relationship']['properties'] = OrderedDict() |
| |
| # Merge our properties |
| for property_name, prop in our_relationship_properties.iteritems(): |
| if property_name in property_definitions: |
| definition = property_definitions[property_name] |
| requirement._raw['relationship']['properties'][property_name] = \ |
| coerce_parameter_value(context, presentation, definition, prop) |
| else: |
| context.validation.report( |
| 'relationship property "%s" not declared at definition of requirement "%s"' |
| ' in "%s"' |
| % (property_name, requirement._fullname, |
| presentation._container._container._fullname), |
| locator=our_relationship._get_child_locator('properties', property_name), |
| level=Issue.BETWEEN_TYPES) |
| |
| our_interfaces = our_relationship.interfaces |
| if our_interfaces: |
| # Make sure we have a dict |
| if 'interfaces' not in requirement._raw['relationship']: |
| requirement._raw['relationship']['interfaces'] = OrderedDict() |
| |
| # Merge interfaces |
| for interface_name, our_interface in our_interfaces.iteritems(): |
| if interface_name not in requirement._raw['relationship']['interfaces']: |
| requirement._raw['relationship']['interfaces'][interface_name] = OrderedDict() |
| |
| if (interface_definitions is not None) and (interface_name in interface_definitions): |
| interface_definition = interface_definitions[interface_name] |
| interface_assignment = requirement.relationship.interfaces[interface_name] |
| merge_interface(context, presentation, interface_assignment, our_interface, |
| interface_definition, interface_name) |
| else: |
| context.validation.report( |
| 'relationship interface "%s" not declared at definition of requirement "%s"' |
| ' in "%s"' |
| % (interface_name, requirement._fullname, |
| presentation._container._container._fullname), |
| locator=our_relationship._locator, level=Issue.BETWEEN_TYPES) |
| |
| |
| def validate_requirement_assignment(context, presentation, requirement_assignment, |
| relationship_property_definitions, |
| relationship_interface_definitions): |
| relationship = requirement_assignment.relationship |
| if relationship is None: |
| return |
| |
| validate_required_values(context, presentation, relationship.properties, |
| relationship_property_definitions) |
| |
| if relationship_interface_definitions: |
| for interface_name, relationship_interface_definition \ |
| in relationship_interface_definitions.iteritems(): |
| interface_assignment = relationship.interfaces.get(interface_name) \ |
| if relationship.interfaces is not None else None |
| validate_required_inputs(context, presentation, interface_assignment, |
| relationship_interface_definition, None, interface_name) |
| |
| |
| def get_first_requirement(requirement_definitions, name): |
| if requirement_definitions is not None: |
| for requirement_name, requirement_definition in requirement_definitions: |
| if requirement_name == name: |
| return requirement_definition |
| return None |
