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
8 # http://www.apache.org/licenses/LICENSE-2.0
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.
16 from aria.utils.collections import (merge, deepcopy_with_locators, OrderedDict)
17 from aria.parser.presentation import get_locator
18 from aria.parser.validation import Issue
20 from .parameters import (coerce_parameter_value, convert_parameter_definitions_to_values)
27 def get_inherited_operations(context, presentation):
29 Returns our operation definitions added on top of those of our parent, if we have one
32 Allows overriding all aspects of parent operations except input data types.
35 # Get operations from parent
36 parent = presentation._get_parent(context)
37 operations = get_inherited_operations(context, parent) if parent is not None else OrderedDict()
39 # Add/merge our operations
40 our_operations = presentation.operations # OperationDefinition
41 merge_operation_definitions(context, operations, our_operations, presentation._name,
44 for operation in operations.itervalues():
45 operation._reset_method_cache()
54 def get_and_override_input_definitions_from_type(context, presentation):
56 Returns our input definitions added on top of those of the interface type, if specified.
58 Allows overriding all aspects of parent interface type inputs except data types.
61 inputs = OrderedDict()
63 # Get inputs from type
64 the_type = presentation._get_type(context) # InterfaceType
65 type_inputs = the_type._get_inputs(context) if the_type is not None else None
67 for input_name, type_input in type_inputs.iteritems():
68 inputs[input_name] = type_input._clone(presentation)
70 # Add/merge our inputs
71 our_inputs = presentation.inputs # PropertyDefinition
73 merge_input_definitions(context, inputs, our_inputs, presentation._name, None, presentation,
79 def get_and_override_operation_definitions_from_type(context, presentation):
81 Returns our operation definitions added on top of those of the interface type, if specified.
83 Allows overriding all aspects of parent interface type inputs except data types.
86 operations = OrderedDict()
88 # Get operations from type
89 the_type = presentation._get_type(context) # InterfaceType
90 type_operations = the_type._get_operations(context) if the_type is not None else None
92 for operations_name, type_operation in type_operations.iteritems():
93 operations[operations_name] = type_operation._clone(presentation)
95 # Add/merge our operations
96 our_operations = presentation.operations # OperationDefinition
97 merge_operation_definitions(context, operations, our_operations, presentation._name,
98 presentation, 'definition')
104 # NodeType, RelationshipType, GroupType
107 def get_inherited_interface_definitions(context, presentation, type_name, for_presentation=None):
109 Returns our interface definitions added on top of those of our parent, if we have one
112 Allows overriding all aspects of parent interfaces except interface and operation input data
116 if for_presentation is None:
117 for_presentation = presentation
119 # Get interfaces from parent
120 parent = presentation._get_parent(context)
121 interfaces = get_inherited_interface_definitions(context, parent, type_name, for_presentation) \
122 if parent is not None else OrderedDict()
124 # Add/merge interfaces from their types
125 merge_interface_definitions_from_their_types(context, interfaces, presentation)
127 # Add/merge our interfaces
128 our_interfaces = presentation.interfaces
129 merge_interface_definitions(context, interfaces, our_interfaces, presentation, for_presentation)
135 # NodeTemplate, RelationshipTemplate, GroupTemplate
138 def get_template_interfaces(context, presentation, type_name):
140 Returns the assigned interface_template values while making sure they are defined in the type.
141 This includes the interfaces themselves, their operations, and inputs for interfaces and
144 Interface and operation inputs' default values, if available, will be used if we did not assign
147 Makes sure that required inputs indeed end up with a value.
149 This code is especially complex due to the many levels of nesting involved.
152 template_interfaces = OrderedDict()
154 the_type = presentation._get_type(context) # NodeType, RelationshipType, GroupType
155 # InterfaceDefinition (or InterfaceAssignment in the case of RelationshipTemplate):
156 interface_definitions = the_type._get_interfaces(context) if the_type is not None else None
158 # Copy over interfaces from the type (will initialize inputs with default values)
159 if interface_definitions is not None:
160 for interface_name, interface_definition in interface_definitions.iteritems():
161 # Note that in the case of a RelationshipTemplate, we will already have the values as
162 # InterfaceAssignment. It will not be converted, just cloned.
163 template_interfaces[interface_name] = \
164 convert_interface_definition_from_type_to_template(context, interface_definition,
167 # Fill in our interfaces
168 our_interface_assignments = presentation.interfaces
169 if our_interface_assignments:
170 # InterfaceAssignment:
171 for interface_name, our_interface_assignment in our_interface_assignments.iteritems():
172 if interface_name in template_interfaces:
173 interface_assignment = template_interfaces[interface_name] # InterfaceAssignment
174 # InterfaceDefinition (or InterfaceAssignment in the case of RelationshipTemplate):
175 interface_definition = interface_definitions[interface_name]
176 merge_interface(context, presentation, interface_assignment,
177 our_interface_assignment, interface_definition, interface_name)
179 context.validation.report(
180 'interface definition "%s" not declared at %s "%s" in "%s"'
181 % (interface_name, type_name, presentation.type, presentation._fullname),
182 locator=our_interface_assignment._locator, level=Issue.BETWEEN_TYPES)
184 # Check that there are no required inputs that we haven't assigned
185 for interface_name, interface_template in template_interfaces.iteritems():
186 if interface_name in interface_definitions:
187 # InterfaceDefinition (or InterfaceAssignment in the case of RelationshipTemplate):
188 interface_definition = interface_definitions[interface_name]
189 our_interface_assignment = our_interface_assignments.get(interface_name) \
190 if our_interface_assignments is not None else None
191 validate_required_inputs(context, presentation, interface_template,
192 interface_definition, our_interface_assignment, interface_name)
194 return template_interfaces
201 def convert_interface_definition_from_type_to_template(context, presentation, container):
202 from ..assignments import InterfaceAssignment
204 if isinstance(presentation, InterfaceAssignment):
205 # Nothing to convert, so just clone
206 return presentation._clone(container)
208 raw = convert_interface_definition_from_type_to_raw_template(context, presentation)
209 return InterfaceAssignment(name=presentation._name, raw=raw, container=container)
212 def convert_interface_definition_from_type_to_raw_template(context, presentation): # pylint: disable=invalid-name
215 # Copy default values for inputs
216 interface_inputs = presentation._get_inputs(context)
217 if interface_inputs is not None:
218 raw['inputs'] = convert_parameter_definitions_to_values(context, interface_inputs)
221 operations = presentation._get_operations(context)
223 for operation_name, operation in operations.iteritems():
224 raw[operation_name] = OrderedDict()
225 description = operation.description
226 if description is not None:
227 raw[operation_name]['description'] = deepcopy_with_locators(description._raw)
228 implementation = operation.implementation
229 if implementation is not None:
230 raw[operation_name]['implementation'] = deepcopy_with_locators(implementation._raw)
231 inputs = operation.inputs
232 if inputs is not None:
233 raw[operation_name]['inputs'] = convert_parameter_definitions_to_values(context,
239 def convert_requirement_interface_definitions_from_type_to_raw_template(context, raw_requirement, # pylint: disable=invalid-name
240 interface_definitions):
241 if not interface_definitions:
243 if 'interfaces' not in raw_requirement:
244 raw_requirement['interfaces'] = OrderedDict()
245 for interface_name, interface_definition in interface_definitions.iteritems():
246 raw_interface = convert_interface_definition_from_type_to_raw_template(context,
247 interface_definition)
248 if interface_name in raw_requirement['interfaces']:
249 merge(raw_requirement['interfaces'][interface_name], raw_interface)
251 raw_requirement['interfaces'][interface_name] = raw_interface
254 def merge_interface(context, presentation, interface_assignment, our_interface_assignment,
255 interface_definition, interface_name):
256 # Assign/merge interface inputs
257 assign_raw_inputs(context, interface_assignment._raw, our_interface_assignment.inputs,
258 interface_definition._get_inputs(context), interface_name, None, presentation)
260 # Assign operation implementations and inputs
261 our_operation_templates = our_interface_assignment.operations # OperationAssignment
262 # OperationDefinition or OperationAssignment:
263 operation_definitions = interface_definition._get_operations(context) \
264 if hasattr(interface_definition, '_get_operations') else interface_definition.operations
265 if our_operation_templates:
266 # OperationAssignment:
267 for operation_name, our_operation_template in our_operation_templates.iteritems():
268 operation_definition = operation_definitions.get(operation_name) # OperationDefinition
270 our_input_assignments = our_operation_template.inputs
271 our_implementation = our_operation_template.implementation
273 if operation_definition is None:
274 context.validation.report(
275 'interface definition "%s" refers to an unknown operation "%s" in "%s"'
276 % (interface_name, operation_name, presentation._fullname),
277 locator=our_operation_template._locator, level=Issue.BETWEEN_TYPES)
279 if (our_input_assignments is not None) or (our_implementation is not None):
280 # Make sure we have the dict
281 if (operation_name not in interface_assignment._raw) \
282 or (interface_assignment._raw[operation_name] is None):
283 interface_assignment._raw[operation_name] = OrderedDict()
285 if our_implementation is not None:
286 interface_assignment._raw[operation_name]['implementation'] = \
287 deepcopy_with_locators(our_implementation._raw)
289 # Assign/merge operation inputs
290 input_definitions = operation_definition.inputs \
291 if operation_definition is not None else None
292 assign_raw_inputs(context, interface_assignment._raw[operation_name],
293 our_input_assignments, input_definitions, interface_name,
294 operation_name, presentation)
297 def merge_raw_input_definition(context, the_raw_input, our_input, interface_name, operation_name,
298 presentation, type_name):
299 # Check if we changed the type
300 # TODO: allow a sub-type?
301 input_type1 = the_raw_input.get('type')
302 input_type2 = our_input.type
303 if input_type1 != input_type2:
304 if operation_name is not None:
305 context.validation.report(
306 'interface %s "%s" changes operation input "%s.%s" type from "%s" to "%s" in "%s"'
307 % (type_name, interface_name, operation_name, our_input._name, input_type1,
308 input_type2, presentation._fullname),
309 locator=input_type2._locator, level=Issue.BETWEEN_TYPES)
311 context.validation.report(
312 'interface %s "%s" changes input "%s" type from "%s" to "%s" in "%s"'
313 % (type_name, interface_name, our_input._name, input_type1, input_type2,
314 presentation._fullname),
315 locator=input_type2._locator, level=Issue.BETWEEN_TYPES)
318 merge(the_raw_input, our_input._raw)
321 def merge_input_definitions(context, inputs, our_inputs, interface_name, operation_name,
322 presentation, type_name):
323 for input_name, our_input in our_inputs.iteritems():
324 if input_name in inputs:
325 merge_raw_input_definition(context, inputs[input_name]._raw, our_input, interface_name,
326 operation_name, presentation, type_name)
328 inputs[input_name] = our_input._clone(presentation)
331 def merge_raw_input_definitions(context, raw_inputs, our_inputs, interface_name, operation_name,
332 presentation, type_name):
333 for input_name, our_input in our_inputs.iteritems():
334 if input_name in raw_inputs:
335 merge_raw_input_definition(context, raw_inputs[input_name], our_input, interface_name,
336 operation_name, presentation, type_name)
338 raw_inputs[input_name] = deepcopy_with_locators(our_input._raw)
341 def merge_raw_operation_definition(context, raw_operation, our_operation, interface_name,
342 presentation, type_name):
343 if not isinstance(our_operation._raw, dict):
344 # Convert short form to long form
345 raw_operation['implementation'] = deepcopy_with_locators(our_operation._raw)
349 our_operation_inputs = our_operation.inputs
350 if our_operation_inputs:
351 # Make sure we have the dict
352 if ('inputs' not in raw_operation) or (raw_operation.get('inputs') is None):
353 raw_operation['inputs'] = OrderedDict()
355 merge_raw_input_definitions(context, raw_operation['inputs'], our_operation_inputs,
356 interface_name, our_operation._name, presentation, type_name)
358 # Override the description
359 if our_operation._raw.get('description') is not None:
360 raw_operation['description'] = deepcopy_with_locators(our_operation._raw['description'])
362 # Add/merge implementation
363 if our_operation._raw.get('implementation') is not None:
364 if raw_operation.get('implementation') is not None:
365 merge(raw_operation['implementation'],
366 deepcopy_with_locators(our_operation._raw['implementation']))
368 raw_operation['implementation'] = \
369 deepcopy_with_locators(our_operation._raw['implementation'])
372 def merge_operation_definitions(context, operations, our_operations, interface_name, presentation,
374 if not our_operations:
376 for operation_name, our_operation in our_operations.iteritems():
377 if operation_name in operations:
378 merge_raw_operation_definition(context, operations[operation_name]._raw, our_operation,
379 interface_name, presentation, type_name)
381 operations[operation_name] = our_operation._clone(presentation)
384 def merge_raw_operation_definitions(context, raw_operations, our_operations, interface_name,
385 presentation, type_name):
386 for operation_name, our_operation in our_operations.iteritems():
387 if operation_name in raw_operations:
388 raw_operation = raw_operations[operation_name]
389 if isinstance(raw_operation, basestring):
390 # Convert short form to long form
391 raw_operations[operation_name] = OrderedDict((('implementation', raw_operation),))
392 raw_operation = raw_operations[operation_name]
393 merge_raw_operation_definition(context, raw_operation, our_operation, interface_name,
394 presentation, type_name)
396 raw_operations[operation_name] = deepcopy_with_locators(our_operation._raw)
399 # From either an InterfaceType or an InterfaceDefinition:
400 def merge_interface_definition(context, interface, our_source, presentation, type_name):
401 if hasattr(our_source, 'type'):
402 # Check if we changed the interface type
403 type1 = interface._get_type(context)
404 type2 = our_source._get_type(context)
406 if (type2 is not None) and not type1._is_descendant(context, type2):
407 context.validation.report(
408 'interface definition type "{0}" is not a descendant of overridden '
409 'interface definition type "{1}"' \
410 .format(type1._name, type2._name),
411 locator=our_source._locator, level=Issue.BETWEEN_TYPES)
414 our_interface_inputs = our_source._get_inputs(context) \
415 if hasattr(our_source, '_get_inputs') else our_source.inputs
416 if our_interface_inputs:
417 # Make sure we have the dict
418 if ('inputs' not in interface._raw) or (interface._raw.get('inputs') is None):
419 interface._raw['inputs'] = OrderedDict()
421 merge_raw_input_definitions(context, interface._raw['inputs'], our_interface_inputs,
422 our_source._name, None, presentation, type_name)
424 # Add/merge operations
425 our_operations = our_source._get_operations(context) \
426 if hasattr(our_source, '_get_operations') else our_source.operations
427 if our_operations is not None:
428 merge_raw_operation_definitions(context, interface._raw, our_operations, our_source._name,
429 presentation, type_name)
432 def merge_interface_definitions(context, interfaces, our_interfaces, presentation,
433 for_presentation=None):
434 if not our_interfaces:
436 for name, our_interface in our_interfaces.iteritems():
437 if name in interfaces:
438 merge_interface_definition(context, interfaces[name], our_interface, presentation,
441 interfaces[name] = our_interface._clone(for_presentation)
444 def merge_interface_definitions_from_their_types(context, interfaces, presentation):
445 for interface in interfaces.itervalues():
446 the_type = interface._get_type(context) # InterfaceType
447 if the_type is not None:
448 merge_interface_definition(context, interface, the_type, presentation, 'type')
451 def assign_raw_inputs(context, values, assignments, definitions, interface_name, operation_name,
456 # Make sure we have the dict
457 if ('inputs' not in values) or (values['inputs'] is None):
458 values['inputs'] = OrderedDict()
461 for input_name, assignment in assignments.iteritems():
462 if (definitions is not None) and (input_name not in definitions):
463 if operation_name is not None:
464 context.validation.report(
465 'interface definition "%s" assigns a value to an unknown operation input'
467 % (interface_name, operation_name, input_name, presentation._fullname),
468 locator=assignment._locator, level=Issue.BETWEEN_TYPES)
470 context.validation.report(
471 'interface definition "%s" assigns a value to an unknown input "%s" in "%s"'
472 % (interface_name, input_name, presentation._fullname),
473 locator=assignment._locator, level=Issue.BETWEEN_TYPES)
475 definition = definitions.get(input_name) if definitions is not None else None
477 # Note: default value has already been assigned
480 values['inputs'][input_name] = coerce_parameter_value(context, assignment, definition,
484 def validate_required_inputs(context, presentation, assignment, definition, original_assignment,
485 interface_name, operation_name=None):
486 # The validation of the `required` field of inputs that belong to operations and interfaces
487 # (as opposed to topology template and workflow inputs) is done only in the parsing stage.
488 # This reasoning follows the TOSCA spirit, where anything that is declared as required in the
489 # type, must be assigned in the corresponding template.
490 input_definitions = definition.inputs
491 if input_definitions:
492 for input_name, input_definition in input_definitions.iteritems():
493 if input_definition.required:
494 prop = assignment.inputs.get(input_name) \
495 if ((assignment is not None) and (assignment.inputs is not None)) else None
496 value = prop.value if prop is not None else None
497 value = value.value if value is not None else None
499 if operation_name is not None:
500 context.validation.report(
501 'interface definition "%s" does not assign a value to a required'
502 ' operation input "%s.%s" in "%s"'
503 % (interface_name, operation_name, input_name, presentation._fullname),
504 locator=get_locator(original_assignment, presentation._locator),
505 level=Issue.BETWEEN_TYPES)
507 context.validation.report(
508 'interface definition "%s" does not assign a value to a required input'
510 % (interface_name, input_name, presentation._fullname),
511 locator=get_locator(original_assignment, presentation._locator),
512 level=Issue.BETWEEN_TYPES)
514 if operation_name is not None:
517 assignment_operations = assignment.operations
518 operation_definitions = definition._get_operations(context)
519 if operation_definitions:
520 for operation_name, operation_definition in operation_definitions.iteritems():
521 assignment_operation = assignment_operations.get(operation_name) \
522 if assignment_operations is not None else None
523 original_operation = \
524 original_assignment.operations.get(operation_name, original_assignment) \
525 if (original_assignment is not None) \
526 and (original_assignment.operations is not None) \
527 else original_assignment
528 validate_required_inputs(context, presentation, assignment_operation,
529 operation_definition, original_operation, interface_name,