You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@ariatosca.apache.org by mx...@apache.org on 2017/06/22 11:29:06 UTC
[4/4] incubator-ariatosca git commit: wip
wip
Project: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/commit/827230da
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/tree/827230da
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/diff/827230da
Branch: refs/heads/ARIA-284-Cleanup-and-optimize-the-task-execution
Commit: 827230da2f515c2c087d847a68137f27241de161
Parents: 75112ab
Author: max-orlov <ma...@gigaspaces.com>
Authored: Wed Jun 21 12:41:33 2017 +0300
Committer: max-orlov <ma...@gigaspaces.com>
Committed: Thu Jun 22 14:28:55 2017 +0300
----------------------------------------------------------------------
aria/modeling/orchestration.py | 2 -
aria/orchestrator/context/workflow.py | 19 --
aria/orchestrator/workflow_runner.py | 5 +-
aria/orchestrator/workflows/core/compile.py | 198 ++++++++++---------
aria/orchestrator/workflows/core/engine.py | 110 +++++++----
tests/orchestrator/context/__init__.py | 2 +-
tests/orchestrator/context/test_serialize.py | 2 +-
.../orchestrator/execution_plugin/test_local.py | 2 +-
tests/orchestrator/execution_plugin/test_ssh.py | 3 +-
.../orchestrator/workflows/core/test_engine.py | 2 +-
.../orchestrator/workflows/core/test_events.py | 7 +-
.../test_task_graph_into_execution_graph.py | 19 +-
.../executor/test_process_executor_extension.py | 2 +-
.../test_process_executor_tracked_changes.py | 2 +-
14 files changed, 198 insertions(+), 177 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/827230da/aria/modeling/orchestration.py
----------------------------------------------------------------------
diff --git a/aria/modeling/orchestration.py b/aria/modeling/orchestration.py
index 276b68e..5b02d1b 100644
--- a/aria/modeling/orchestration.py
+++ b/aria/modeling/orchestration.py
@@ -306,7 +306,6 @@ class TaskBase(mixins.ModelMixin):
ended_at = Column(DateTime, default=None)
attempts_count = Column(Integer, default=1)
- _api_id = Column(String)
_executor = Column(PickleType)
_context_cls = Column(PickleType)
_stub_type = Column(Enum(*STUB_TYPES))
@@ -442,7 +441,6 @@ class TaskBase(mixins.ModelMixin):
'plugin': api_task.plugin,
'function': api_task.function,
'arguments': api_task.arguments,
- '_api_id': api_task.id,
'_context_cls': api_task._context_cls,
'_executor': executor,
}
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/827230da/aria/orchestrator/context/workflow.py
----------------------------------------------------------------------
diff --git a/aria/orchestrator/context/workflow.py b/aria/orchestrator/context/workflow.py
index adcd635..18334f3 100644
--- a/aria/orchestrator/context/workflow.py
+++ b/aria/orchestrator/context/workflow.py
@@ -20,8 +20,6 @@ Workflow and operation contexts
import threading
from contextlib import contextmanager
-from networkx import DiGraph
-
from .exceptions import ContextException
from .common import BaseContext
@@ -96,23 +94,6 @@ class WorkflowContext(BaseContext):
)
@property
- def _graph(self):
- # Constructing a graph with only not ended nodes
- if self._execution_graph is None:
- graph = DiGraph()
- for task in self.execution.tasks:
- if task.has_ended():
- continue
- for dependency in task.dependencies:
- if dependency.has_ended():
- continue
- graph.add_edge(dependency, task)
-
- self._execution_graph = graph
-
- return self._execution_graph
-
- @property
@contextmanager
def persist_changes(self):
yield
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/827230da/aria/orchestrator/workflow_runner.py
----------------------------------------------------------------------
diff --git a/aria/orchestrator/workflow_runner.py b/aria/orchestrator/workflow_runner.py
index 3ccb1ee..b3f100d 100644
--- a/aria/orchestrator/workflow_runner.py
+++ b/aria/orchestrator/workflow_runner.py
@@ -96,8 +96,9 @@ class WorkflowRunner(object):
if not self._is_resume:
workflow_fn = self._get_workflow_fn()
- tasks_graph = workflow_fn(ctx=self._workflow_context, **execution_inputs_dict)
- compile.create_execution_tasks(self._workflow_context, tasks_graph, executor.__class__)
+ self._tasks_graph = workflow_fn(ctx=self._workflow_context, **execution_inputs_dict)
+ compile.GraphCompiler(self._workflow_context, executor.__class__).compile(
+ self._tasks_graph)
self._engine = engine.Engine(executors={executor.__class__: executor})
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/827230da/aria/orchestrator/workflows/core/compile.py
----------------------------------------------------------------------
diff --git a/aria/orchestrator/workflows/core/compile.py b/aria/orchestrator/workflows/core/compile.py
index 932268a..83de22c 100644
--- a/aria/orchestrator/workflows/core/compile.py
+++ b/aria/orchestrator/workflows/core/compile.py
@@ -18,99 +18,105 @@ from ....modeling import models
from .. import executor, api
-def create_execution_tasks(ctx, task_graph, default_executor):
- execution = ctx.execution
- _construct_execution_tasks(execution, task_graph, default_executor)
- ctx.model.execution.update(execution)
- return execution.tasks
-
-
-def _construct_execution_tasks(execution,
- task_graph,
- default_executor,
- stub_executor=executor.base.StubTaskExecutor,
- start_stub_type=models.Task.START_WORKFLOW,
- end_stub_type=models.Task.END_WORKFLOW,
- depends_on=()):
- """
- Translates the user graph to the execution graph
- :param task_graph: The user's graph
- :param start_stub_type: internal use
- :param end_stub_type: internal use
- :param depends_on: internal use
- """
- depends_on = list(depends_on)
-
- # Insert start marker
- start_task = models.Task(execution=execution,
- dependencies=depends_on,
- _api_id=_start_graph_suffix(task_graph.id),
- _stub_type=start_stub_type,
- _executor=stub_executor)
-
- for task in task_graph.topological_order(reverse=True):
- operation_dependencies = _get_tasks_from_dependencies(
- execution, task_graph.get_dependencies(task), [start_task])
-
- if isinstance(task, api.task.OperationTask):
- models.Task.from_api_task(api_task=task,
- executor=default_executor,
- dependencies=operation_dependencies)
-
- elif isinstance(task, api.task.WorkflowTask):
- # Build the graph recursively while adding start and end markers
- _construct_execution_tasks(
- execution=execution,
- task_graph=task,
- default_executor=default_executor,
- stub_executor=stub_executor,
- start_stub_type=models.Task.START_SUBWROFKLOW,
- end_stub_type=models.Task.END_SUBWORKFLOW,
- depends_on=operation_dependencies
- )
- elif isinstance(task, api.task.StubTask):
- models.Task(execution=execution,
- dependencies=operation_dependencies,
- _api_id=task.id,
- _executor=stub_executor,
- _stub_type=models.Task.STUB,
- )
- else:
- raise RuntimeError('Undefined state')
-
- # Insert end marker
- models.Task(dependencies=_get_non_dependent_tasks(execution) or [start_task],
- execution=execution,
- _api_id=_end_graph_suffix(task_graph.id),
- _executor=stub_executor,
- _stub_type=end_stub_type)
-
-
-def _start_graph_suffix(api_id):
- return '{0}-Start'.format(api_id)
-
-
-def _end_graph_suffix(api_id):
- return '{0}-End'.format(api_id)
-
-
-def _get_non_dependent_tasks(execution):
- tasks_with_dependencies = set()
- for task in execution.tasks:
- tasks_with_dependencies.update(task.dependencies)
- return list(set(execution.tasks) - set(tasks_with_dependencies))
-
-
-def _get_tasks_from_dependencies(execution, dependencies, default=()):
- """
- Returns task list from dependencies.
- """
- tasks = []
- for dependency in dependencies:
- if getattr(dependency, 'actor', False):
- # This is
- dependency_name = dependency.id
- else:
- dependency_name = _end_graph_suffix(dependency.id)
- tasks.extend(task for task in execution.tasks if task._api_id == dependency_name)
- return tasks or default
+# TODO: is class really needed?
+
+class GraphCompiler(object):
+ def __init__(self, ctx, default_executor):
+ self._ctx = ctx
+ self._default_executor = default_executor
+ self._stub_executor = executor.base.StubTaskExecutor
+ self._model_to_api_id = {}
+
+ def compile(self,
+ task_graph,
+ start_stub_type=models.Task.START_WORKFLOW,
+ end_stub_type=models.Task.END_WORKFLOW,
+ depends_on=()):
+ """
+ Translates the user graph to the execution graph
+ :param task_graph: The user's graph
+ :param start_stub_type: internal use
+ :param end_stub_type: internal use
+ :param depends_on: internal use
+ """
+ task_graph = task_graph or self._task_graph
+ depends_on = list(depends_on)
+
+ # Insert start marker
+ start_task = self._create_stub_task(
+ start_stub_type, depends_on, self._start_graph_suffix(task_graph.id), task_graph.name,
+ )
+
+ for task in task_graph.topological_order(reverse=True):
+ dependencies = \
+ (self._get_tasks_from_dependencies(task_graph.get_dependencies(task))
+ or [start_task])
+
+ if isinstance(task, api.task.OperationTask):
+ self._create_operation_task(task, dependencies)
+
+ elif isinstance(task, api.task.WorkflowTask):
+ # Build the graph recursively while adding start and end markers
+ self.compile(
+ task, models.Task.START_SUBWROFKLOW, models.Task.END_SUBWORKFLOW, dependencies
+ )
+ elif isinstance(task, api.task.StubTask):
+ self._create_stub_task(models.Task.STUB, dependencies, task.id)
+ else:
+ raise RuntimeError('Undefined state')
+
+ # Insert end marker
+ self._create_stub_task(
+ end_stub_type,
+ self._get_non_dependent_tasks(self._ctx.execution) or [start_task],
+ self._end_graph_suffix(task_graph.id),
+ task_graph.name
+ )
+
+ def _create_stub_task(self, stub_type, dependencies, api_id, name=None):
+ model_task = models.Task(
+ name=name,
+ dependencies=dependencies,
+ execution=self._ctx.execution,
+ _executor=self._stub_executor,
+ _stub_type=stub_type)
+ self._ctx.model.task.put(model_task)
+ self._model_to_api_id[model_task.id] = api_id
+ return model_task
+
+ def _create_operation_task(self, api_task, dependencies):
+ model_task = models.Task.from_api_task(
+ api_task, self._default_executor, dependencies=dependencies)
+ self._ctx.model.task.put(model_task)
+ self._model_to_api_id[model_task.id] = api_task.id
+ return model_task
+
+ @staticmethod
+ def _start_graph_suffix(api_id):
+ return '{0}-Start'.format(api_id)
+
+ @staticmethod
+ def _end_graph_suffix(api_id):
+ return '{0}-End'.format(api_id)
+
+ @staticmethod
+ def _get_non_dependent_tasks(execution):
+ tasks_with_dependencies = set()
+ for task in execution.tasks:
+ tasks_with_dependencies.update(task.dependencies)
+ return list(set(execution.tasks) - set(tasks_with_dependencies))
+
+ def _get_tasks_from_dependencies(self, dependencies):
+ """
+ Returns task list from dependencies.
+ """
+ tasks = []
+ for dependency in dependencies:
+ if getattr(dependency, 'actor', False):
+ # This is
+ dependency_name = dependency.id
+ else:
+ dependency_name = self._end_graph_suffix(dependency.id)
+ tasks.extend(task for task in self._ctx.execution.tasks
+ if self._model_to_api_id.get(task.id, None) == dependency_name)
+ return tasks
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/827230da/aria/orchestrator/workflows/core/engine.py
----------------------------------------------------------------------
diff --git a/aria/orchestrator/workflows/core/engine.py b/aria/orchestrator/workflows/core/engine.py
index d5a6e70..f594e36 100644
--- a/aria/orchestrator/workflows/core/engine.py
+++ b/aria/orchestrator/workflows/core/engine.py
@@ -45,22 +45,23 @@ class Engine(logger.LoggerMixin):
"""
execute the workflow
"""
- executing_tasks = []
-
if resuming:
events.on_resume_workflow_signal.send(ctx)
+ task_tracker = _TasksTracker(ctx)
try:
events.start_workflow_signal.send(ctx)
while True:
cancel = self._is_cancel(ctx)
if cancel:
break
- for task in self._ended_tasks(ctx, executing_tasks):
- self._handle_ended_tasks(ctx, task, executing_tasks)
- for task in self._executable_tasks(ctx):
- self._handle_executable_task(ctx, task, executing_tasks)
- if self._all_tasks_consumed(ctx):
+ for task in task_tracker.ended_tasks:
+ self._handle_ended_tasks(task)
+ task_tracker.finished_(task)
+ for task in task_tracker.executable_tasks:
+ task_tracker.executing_(task)
+ self._handle_executable_task(ctx, task)
+ if task_tracker.all_tasks_consumed:
break
else:
time.sleep(0.1)
@@ -86,34 +87,7 @@ class Engine(logger.LoggerMixin):
execution = ctx.model.execution.refresh(ctx.execution)
return execution.status in (models.Execution.CANCELLING, models.Execution.CANCELLED)
- def _executable_tasks(self, ctx):
- now = datetime.utcnow()
- return (
- task for task in self._tasks_iter(ctx)
- if task.is_waiting() and task.due_at <= now and \
- not self._task_has_dependencies(ctx, task)
- )
-
- @staticmethod
- def _ended_tasks(ctx, executing_tasks):
- for task in executing_tasks:
- if task.has_ended() and task in ctx._graph:
- yield task
-
- @staticmethod
- def _task_has_dependencies(ctx, task):
- return len(ctx._graph.pred.get(task, [])) > 0
-
- @staticmethod
- def _all_tasks_consumed(ctx):
- return len(ctx._graph.node) == 0
-
- @staticmethod
- def _tasks_iter(ctx):
- for task in ctx.execution.tasks:
- yield ctx.model.task.refresh(task)
-
- def _handle_executable_task(self, ctx, task, executing_tasks):
+ def _handle_executable_task(self, ctx, task):
task_executor = self._executors[task._executor]
# If the task is a stub, a default context is provided, else it should hold the context cls
@@ -129,16 +103,70 @@ class Engine(logger.LoggerMixin):
name=task.name
)
- executing_tasks.append(task)
-
if not task._stub_type:
events.sent_task_signal.send(op_ctx)
task_executor.execute(op_ctx)
@staticmethod
- def _handle_ended_tasks(ctx, task, executing_tasks):
- executing_tasks.remove(task)
+ def _handle_ended_tasks(task):
if task.status == models.Task.FAILED and not task.ignore_failure:
raise exceptions.ExecutorException('Workflow failed')
- else:
- ctx._graph.remove_node(task)
+
+
+class _TasksTracker(object):
+ def __init__(self, ctx):
+ self._ctx = ctx
+ self._tasks = ctx.execution.tasks
+ self._executed_tasks = [task for task in self._tasks if task.has_ended()]
+ self._executable_tasks = list(set(self._tasks) - set(self._executed_tasks))
+ self._executing_tasks = []
+
+ @property
+ def all_tasks_consumed(self):
+ return len(self._executed_tasks) == len(self._tasks) and len(self._executing_tasks) == 0
+
+ def executing_(self, task):
+ # Task executing could be retrying (thus removed and added earlier)
+ if task not in self._executing_tasks:
+ self._executable_tasks.remove(task)
+ self._executing_tasks.append(task)
+
+ def finished_(self, task):
+ self._executing_tasks.remove(task)
+ self._executed_tasks.append(task)
+
+ @property
+ def ended_tasks(self):
+ for task in self.executing_tasks:
+ if task.has_ended():
+ yield task
+
+ @property
+ def executable_tasks(self):
+ now = datetime.utcnow()
+ # we need both list since retrying task are in the executing task list.
+ for task in self._update_tasks(self._executing_tasks + self._executable_tasks):
+ if all([task.is_waiting(),
+ task.due_at <= now,
+ all(dependency in self._executed_tasks for dependency in task.dependencies)
+ ]):
+ yield task
+
+ @property
+ def executing_tasks(self):
+ for task in self._update_tasks(self._executing_tasks):
+ yield task
+
+ @property
+ def executed_tasks(self):
+ for task in self._update_tasks(self._executed_tasks):
+ yield task
+
+ @property
+ def tasks(self):
+ for task in self._update_tasks(self._tasks):
+ yield task
+
+ def _update_tasks(self, tasks):
+ for task in tasks:
+ yield self._ctx.model.task.refresh(task)
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/827230da/tests/orchestrator/context/__init__.py
----------------------------------------------------------------------
diff --git a/tests/orchestrator/context/__init__.py b/tests/orchestrator/context/__init__.py
index 086a066..752706e 100644
--- a/tests/orchestrator/context/__init__.py
+++ b/tests/orchestrator/context/__init__.py
@@ -26,7 +26,7 @@ def op_path(func, module_path=None):
def execute(workflow_func, workflow_context, executor):
graph = workflow_func(ctx=workflow_context)
- compile.create_execution_tasks(workflow_context, graph, executor.__class__)
+ compile.GraphCompiler(workflow_context, executor.__class__).compile(graph)
eng = engine.Engine(executors={executor.__class__: executor})
eng.execute(workflow_context)
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/827230da/tests/orchestrator/context/test_serialize.py
----------------------------------------------------------------------
diff --git a/tests/orchestrator/context/test_serialize.py b/tests/orchestrator/context/test_serialize.py
index 5db5b63..b7335a0 100644
--- a/tests/orchestrator/context/test_serialize.py
+++ b/tests/orchestrator/context/test_serialize.py
@@ -48,7 +48,7 @@ def test_serialize_operation_context(context, executor, tmpdir):
context.model.node.update(node)
graph = _mock_workflow(ctx=context) # pylint: disable=no-value-for-parameter
- compile.create_execution_tasks(context, graph, executor.__class__)
+ compile.GraphCompiler(context, executor.__class__).compile(graph)
eng = engine.Engine({executor.__class__: executor})
eng.execute(context)
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/827230da/tests/orchestrator/execution_plugin/test_local.py
----------------------------------------------------------------------
diff --git a/tests/orchestrator/execution_plugin/test_local.py b/tests/orchestrator/execution_plugin/test_local.py
index 1695320..ab6310c 100644
--- a/tests/orchestrator/execution_plugin/test_local.py
+++ b/tests/orchestrator/execution_plugin/test_local.py
@@ -500,7 +500,7 @@ if __name__ == '__main__':
arguments=arguments))
return graph
tasks_graph = mock_workflow(ctx=workflow_context) # pylint: disable=no-value-for-parameter
- compile.create_execution_tasks(workflow_context, tasks_graph, executor.__class__)
+ compile.GraphCompiler(workflow_context, executor.__class__).compile(tasks_graph)
eng = engine.Engine({executor.__class__: executor})
eng.execute(workflow_context)
return workflow_context.model.node.get_by_name(
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/827230da/tests/orchestrator/execution_plugin/test_ssh.py
----------------------------------------------------------------------
diff --git a/tests/orchestrator/execution_plugin/test_ssh.py b/tests/orchestrator/execution_plugin/test_ssh.py
index fb1dc09..13ad1a3 100644
--- a/tests/orchestrator/execution_plugin/test_ssh.py
+++ b/tests/orchestrator/execution_plugin/test_ssh.py
@@ -254,8 +254,7 @@ class TestWithActualSSHServer(object):
graph.sequence(*ops)
return graph
tasks_graph = mock_workflow(ctx=self._workflow_context) # pylint: disable=no-value-for-parameter
- compile.create_execution_tasks(
- self._workflow_context, tasks_graph, self._executor.__class__)
+ compile.GraphCompiler(self._workflow_context, self._executor.__class__).compile(tasks_graph)
eng = engine.Engine({self._executor.__class__: self._executor})
eng.execute(self._workflow_context)
return self._workflow_context.model.node.get_by_name(
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/827230da/tests/orchestrator/workflows/core/test_engine.py
----------------------------------------------------------------------
diff --git a/tests/orchestrator/workflows/core/test_engine.py b/tests/orchestrator/workflows/core/test_engine.py
index b77d284..7275723 100644
--- a/tests/orchestrator/workflows/core/test_engine.py
+++ b/tests/orchestrator/workflows/core/test_engine.py
@@ -50,7 +50,7 @@ class BaseTest(object):
@staticmethod
def _engine(workflow_func, workflow_context, executor):
graph = workflow_func(ctx=workflow_context)
- compile.create_execution_tasks(workflow_context, graph, executor.__class__)
+ compile.GraphCompiler(workflow_context, executor.__class__).compile(graph)
return engine.Engine(executors={executor.__class__: executor})
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/827230da/tests/orchestrator/workflows/core/test_events.py
----------------------------------------------------------------------
diff --git a/tests/orchestrator/workflows/core/test_events.py b/tests/orchestrator/workflows/core/test_events.py
index 2b82443..32a6b7b 100644
--- a/tests/orchestrator/workflows/core/test_events.py
+++ b/tests/orchestrator/workflows/core/test_events.py
@@ -113,10 +113,9 @@ def run_operation_on_node(ctx, op_name, interface_name):
operation_name=op_name,
operation_kwargs=dict(function='{name}.{func.__name__}'.format(name=__name__, func=func)))
node.interfaces[interface.name] = interface
- compile.create_execution_tasks(
- ctx,
- single_operation_workflow(ctx, node=node, interface_name=interface_name, op_name=op_name),
- ThreadExecutor)
+ compile.GraphCompiler(ctx, ThreadExecutor).compile(
+ single_operation_workflow(ctx, node=node, interface_name=interface_name, op_name=op_name)
+ )
eng = engine.Engine(executors={ThreadExecutor: ThreadExecutor()})
eng.execute(ctx)
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/827230da/tests/orchestrator/workflows/core/test_task_graph_into_execution_graph.py
----------------------------------------------------------------------
diff --git a/tests/orchestrator/workflows/core/test_task_graph_into_execution_graph.py b/tests/orchestrator/workflows/core/test_task_graph_into_execution_graph.py
index f5fb17a..3d47d54 100644
--- a/tests/orchestrator/workflows/core/test_task_graph_into_execution_graph.py
+++ b/tests/orchestrator/workflows/core/test_task_graph_into_execution_graph.py
@@ -13,7 +13,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-from networkx import topological_sort
+from networkx import topological_sort, DiGraph
from aria.modeling import models
from aria.orchestrator import context
@@ -65,9 +65,10 @@ def test_task_graph_into_execution_graph(tmpdir):
test_task_graph.add_dependency(inner_task_graph, simple_before_task)
test_task_graph.add_dependency(simple_after_task, inner_task_graph)
- compile.create_execution_tasks(workflow_context, test_task_graph, base.StubTaskExecutor)
+ graph_compiler = compile.GraphCompiler(workflow_context, base.StubTaskExecutor)
+ graph_compiler.compile(test_task_graph)
- execution_tasks = topological_sort(workflow_context._graph)
+ execution_tasks = topological_sort(_graph(workflow_context.execution.tasks))
assert len(execution_tasks) == 7
@@ -81,7 +82,7 @@ def test_task_graph_into_execution_graph(tmpdir):
'{0}-End'.format(test_task_graph.id)
]
- assert expected_tasks_names == [t._api_id for t in execution_tasks]
+ assert expected_tasks_names == [graph_compiler._model_to_api_id[t.id] for t in execution_tasks]
assert all(isinstance(task, models.Task) for task in execution_tasks)
execution_tasks = iter(execution_tasks)
@@ -97,7 +98,6 @@ def test_task_graph_into_execution_graph(tmpdir):
def _assert_execution_is_api_task(execution_task, api_task):
- assert execution_task._api_id == api_task.id
assert execution_task.name == api_task.name
assert execution_task.function == api_task.function
assert execution_task.actor == api_task.actor
@@ -106,3 +106,12 @@ def _assert_execution_is_api_task(execution_task, api_task):
def _get_task_by_name(task_name, graph):
return graph.node[task_name]['task']
+
+
+def _graph(tasks):
+ graph = DiGraph()
+ for task in tasks:
+ for dependency in task.dependencies:
+ graph.add_edge(dependency, task)
+
+ return graph
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/827230da/tests/orchestrator/workflows/executor/test_process_executor_extension.py
----------------------------------------------------------------------
diff --git a/tests/orchestrator/workflows/executor/test_process_executor_extension.py b/tests/orchestrator/workflows/executor/test_process_executor_extension.py
index ba98c4f..aa08685 100644
--- a/tests/orchestrator/workflows/executor/test_process_executor_extension.py
+++ b/tests/orchestrator/workflows/executor/test_process_executor_extension.py
@@ -57,7 +57,7 @@ def test_decorate_extension(context, executor):
graph.add_tasks(task)
return graph
graph = mock_workflow(ctx=context) # pylint: disable=no-value-for-parameter
- compile.create_execution_tasks(context, graph, executor.__class__)
+ compile.GraphCompiler(context, executor.__class__).compile(graph)
eng = engine.Engine({executor.__class__: executor})
eng.execute(context)
out = get_node(context).attributes.get('out').value
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/827230da/tests/orchestrator/workflows/executor/test_process_executor_tracked_changes.py
----------------------------------------------------------------------
diff --git a/tests/orchestrator/workflows/executor/test_process_executor_tracked_changes.py b/tests/orchestrator/workflows/executor/test_process_executor_tracked_changes.py
index 2f1c325..7102b13 100644
--- a/tests/orchestrator/workflows/executor/test_process_executor_tracked_changes.py
+++ b/tests/orchestrator/workflows/executor/test_process_executor_tracked_changes.py
@@ -107,7 +107,7 @@ def _run_workflow(context, executor, op_func, arguments=None):
graph.add_tasks(task)
return graph
graph = mock_workflow(ctx=context) # pylint: disable=no-value-for-parameter
- compile.create_execution_tasks(context, graph, executor.__class__)
+ compile.GraphCompiler(context, executor.__class__).compile(graph)
eng = engine.Engine({executor.__class__: executor})
eng.execute(context)
out = context.model.node.get_by_name(mock.models.DEPENDENCY_NODE_NAME).attributes.get('out')