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/18 14:25:29 UTC
incubator-ariatosca git commit: moved things around
Repository: incubator-ariatosca
Updated Branches:
refs/heads/ARIA-278-Remove-core-tasks b314f388b -> e3977fd5d
moved things around
Project: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/commit/e3977fd5
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/tree/e3977fd5
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/diff/e3977fd5
Branch: refs/heads/ARIA-278-Remove-core-tasks
Commit: e3977fd5d8eae5e5dd9af1d3faa72560dc979f0e
Parents: b314f38
Author: max-orlov <ma...@gigaspaces.com>
Authored: Sun Jun 18 17:25:25 2017 +0300
Committer: max-orlov <ma...@gigaspaces.com>
Committed: Sun Jun 18 17:25:25 2017 +0300
----------------------------------------------------------------------
aria/orchestrator/workflow_runner.py | 97 +------------------
aria/orchestrator/workflows/core/engine.py | 99 +++++++++++++++++++-
tests/orchestrator/context/__init__.py | 2 +-
tests/orchestrator/context/test_serialize.py | 4 +-
.../orchestrator/execution_plugin/test_local.py | 4 +-
tests/orchestrator/execution_plugin/test_ssh.py | 4 +-
tests/orchestrator/test_workflow_runner.py | 18 ++--
.../orchestrator/workflows/core/test_engine.py | 5 +-
.../orchestrator/workflows/core/test_events.py | 5 +-
.../test_task_graph_into_execution_graph.py | 10 +-
.../executor/test_process_executor_extension.py | 4 +-
.../test_process_executor_tracked_changes.py | 4 +-
12 files changed, 126 insertions(+), 130 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/e3977fd5/aria/orchestrator/workflow_runner.py
----------------------------------------------------------------------
diff --git a/aria/orchestrator/workflow_runner.py b/aria/orchestrator/workflow_runner.py
index c4d8666..c30ec4b 100644
--- a/aria/orchestrator/workflow_runner.py
+++ b/aria/orchestrator/workflow_runner.py
@@ -24,10 +24,8 @@ from datetime import datetime
from . import exceptions
from .context.workflow import WorkflowContext
from .workflows import builtin
-from .workflows.core.engine import Engine
+from .workflows.core import engine
from .workflows.executor.process import ProcessExecutor
-from .workflows.executor.base import StubTaskExecutor
-from .workflows.api import task as api_task
from ..modeling import models
from ..modeling import utils as modeling_utils
from ..utils.imports import import_fullname
@@ -89,12 +87,12 @@ class WorkflowRunner(object):
execution_inputs_dict = dict(inp.unwrapped for inp in self.execution.inputs.values())
self._tasks_graph = workflow_fn(ctx=self._workflow_context, **execution_inputs_dict)
- construct_execution_tasks(self.execution, self._tasks_graph, executor.__class__)
+ engine.construct_execution_tasks(self.execution, self._tasks_graph, executor.__class__)
# Update the state
self._model_storage.execution.update(execution)
- self._engine = Engine(executor=executor)
+ self._engine = engine.Engine(default_executor=executor)
@property
def execution_id(self):
@@ -172,92 +170,3 @@ class WorkflowRunner(object):
self._workflow_name, workflow.function))
return workflow_fn
-
-
-def construct_execution_tasks(execution,
- task_graph,
- default_executor,
- stub_executor=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(api_id=_start_graph_suffix(task_graph.id),
- _executor=stub_executor,
- execution=execution,
- stub_type=start_stub_type,
- dependencies=depends_on)
-
- 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(api_id=task.id,
- _executor=stub_executor,
- execution=execution,
- stub_type=models.Task.STUB,
- dependencies=operation_dependencies)
- else:
- raise RuntimeError('Undefined state')
-
- # Insert end marker
- models.Task(api_id=_end_graph_suffix(task_graph.id),
- _executor=stub_executor,
- execution=execution,
- stub_type=end_stub_type,
- dependencies=_get_non_dependent_tasks(execution) or [start_task])
-
-
-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):
- dependency_tasks = set()
- for task in execution.tasks:
- dependency_tasks.update(task.dependencies)
- return list(set(execution.tasks) - set(dependency_tasks))
-
-
-def _get_tasks_from_dependencies(execution, dependencies, default=()):
- """
- Returns task list from dependencies.
- """
- tasks = []
- for dependency in dependencies:
- if getattr(dependency, 'actor', False):
- 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
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/e3977fd5/aria/orchestrator/workflows/core/engine.py
----------------------------------------------------------------------
diff --git a/aria/orchestrator/workflows/core/engine.py b/aria/orchestrator/workflows/core/engine.py
index e547aa1..cec561f 100644
--- a/aria/orchestrator/workflows/core/engine.py
+++ b/aria/orchestrator/workflows/core/engine.py
@@ -25,7 +25,7 @@ from aria.modeling import models
from aria.orchestrator import events
from aria.orchestrator.context import operation
-from .. import exceptions
+from .. import exceptions, executor, api
# Import required so all signals are registered
from . import events_handler # pylint: disable=unused-import
@@ -35,9 +35,9 @@ class Engine(logger.LoggerMixin):
The workflow engine. Executes workflows
"""
- def __init__(self, executor, **kwargs):
+ def __init__(self, default_executor, **kwargs):
super(Engine, self).__init__(**kwargs)
- self._executors = {executor.__class__: executor}
+ self._executors = {default_executor.__class__: default_executor}
self._executing_tasks = []
def execute(self, ctx):
@@ -109,7 +109,7 @@ class Engine(logger.LoggerMixin):
def _handle_executable_task(self, ctx, task):
if task._executor not in self._executors:
self._executors[task._executor] = task._executor()
- executor = self._executors[task._executor]
+ task_executor = self._executors[task._executor]
context_cls = task._context_cls or operation.BaseOperationContext
op_ctx = context_cls(
@@ -127,7 +127,7 @@ class Engine(logger.LoggerMixin):
if not task.stub_type:
events.sent_task_signal.send(op_ctx)
- executor.execute(op_ctx)
+ task_executor.execute(op_ctx)
def _handle_ended_tasks(self, ctx, task):
self._executing_tasks.remove(task)
@@ -135,3 +135,92 @@ class Engine(logger.LoggerMixin):
raise exceptions.ExecutorException('Workflow failed')
else:
ctx._graph.remove_node(task)
+
+
+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(api_id=_start_graph_suffix(task_graph.id),
+ _executor=stub_executor,
+ execution=execution,
+ stub_type=start_stub_type,
+ dependencies=depends_on)
+
+ 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(api_id=task.id,
+ _executor=stub_executor,
+ execution=execution,
+ stub_type=models.Task.STUB,
+ dependencies=operation_dependencies)
+ else:
+ raise RuntimeError('Undefined state')
+
+ # Insert end marker
+ models.Task(api_id=_end_graph_suffix(task_graph.id),
+ _executor=stub_executor,
+ execution=execution,
+ stub_type=end_stub_type,
+ dependencies=_get_non_dependent_tasks(execution) or [start_task])
+
+
+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):
+ dependency_tasks = set()
+ for task in execution.tasks:
+ dependency_tasks.update(task.dependencies)
+ return list(set(execution.tasks) - set(dependency_tasks))
+
+
+def _get_tasks_from_dependencies(execution, dependencies, default=()):
+ """
+ Returns task list from dependencies.
+ """
+ tasks = []
+ for dependency in dependencies:
+ if getattr(dependency, 'actor', False):
+ 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
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/e3977fd5/tests/orchestrator/context/__init__.py
----------------------------------------------------------------------
diff --git a/tests/orchestrator/context/__init__.py b/tests/orchestrator/context/__init__.py
index 60ce234..abe92b9 100644
--- a/tests/orchestrator/context/__init__.py
+++ b/tests/orchestrator/context/__init__.py
@@ -27,7 +27,7 @@ def op_path(func, module_path=None):
def execute(workflow_func, workflow_context, executor):
graph = workflow_func(ctx=workflow_context)
- workflow_runner.construct_execution_tasks(workflow_context.execution, graph, executor.__class__)
+ engine.construct_execution_tasks(workflow_context.execution, graph, executor.__class__)
workflow_context.execution = workflow_context.execution
eng = engine.Engine(executor)
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/e3977fd5/tests/orchestrator/context/test_serialize.py
----------------------------------------------------------------------
diff --git a/tests/orchestrator/context/test_serialize.py b/tests/orchestrator/context/test_serialize.py
index 2730ef4..84d8952 100644
--- a/tests/orchestrator/context/test_serialize.py
+++ b/tests/orchestrator/context/test_serialize.py
@@ -18,7 +18,7 @@ import pytest
from aria.orchestrator.workflows import api
from aria.orchestrator.workflows.core import engine
from aria.orchestrator.workflows.executor import process
-from aria.orchestrator import workflow, operation, workflow_runner
+from aria.orchestrator import workflow, operation
import tests
from tests import mock
from tests import storage
@@ -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
- workflow_runner.construct_execution_tasks(context.execution, graph, executor.__class__)
+ engine.construct_execution_tasks(context.execution, graph, executor.__class__)
context.execution = context.execution
eng = engine.Engine(executor)
eng.execute(context)
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/e3977fd5/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 a2265c3..cb1503b 100644
--- a/tests/orchestrator/execution_plugin/test_local.py
+++ b/tests/orchestrator/execution_plugin/test_local.py
@@ -19,7 +19,7 @@ import os
import pytest
from aria import workflow
-from aria.orchestrator import events, workflow_runner
+from aria.orchestrator import events
from aria.orchestrator.workflows import api
from aria.orchestrator.workflows.exceptions import ExecutorException
from aria.orchestrator.exceptions import TaskAbortException, TaskRetryException
@@ -500,7 +500,7 @@ if __name__ == '__main__':
arguments=arguments))
return graph
tasks_graph = mock_workflow(ctx=workflow_context) # pylint: disable=no-value-for-parameter
- workflow_runner.construct_execution_tasks(
+ engine.construct_execution_tasks(
workflow_context.execution, tasks_graph, executor.__class__)
workflow_context.execution = workflow_context.execution
eng = engine.Engine(executor)
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/e3977fd5/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 e7221f2..2eb3c0b 100644
--- a/tests/orchestrator/execution_plugin/test_ssh.py
+++ b/tests/orchestrator/execution_plugin/test_ssh.py
@@ -25,7 +25,7 @@ from fabric.contrib import files
from fabric import context_managers
from aria.modeling import models
-from aria.orchestrator import events, workflow_runner
+from aria.orchestrator import events
from aria.orchestrator import workflow
from aria.orchestrator.workflows import api
from aria.orchestrator.workflows.executor import process
@@ -254,7 +254,7 @@ class TestWithActualSSHServer(object):
graph.sequence(*ops)
return graph
tasks_graph = mock_workflow(ctx=self._workflow_context) # pylint: disable=no-value-for-parameter
- workflow_runner.construct_execution_tasks(
+ engine.construct_execution_tasks(
self._workflow_context.execution, tasks_graph, self._executor.__class__)
self._workflow_context.execution = self._workflow_context.execution
eng = engine.Engine(self._executor)
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/e3977fd5/tests/orchestrator/test_workflow_runner.py
----------------------------------------------------------------------
diff --git a/tests/orchestrator/test_workflow_runner.py b/tests/orchestrator/test_workflow_runner.py
index 2ae1e59..11bc7b9 100644
--- a/tests/orchestrator/test_workflow_runner.py
+++ b/tests/orchestrator/test_workflow_runner.py
@@ -96,20 +96,20 @@ def test_default_executor(request):
# validates the ProcessExecutor is used by the workflow runner by default
mock_workflow = _setup_mock_workflow_in_service(request)
- with mock.patch('aria.orchestrator.workflow_runner.Engine') as mock_engine_cls:
+ with mock.patch('aria.orchestrator.workflow_runner.engine.Engine') as mock_engine_cls:
_create_workflow_runner(request, mock_workflow)
_, engine_kwargs = mock_engine_cls.call_args
- assert isinstance(engine_kwargs.get('executor'), ProcessExecutor)
+ assert isinstance(engine_kwargs.get('default_executor'), ProcessExecutor)
def test_custom_executor(request):
mock_workflow = _setup_mock_workflow_in_service(request)
custom_executor = mock.MagicMock()
- with mock.patch('aria.orchestrator.workflow_runner.Engine') as mock_engine_cls:
+ with mock.patch('aria.orchestrator.workflow_runner.engine.Engine') as mock_engine_cls:
_create_workflow_runner(request, mock_workflow, executor=custom_executor)
_, engine_kwargs = mock_engine_cls.call_args
- assert engine_kwargs.get('executor') == custom_executor
+ assert engine_kwargs.get('default_executor') == custom_executor
def test_task_configuration_parameters(request):
@@ -117,7 +117,7 @@ def test_task_configuration_parameters(request):
task_max_attempts = 5
task_retry_interval = 7
- with mock.patch('aria.orchestrator.workflow_runner.Engine.execute') as mock_engine_execute:
+ with mock.patch('aria.orchestrator.workflow_runner.engine.Engine.execute') as mock_engine_execute:
_create_workflow_runner(request, mock_workflow, task_max_attempts=task_max_attempts,
task_retry_interval=task_retry_interval).execute()
_, engine_kwargs = mock_engine_execute.call_args
@@ -129,7 +129,7 @@ def test_execute(request, service):
mock_workflow = _setup_mock_workflow_in_service(request)
mock_engine = mock.MagicMock()
- with mock.patch('aria.orchestrator.workflow_runner.Engine.execute', return_value=mock_engine) \
+ with mock.patch('aria.orchestrator.workflow_runner.engine.Engine.execute', return_value=mock_engine) \
as mock_engine_execute:
workflow_runner = _create_workflow_runner(request, mock_workflow)
workflow_runner.execute()
@@ -145,7 +145,7 @@ def test_cancel_execution(request):
mock_workflow = _setup_mock_workflow_in_service(request)
mock_engine = mock.MagicMock()
- with mock.patch('aria.orchestrator.workflow_runner.Engine', return_value=mock_engine):
+ with mock.patch('aria.orchestrator.workflow_runner.engine.Engine', return_value=mock_engine):
workflow_runner = _create_workflow_runner(request, mock_workflow)
workflow_runner.cancel()
mock_engine.cancel_execution.assert_called_once_with(ctx=workflow_runner._workflow_context)
@@ -154,7 +154,7 @@ def test_cancel_execution(request):
def test_execution_model_creation(request, service, model):
mock_workflow = _setup_mock_workflow_in_service(request)
- with mock.patch('aria.orchestrator.workflow_runner.Engine'):
+ with mock.patch('aria.orchestrator.workflow_runner.engine.Engine'):
workflow_runner = _create_workflow_runner(request, mock_workflow)
assert model.execution.get(workflow_runner.execution.id) == workflow_runner.execution
@@ -171,7 +171,7 @@ def test_execution_inputs_override_workflow_inputs(request):
inputs=dict((name, models.Input.wrap(name, val)) for name, val
in wf_inputs.iteritems()))
- with mock.patch('aria.orchestrator.workflow_runner.Engine'):
+ with mock.patch('aria.orchestrator.workflow_runner.engine.Engine'):
workflow_runner = _create_workflow_runner(
request, mock_workflow, inputs={'input2': 'overriding-value2', 'input3': 7})
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/e3977fd5/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 64abaa9..7bdac67 100644
--- a/tests/orchestrator/workflows/core/test_engine.py
+++ b/tests/orchestrator/workflows/core/test_engine.py
@@ -24,7 +24,6 @@ from aria.orchestrator import (
operation,
)
from aria.modeling import models
-from aria.orchestrator import workflow_runner
from aria.orchestrator.workflows import (
api,
exceptions,
@@ -52,10 +51,10 @@ class BaseTest(object):
def _engine(workflow_func, workflow_context, executor):
graph = workflow_func(ctx=workflow_context)
execution = workflow_context.execution
- workflow_runner.construct_execution_tasks(execution, graph, executor.__class__)
+ engine.construct_execution_tasks(execution, graph, executor.__class__)
workflow_context.execution = execution
- return engine.Engine(executor=executor)
+ return engine.Engine(default_executor=executor)
@staticmethod
def _create_interface(ctx, func, arguments=None):
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/e3977fd5/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 0e22dd0..f0699df 100644
--- a/tests/orchestrator/workflows/core/test_events.py
+++ b/tests/orchestrator/workflows/core/test_events.py
@@ -15,7 +15,6 @@
import pytest
-from aria.orchestrator import workflow_runner
from aria.orchestrator.decorators import operation, workflow
from aria.orchestrator.workflows.core import engine
from aria.orchestrator.workflows.executor.thread import ThreadExecutor
@@ -114,13 +113,13 @@ 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
- workflow_runner.construct_execution_tasks(
+ engine.construct_execution_tasks(
ctx.execution,
single_operation_workflow(ctx, node=node, interface_name=interface_name, op_name=op_name),
ThreadExecutor)
ctx.execution = ctx.execution
- eng = engine.Engine(executor=ThreadExecutor())
+ eng = engine.Engine(default_executor=ThreadExecutor())
eng.execute(ctx)
return node
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/e3977fd5/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 de40fcf..569e8be 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
@@ -16,11 +16,11 @@
from networkx import topological_sort
from aria.modeling import models
-from aria.orchestrator import (
- context,
- workflow_runner
+from aria.orchestrator import context
+from aria.orchestrator.workflows import (
+ api,
+ core
)
-from aria.orchestrator.workflows import api
from aria.orchestrator.workflows.executor import base
from tests import mock
from tests import storage
@@ -70,7 +70,7 @@ def test_task_graph_into_execution_graph(tmpdir):
# Direct check
execution = workflow_context.model.execution.list()[0]
- workflow_runner.construct_execution_tasks(execution, test_task_graph, base.StubTaskExecutor)
+ core.engine.construct_execution_tasks(execution, test_task_graph, base.StubTaskExecutor)
workflow_context.execution = execution
execution_tasks = topological_sort(workflow_context._graph)
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/e3977fd5/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 4ba2670..8ede925 100644
--- a/tests/orchestrator/workflows/executor/test_process_executor_extension.py
+++ b/tests/orchestrator/workflows/executor/test_process_executor_extension.py
@@ -19,7 +19,7 @@ from aria import extension
from aria.orchestrator.workflows import api
from aria.orchestrator.workflows.core import engine
from aria.orchestrator.workflows.executor import process
-from aria.orchestrator import workflow, operation, workflow_runner
+from aria.orchestrator import workflow, operation
import tests
from tests import mock
@@ -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
- workflow_runner.construct_execution_tasks(context.execution, graph, executor.__class__)
+ engine.construct_execution_tasks(context.execution, graph, executor.__class__)
context.execution = context.execution
eng = engine.Engine(executor)
eng.execute(context)
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/e3977fd5/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 0edc009..9eb8916 100644
--- a/tests/orchestrator/workflows/executor/test_process_executor_tracked_changes.py
+++ b/tests/orchestrator/workflows/executor/test_process_executor_tracked_changes.py
@@ -20,7 +20,7 @@ import pytest
from aria.orchestrator.workflows import api
from aria.orchestrator.workflows.core import engine
from aria.orchestrator.workflows.executor import process
-from aria.orchestrator import workflow, operation, workflow_runner
+from aria.orchestrator import workflow, operation
from aria.orchestrator.workflows import exceptions
import tests
@@ -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
- workflow_runner.construct_execution_tasks(context.execution, graph, executor.__class__)
+ engine.construct_execution_tasks(context.execution, graph, executor.__class__)
context.execution = context.execution
eng = engine.Engine(executor)
eng.execute(context)