azure/aria/aria-extension-cloudify/src/aria/tests/orchestrator/workflows/core/test_task_graph_into_execution_graph.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 from networkx import topological_sort, DiGraph
  17
  18 from aria.modeling import models
  19 from aria.orchestrator import context
  20 from aria.orchestrator.workflows import api
  21 from aria.orchestrator.workflows.core import graph_compiler
  22 from aria.orchestrator.workflows.executor import base
  23 from tests import mock
  24 from tests import storage
  25
  26
  27 def test_task_graph_into_execution_graph(tmpdir):
  28     interface_name = 'Standard'
  29     op1_name, op2_name, op3_name = 'create', 'configure', 'start'
  30     workflow_context = mock.context.simple(str(tmpdir))
  31     node = workflow_context.model.node.get_by_name(mock.models.DEPENDENCY_NODE_NAME)
  32     interface = mock.models.create_interface(
  33         node.service,
  34         interface_name,
  35         op1_name,
  36         operation_kwargs=dict(function='test')
  37     )
  38     interface.operations[op2_name] = mock.models.create_operation(op2_name)                         # pylint: disable=unsubscriptable-object
  39     interface.operations[op3_name] = mock.models.create_operation(op3_name)                         # pylint: disable=unsubscriptable-object
  40     node.interfaces[interface.name] = interface
  41     workflow_context.model.node.update(node)
  42
  43     def sub_workflow(name, **_):
  44         return api.task_graph.TaskGraph(name)
  45
  46     with context.workflow.current.push(workflow_context):
  47         test_task_graph = api.task.WorkflowTask(sub_workflow, name='test_task_graph')
  48         simple_before_task = api.task.OperationTask(
  49             node,
  50             interface_name=interface_name,
  51             operation_name=op1_name)
  52         simple_after_task = api.task.OperationTask(
  53             node,
  54             interface_name=interface_name,
  55             operation_name=op1_name)
  56
  57         inner_task_graph = api.task.WorkflowTask(sub_workflow, name='test_inner_task_graph')
  58         inner_task_1 = api.task.OperationTask(
  59             node,
  60             interface_name=interface_name,
  61             operation_name=op1_name)
  62         inner_task_2 = api.task.OperationTask(
  63             node,
  64             interface_name=interface_name,
  65             operation_name=op2_name)
  66         inner_task_3 = api.task.OperationTask(
  67             node,
  68             interface_name=interface_name,
  69             operation_name=op3_name)
  70         inner_task_graph.add_tasks(inner_task_1)
  71         inner_task_graph.add_tasks(inner_task_2)
  72         inner_task_graph.add_tasks(inner_task_3)
  73         inner_task_graph.add_dependency(inner_task_2, inner_task_1)
  74         inner_task_graph.add_dependency(inner_task_3, inner_task_1)
  75         inner_task_graph.add_dependency(inner_task_3, inner_task_2)
  76
  77     test_task_graph.add_tasks(simple_before_task)
  78     test_task_graph.add_tasks(simple_after_task)
  79     test_task_graph.add_tasks(inner_task_graph)
  80     test_task_graph.add_dependency(inner_task_graph, simple_before_task)
  81     test_task_graph.add_dependency(simple_after_task, inner_task_graph)
  82
  83     compiler = graph_compiler.GraphCompiler(workflow_context, base.StubTaskExecutor)
  84     compiler.compile(test_task_graph)
  85
  86     execution_tasks = topological_sort(_graph(workflow_context.execution.tasks))
  87
  88     assert len(execution_tasks) == 9
  89
  90     expected_tasks_names = [
  91         '{0}-Start'.format(test_task_graph.id),
  92         simple_before_task.id,
  93         '{0}-Start'.format(inner_task_graph.id),
  94         inner_task_1.id,
  95         inner_task_2.id,
  96         inner_task_3.id,
  97         '{0}-End'.format(inner_task_graph.id),
  98         simple_after_task.id,
  99         '{0}-End'.format(test_task_graph.id)
 100     ]
 101
 102     assert expected_tasks_names == [compiler._model_to_api_id[t.id] for t in execution_tasks]
 103     assert all(isinstance(task, models.Task) for task in execution_tasks)
 104     execution_tasks = iter(execution_tasks)
 105
 106     _assert_tasks(
 107         iter(execution_tasks),
 108         iter([simple_after_task, inner_task_1, inner_task_2, inner_task_3, simple_after_task])
 109     )
 110     storage.release_sqlite_storage(workflow_context.model)
 111
 112
 113 def _assert_tasks(execution_tasks, api_tasks):
 114     start_workflow_exec_task = next(execution_tasks)
 115     assert start_workflow_exec_task._stub_type == models.Task.START_WORKFLOW
 116
 117     before_exec_task = next(execution_tasks)
 118     simple_before_task = next(api_tasks)
 119     _assert_execution_is_api_task(before_exec_task, simple_before_task)
 120     assert before_exec_task.dependencies == [start_workflow_exec_task]
 121
 122     start_subworkflow_exec_task = next(execution_tasks)
 123     assert start_subworkflow_exec_task._stub_type == models.Task.START_SUBWROFKLOW
 124     assert start_subworkflow_exec_task.dependencies == [before_exec_task]
 125
 126     inner_exec_task_1 = next(execution_tasks)
 127     inner_task_1 = next(api_tasks)
 128     _assert_execution_is_api_task(inner_exec_task_1, inner_task_1)
 129     assert inner_exec_task_1.dependencies == [start_subworkflow_exec_task]
 130
 131     inner_exec_task_2 = next(execution_tasks)
 132     inner_task_2 = next(api_tasks)
 133     _assert_execution_is_api_task(inner_exec_task_2, inner_task_2)
 134     assert inner_exec_task_2.dependencies == [inner_exec_task_1]
 135
 136     inner_exec_task_3 = next(execution_tasks)
 137     inner_task_3 = next(api_tasks)
 138     _assert_execution_is_api_task(inner_exec_task_3, inner_task_3)
 139     assert sorted(inner_exec_task_3.dependencies) == sorted([inner_exec_task_1, inner_exec_task_2])
 140
 141     end_subworkflow_exec_task = next(execution_tasks)
 142     assert end_subworkflow_exec_task._stub_type == models.Task.END_SUBWORKFLOW
 143     assert end_subworkflow_exec_task.dependencies == [inner_exec_task_3]
 144
 145     after_exec_task = next(execution_tasks)
 146     simple_after_task = next(api_tasks)
 147     _assert_execution_is_api_task(after_exec_task, simple_after_task)
 148     assert after_exec_task.dependencies == [end_subworkflow_exec_task]
 149
 150     end_workflow_exec_task = next(execution_tasks)
 151     assert end_workflow_exec_task._stub_type == models.Task.END_WORKFLOW
 152     assert end_workflow_exec_task.dependencies == [after_exec_task]
 153
 154
 155 def _assert_execution_is_api_task(execution_task, api_task):
 156     assert execution_task.name == api_task.name
 157     assert execution_task.function == api_task.function
 158     assert execution_task.actor == api_task.actor
 159     assert execution_task.arguments == api_task.arguments
 160
 161
 162 def _get_task_by_name(task_name, graph):
 163     return graph.node[task_name]['task']
 164
 165
 166 def _graph(tasks):
 167     graph = DiGraph()
 168     for task in tasks:
 169         for dependency in task.dependencies:
 170             graph.add_edge(dependency, task)
 171
 172     return graph