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 2016/11/06 09:04:35 UTC
[1/3] incubator-ariatosca git commit: ARIA-4 Create an API for
creating workflows [Forced Update!]
Repository: incubator-ariatosca
Updated Branches:
refs/heads/ARIA-3-api-for-creating-workflows 9002019e8 -> 017779dd5 (forced update)
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/tests/workflows/core/test_task_graph_into_exececution_graph.py
----------------------------------------------------------------------
diff --git a/tests/workflows/core/test_task_graph_into_exececution_graph.py b/tests/workflows/core/test_task_graph_into_exececution_graph.py
new file mode 100644
index 0000000..fe1624a
--- /dev/null
+++ b/tests/workflows/core/test_task_graph_into_exececution_graph.py
@@ -0,0 +1,83 @@
+# 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 networkx import topological_sort, DiGraph
+
+from aria import context
+from aria.workflows import api, core
+
+from ... import mock
+
+
+def test_task_graph_into_execution_graph():
+ task_context = mock.context.simple()
+ node = mock.models.get_dependency_node()
+ node_instance = mock.models.get_dependency_node_instance()
+ task_context.model.node.store(node)
+ task_context.model.node_instance.store(node_instance)
+
+ def sub_workflow(name, **_):
+ return api.task_graph.TaskGraph(name)
+
+ with context.workflow.current.push(task_context):
+ test_task_graph = api.task.WorkflowTask(sub_workflow, name='test_task_graph')
+ simple_before_task = api.task.OperationTask('test_simple_before_task', {}, node_instance)
+ simple_after_task = api.task.OperationTask('test_simple_after_task', {}, node_instance)
+
+ inner_task_graph = api.task.WorkflowTask(sub_workflow, name='test_inner_task_graph')
+ inner_task = api.task.OperationTask('test_inner_task', {}, node_instance)
+ inner_task_graph.add_tasks(inner_task)
+
+ test_task_graph.add_tasks(simple_before_task)
+ test_task_graph.add_tasks(simple_after_task)
+ test_task_graph.add_tasks(inner_task_graph)
+ test_task_graph.add_dependency(inner_task_graph, simple_before_task)
+ test_task_graph.add_dependency(simple_after_task, inner_task_graph)
+
+ # Direct check
+ execution_graph = DiGraph()
+ core.translation.build_execution_graph(task_graph=test_task_graph,
+ execution_graph=execution_graph)
+ execution_tasks = topological_sort(execution_graph)
+
+ assert len(execution_tasks) == 7
+
+ expected_tasks_names = [
+ '{0}-Start'.format(test_task_graph.id),
+ simple_before_task.id,
+ '{0}-Start'.format(inner_task_graph.id),
+ inner_task.id,
+ '{0}-End'.format(inner_task_graph.id),
+ simple_after_task.id,
+ '{0}-End'.format(test_task_graph.id)
+ ]
+
+ assert expected_tasks_names == execution_tasks
+
+ assert isinstance(_get_task_by_name(execution_tasks[0], execution_graph),
+ core.task.StartWorkflowTask)
+ assert simple_before_task == _get_task_by_name(execution_tasks[1], execution_graph).api_task
+ assert isinstance(_get_task_by_name(execution_tasks[2], execution_graph),
+ core.task.StartSubWorkflowTask)
+ assert inner_task == _get_task_by_name(execution_tasks[3], execution_graph).api_task
+ assert isinstance(_get_task_by_name(execution_tasks[4], execution_graph),
+ core.task.EndSubWorkflowTask)
+ assert simple_after_task == _get_task_by_name(execution_tasks[5], execution_graph).api_task
+ assert isinstance(_get_task_by_name(execution_tasks[6], execution_graph),
+ core.task.EndWorkflowTask)
+
+
+def _get_task_by_name(task_name, graph):
+ return graph.node[task_name]['task']
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/tests/workflows/test_executor.py
----------------------------------------------------------------------
diff --git a/tests/workflows/test_executor.py b/tests/workflows/test_executor.py
deleted file mode 100644
index 27cb2ad..0000000
--- a/tests/workflows/test_executor.py
+++ /dev/null
@@ -1,117 +0,0 @@
-# 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.
-
-import logging
-import uuid
-
-import pytest
-import retrying
-
-from aria import events
-from aria.storage import models
-from aria.workflows.core import executor
-
-
-class TestExecutor(object):
-
- @pytest.mark.parametrize('pool_size,executor_cls', [
- (1, executor.ThreadExecutor),
- (2, executor.ThreadExecutor),
- (1, executor.MultiprocessExecutor),
- (2, executor.MultiprocessExecutor),
- (0, executor.CurrentThreadBlockingExecutor)
- ])
- def test_execute(self, pool_size, executor_cls):
- self.executor = executor_cls(pool_size)
- expected_value = 'value'
- successful_task = MockTask(mock_successful_task)
- failing_task = MockTask(mock_failing_task)
- task_with_inputs = MockTask(mock_task_with_input, inputs={'input': expected_value})
-
- for task in [successful_task, failing_task, task_with_inputs]:
- self.executor.execute(task)
-
- @retrying.retry(stop_max_delay=10000, wait_fixed=100)
- def assertion():
- assert successful_task.states == ['start', 'success']
- assert failing_task.states == ['start', 'failure']
- assert task_with_inputs.states == ['start', 'failure']
- assert isinstance(failing_task.exception, TestException)
- assert isinstance(task_with_inputs.exception, TestException)
- assert task_with_inputs.exception.message == expected_value
- assertion()
-
- def setup_method(self):
- events.start_task_signal.connect(start_handler)
- events.on_success_task_signal.connect(success_handler)
- events.on_failure_task_signal.connect(failure_handler)
-
- def teardown_method(self):
- events.start_task_signal.disconnect(start_handler)
- events.on_success_task_signal.disconnect(success_handler)
- events.on_failure_task_signal.disconnect(failure_handler)
- if self.executor:
- self.executor.close()
-
-
-def mock_successful_task():
- pass
-
-
-def mock_failing_task():
- raise TestException
-
-
-def mock_task_with_input(input):
- raise TestException(input)
-
-
-class TestException(Exception):
- pass
-
-
-class MockContext(object):
-
- def __init__(self, operation_details, inputs):
- self.operation_details = operation_details
- self.inputs = inputs
- self.operation = models.Operation(execution_id='')
-
-
-class MockTask(object):
-
- def __init__(self, func, inputs=None):
- self.states = []
- self.exception = None
- self.id = str(uuid.uuid4())
- name = func.__name__
- operation = 'tests.workflows.test_executor.{name}'.format(name=name)
- self.context = MockContext(operation_details={'operation': operation},
- inputs=inputs or {})
- self.logger = logging.getLogger()
- self.name = name
-
-
-def start_handler(task, *args, **kwargs):
- task.states.append('start')
-
-
-def success_handler(task, *args, **kwargs):
- task.states.append('success')
-
-
-def failure_handler(task, exception, *args, **kwargs):
- task.states.append('failure')
- task.exception = exception
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/tests/workflows/test_task_graph_into_exececution_graph.py
----------------------------------------------------------------------
diff --git a/tests/workflows/test_task_graph_into_exececution_graph.py b/tests/workflows/test_task_graph_into_exececution_graph.py
deleted file mode 100644
index 1bae713..0000000
--- a/tests/workflows/test_task_graph_into_exececution_graph.py
+++ /dev/null
@@ -1,79 +0,0 @@
-# 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.
-
-import pytest
-from networkx import topological_sort, DiGraph
-
-from aria import contexts
-from aria.workflows.api import tasks_graph
-from aria.workflows.core import tasks, translation
-
-
-@pytest.fixture(autouse=True)
-def no_storage(monkeypatch):
- monkeypatch.setattr(tasks.OperationTask, '_create_operation_in_storage',
- value=lambda *args, **kwargs: None)
-
-
-def test_task_graph_into_execution_graph():
- task_graph = tasks_graph.TaskGraph('test_task_graph')
- simple_before_task = contexts.OperationContext('test_simple_before_task', {}, {}, None)
- simple_after_task = contexts.OperationContext('test_simple_after_task', {}, {}, None)
-
- inner_task_graph = tasks_graph.TaskGraph('test_inner_task_graph')
- inner_task = contexts.OperationContext('test_inner_task', {}, {}, None)
- inner_task_graph.add_task(inner_task)
-
- task_graph.add_task(simple_before_task)
- task_graph.add_task(simple_after_task)
- task_graph.add_task(inner_task_graph)
- task_graph.dependency(inner_task_graph, [simple_before_task])
- task_graph.dependency(simple_after_task, [inner_task_graph])
-
- # Direct check
- execution_graph = DiGraph()
- translation.build_execution_graph(task_graph=task_graph,
- workflow_context=None,
- execution_graph=execution_graph)
- execution_tasks = topological_sort(execution_graph)
-
- assert len(execution_tasks) == 7
-
- expected_tasks_names = [
- '{0}-Start'.format(task_graph.id),
- simple_before_task.id,
- '{0}-Start'.format(inner_task_graph.id),
- inner_task.id,
- '{0}-End'.format(inner_task_graph.id),
- simple_after_task.id,
- '{0}-End'.format(task_graph.id)
- ]
-
- assert expected_tasks_names == execution_tasks
-
- assert isinstance(_get_task_by_name(execution_tasks[0], execution_graph),
- tasks.StartWorkflowTask)
- assert simple_before_task == _get_task_by_name(execution_tasks[1], execution_graph).context
- assert isinstance(_get_task_by_name(execution_tasks[2], execution_graph),
- tasks.StartSubWorkflowTask)
- assert inner_task == _get_task_by_name(execution_tasks[3], execution_graph).context
- assert isinstance(_get_task_by_name(execution_tasks[4], execution_graph), tasks.
- EndSubWorkflowTask)
- assert simple_after_task == _get_task_by_name(execution_tasks[5], execution_graph).context
- assert isinstance(_get_task_by_name(execution_tasks[6], execution_graph), tasks.EndWorkflowTask)
-
-
-def _get_task_by_name(task_name, graph):
- return graph.node[task_name]['task']
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/tox.ini
----------------------------------------------------------------------
diff --git a/tox.ini b/tox.ini
index be7b2b5..f8b95d3 100644
--- a/tox.ini
+++ b/tox.ini
@@ -32,9 +32,9 @@ commands=pytest tests --cov-report term-missing --cov aria
[testenv:pylint_code]
deps =
pylint
-commands=pylint aria --rcfile=aria/.pylintrc
+commands=pylint --rcfile=aria/.pylintrc --errors-only --disable=print-statement aria
[testenv:pylint_tests]
deps =
pylint
-commands=pylint tests --rcfile=tests/.pylintrc
+commands=pylint --rcfile=tests/.pylintrc --disable=print-statement tests
[3/3] incubator-ariatosca git commit: ARIA-4 Create an API for
creating workflows
Posted by mx...@apache.org.
ARIA-4 Create an API for creating workflows
An API for creating workflows. Users can build graphs of tasks and
set depenedencies in between tasks to execute them in a specific
order.
This commit also includes minimal reorganization of a few test
modules, so they are now using the same file system hierarchy as
the modules which they test.
ARIA-5-Adapt-workflow-API-users-to-modified-API
Project: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/commit/017779dd
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/tree/017779dd
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/diff/017779dd
Branch: refs/heads/ARIA-3-api-for-creating-workflows
Commit: 017779dd564513714e67f8e6d6af5a8065736a48
Parents: e1c919d
Author: Ran Ziv <ra...@gigaspaces.com>
Authored: Fri Oct 21 13:21:25 2016 +0300
Committer: mxmrlv <mx...@gmail.com>
Committed: Sun Nov 6 11:03:02 2016 +0200
----------------------------------------------------------------------
aria/.pylintrc | 2 +-
aria/__init__.py | 2 +-
aria/cli/commands.py | 2 +-
aria/context/__init__.py | 20 +
aria/context/operation.py | 68 ++
aria/context/workflow.py | 188 +++++
aria/contexts.py | 211 ------
aria/decorators.py | 47 +-
aria/logger.py | 6 +-
aria/storage/models.py | 58 +-
aria/tools/lru_cache.py | 6 +-
aria/workflows/api/__init__.py | 2 +
aria/workflows/api/task.py | 90 +++
aria/workflows/api/task_graph.py | 294 ++++++++
aria/workflows/api/tasks_graph.py | 203 ------
aria/workflows/builtin/execute_operation.py | 34 +-
aria/workflows/builtin/heal.py | 167 +++--
aria/workflows/builtin/install.py | 23 +-
aria/workflows/builtin/uninstall.py | 17 +-
aria/workflows/builtin/workflows.py | 144 ++--
aria/workflows/core/__init__.py | 2 +
aria/workflows/core/executor.py | 3 +-
aria/workflows/core/task.py | 106 +++
aria/workflows/core/tasks.py | 104 ---
aria/workflows/core/translation.py | 45 +-
requirements.txt | 1 +
setup.py | 5 +
tests/.pylintrc | 6 +-
tests/mock/__init__.py | 16 +
tests/mock/context.py | 36 +
tests/mock/models.py | 132 ++++
tests/mock/operations.py | 33 +
tests/test_logger.py | 2 +-
tests/workflows/__init__.py | 2 +
tests/workflows/api/__init__.py | 0
tests/workflows/api/test_task.py | 120 ++++
tests/workflows/api/test_task_graph.py | 714 +++++++++++++++++++
tests/workflows/builtin/__init__.py | 53 ++
.../workflows/builtin/test_execute_operation.py | 74 ++
tests/workflows/builtin/test_heal.py | 117 +++
tests/workflows/builtin/test_install.py | 63 ++
tests/workflows/builtin/test_uninstall.py | 63 ++
tests/workflows/core/__init__.py | 0
tests/workflows/core/test_executor.py | 122 ++++
.../test_task_graph_into_exececution_graph.py | 83 +++
tests/workflows/test_executor.py | 117 ---
.../test_task_graph_into_exececution_graph.py | 79 --
tox.ini | 4 +-
48 files changed, 2693 insertions(+), 993 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/aria/.pylintrc
----------------------------------------------------------------------
diff --git a/aria/.pylintrc b/aria/.pylintrc
index eb188a0..ee27fdb 100644
--- a/aria/.pylintrc
+++ b/aria/.pylintrc
@@ -62,7 +62,7 @@ confidence=
# --enable=similarities". If you want to run only the classes checker, but have
# no Warning level messages displayed, use"--disable=all --enable=classes
# --disable=W"
-disable=import-star-module-level,old-octal-literal,oct-method,print-statement,unpacking-in-except,parameter-unpacking,backtick,old-raise-syntax,old-ne-operator,long-suffix,dict-view-method,dict-iter-method,metaclass-assignment,next-method-called,raising-string,indexing-exception,raw_input-builtin,long-builtin,file-builtin,execfile-builtin,coerce-builtin,cmp-builtin,buffer-builtin,basestring-builtin,apply-builtin,filter-builtin-not-iterating,using-cmp-argument,useless-suppression,range-builtin-not-iterating,suppressed-message,no-absolute-import,old-division,cmp-method,reload-builtin,zip-builtin-not-iterating,intern-builtin,unichr-builtin,reduce-builtin,standarderror-builtin,unicode-builtin,xrange-builtin,coerce-method,delslice-method,getslice-method,setslice-method,input-builtin,round-builtin,hex-method,nonzero-method,map-builtin-not-iterating,redefined-builtin,logging-format-interpolation
+disable=import-star-module-level,old-octal-literal,oct-method,print-statement,unpacking-in-except,parameter-unpacking,backtick,old-raise-syntax,old-ne-operator,long-suffix,dict-view-method,dict-iter-method,metaclass-assignment,next-method-called,raising-string,indexing-exception,raw_input-builtin,long-builtin,file-builtin,execfile-builtin,coerce-builtin,cmp-builtin,buffer-builtin,basestring-builtin,apply-builtin,filter-builtin-not-iterating,using-cmp-argument,useless-suppression,range-builtin-not-iterating,suppressed-message,no-absolute-import,old-division,cmp-method,reload-builtin,zip-builtin-not-iterating,intern-builtin,unichr-builtin,reduce-builtin,standarderror-builtin,unicode-builtin,xrange-builtin,coerce-method,delslice-method,getslice-method,setslice-method,input-builtin,round-builtin,hex-method,nonzero-method,map-builtin-not-iterating,redefined-builtin,logging-format-interpolation,import-error
[REPORTS]
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/aria/__init__.py
----------------------------------------------------------------------
diff --git a/aria/__init__.py b/aria/__init__.py
index 4d9b0f4..eca7b9b 100644
--- a/aria/__init__.py
+++ b/aria/__init__.py
@@ -56,7 +56,7 @@ def application_model_storage(driver):
models.DeploymentModification,
models.Execution,
models.ProviderContext,
- models.Operation,
+ models.Task,
])
return _model_storage[driver]
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/aria/cli/commands.py
----------------------------------------------------------------------
diff --git a/aria/cli/commands.py b/aria/cli/commands.py
index a38229d..f909a1e 100644
--- a/aria/cli/commands.py
+++ b/aria/cli/commands.py
@@ -29,7 +29,7 @@ from aria import application_model_storage, application_resource_storage
from aria.logger import LoggerMixin
from aria.storage import FileSystemModelDriver, FileSystemResourceDriver
from aria.tools.application import StorageManager
-from aria.contexts import WorkflowContext
+from aria.context.workflow import WorkflowContext
from aria.workflows.core.engine import Engine
from aria.workflows.core.executor import ThreadExecutor
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/aria/context/__init__.py
----------------------------------------------------------------------
diff --git a/aria/context/__init__.py b/aria/context/__init__.py
new file mode 100644
index 0000000..20e19db
--- /dev/null
+++ b/aria/context/__init__.py
@@ -0,0 +1,20 @@
+# 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.
+
+"""
+Provides contexts to workflow and operation
+"""
+
+from . import workflow, operation
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/aria/context/operation.py
----------------------------------------------------------------------
diff --git a/aria/context/operation.py b/aria/context/operation.py
new file mode 100644
index 0000000..d4d229a
--- /dev/null
+++ b/aria/context/operation.py
@@ -0,0 +1,68 @@
+# 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.
+
+"""
+Workflow and operation contexts
+"""
+
+from uuid import uuid4
+
+from aria.logger import LoggerMixin
+
+class OperationContext(LoggerMixin):
+ """
+ Context object used during operation creation and execution
+ """
+
+ def __init__(
+ self,
+ name,
+ operation_details,
+ workflow_context,
+ node_instance,
+ inputs=None):
+ super(OperationContext, self).__init__()
+ self.name = name
+ self.id = str(uuid4())
+ self.operation_details = operation_details
+ self.workflow_context = workflow_context
+ self.node_instance = node_instance
+ self.inputs = inputs or {}
+
+ def __repr__(self):
+ details = ', '.join(
+ '{0}={1}'.format(key, value)
+ for key, value in self.operation_details.items())
+ return '{name}({0})'.format(details, name=self.name)
+
+ def __getattr__(self, attr):
+ try:
+ return getattr(self.workflow_context, attr)
+ except AttributeError:
+ return super(OperationContext, self).__getattribute__(attr)
+
+ @property
+ def operation(self):
+ """
+ The model operation
+ """
+ return self.storage.operation.get(self.id)
+
+ @operation.setter
+ def operation(self, value):
+ """
+ Store the operation in the model storage
+ """
+ self.storage.operation.store(value)
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/aria/context/workflow.py
----------------------------------------------------------------------
diff --git a/aria/context/workflow.py b/aria/context/workflow.py
new file mode 100644
index 0000000..b5b4e4b
--- /dev/null
+++ b/aria/context/workflow.py
@@ -0,0 +1,188 @@
+# 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.
+
+"""
+Workflow and operation contexts
+"""
+
+import threading
+from uuid import uuid4
+from contextlib import contextmanager
+
+from ..logger import LoggerMixin
+from ..tools.lru_cache import lru_cache
+
+
+class WorkflowContext(LoggerMixin):
+ """
+ Context object used during workflow creation and execution
+ """
+
+ def __init__(
+ self,
+ model_storage,
+ resource_storage,
+ deployment_id,
+ workflow_id,
+ # TODO: this should be changed when we decide where the execution is created
+ execution_id=None,
+ parameters=None,
+ **kwargs):
+ super(WorkflowContext, self).__init__(**kwargs)
+ self.id = str(uuid4())
+ self.model = model_storage
+ self.resource = resource_storage
+ self.deployment_id = deployment_id
+ self.workflow_id = workflow_id
+ self.execution_id = execution_id
+ self.parameters = parameters or {}
+
+ def __repr__(self):
+ return (
+ '{name}(deployment_id={self.deployment_id}, '
+ 'workflow_id={self.workflow_id}, '
+ 'execution_id={self.execution_id})'.format(
+ name=self.__class__.__name__, self=self))
+
+ @property
+ def blueprint_id(self):
+ """
+ The blueprint id
+ """
+ return self.deployment.blueprint_id
+
+ @property
+ @lru_cache()
+ def blueprint(self):
+ """
+ The blueprint model
+ """
+ return self.model.blueprint.get(self.blueprint_id)
+
+ @property
+ @lru_cache()
+ def deployment(self):
+ """
+ The deployment model
+ """
+ return self.model.deployment.get(self.deployment_id)
+
+ @property
+ def nodes(self):
+ """
+ Iterator over nodes
+ """
+ return self.model.node.iter(
+ filters={'blueprint_id': self.blueprint_id})
+
+ @property
+ def node_instances(self):
+ """
+ Iterator over node instances
+ """
+ return self.model.node_instance.iter(filters={'deployment_id': self.deployment_id})
+
+ @property
+ def execution(self):
+ """
+ The execution model
+ """
+ return self.model.execution.get(self.execution_id)
+
+ @execution.setter
+ def execution(self, value):
+ """
+ Store the execution in the model storage
+ """
+ self.model.execution.store(value)
+
+ def download_blueprint_resource(self, destination, path=None):
+ """
+ Download a blueprint resource from the resource storage
+ """
+ return self.resource.blueprint.download(
+ entry_id=self.blueprint_id,
+ destination=destination,
+ path=path)
+
+ def download_deployment_resource(self, destination, path=None):
+ """
+ Download a deployment resource from the resource storage
+ """
+ return self.resource.deployment.download(
+ entry_id=self.deployment_id,
+ destination=destination,
+ path=path)
+
+ @lru_cache()
+ def get_deployment_resource_data(self, path=None):
+ """
+ Read a deployment resource as string from the resource storage
+ """
+ return self.resource.deployment.data(entry_id=self.deployment_id, path=path)
+
+ @lru_cache()
+ def get_blueprint_resource_data(self, path=None):
+ """
+ Read a blueprint resource as string from the resource storage
+ """
+ return self.resource.blueprint.data(entry_id=self.blueprint_id, path=path)
+
+
+class _CurrentContext(threading.local):
+ """
+ Provides thread-level context, which sugarcoats the task api.
+ """
+
+ def _set(self, value):
+ setattr(self, '_workflow_context', value)
+
+ def get(self):
+ """
+ Retrieves the current workflow context
+ :return: the workflow context
+ :rtype: WorkflowContext
+ """
+ try:
+ return getattr(self, '_workflow_context')
+ except AttributeError:
+ return None
+
+ def clear(self):
+ """
+ Clears the current context
+ """
+ if hasattr(self, '_workflow_context'):
+ delattr(self, '_workflow_context')
+
+ @contextmanager
+ def push(self, workflow_context):
+ """
+ Switches the current context to the provides context
+ :param workflow_context: the context to switch to.
+ :yields: the current context
+ """
+ if hasattr(self, '_workflow_context'):
+ prev_workflow_context = getattr(self, '_workflow_context')
+ else:
+ prev_workflow_context = None
+ self._set(workflow_context)
+ try:
+ yield self
+ finally:
+ self._set(prev_workflow_context)
+
+current = _CurrentContext()
+
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/aria/contexts.py
----------------------------------------------------------------------
diff --git a/aria/contexts.py b/aria/contexts.py
deleted file mode 100644
index ae7fc66..0000000
--- a/aria/contexts.py
+++ /dev/null
@@ -1,211 +0,0 @@
-# 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.
-
-"""
-Workflow and operation contexts
-"""
-
-from uuid import uuid4
-
-from aria.logger import LoggerMixin
-from aria.tools.lru_cache import lru_cache
-from aria.workflows.api.tasks_graph import TaskGraph
-
-
-class WorkflowContext(LoggerMixin):
- """
- Context object used during workflow creation and execution
- """
-
- def __init__(
- self,
- name,
- model_storage,
- resource_storage,
- deployment_id,
- workflow_id,
- parameters=None,
- **kwargs):
- super(WorkflowContext, self).__init__(**kwargs)
- self.name = name
- self.id = str(uuid4())
- self.model = model_storage
- self.resource = resource_storage
- self.deployment_id = deployment_id
- self.workflow_id = workflow_id
- self.execution_id = str(uuid4())
- self.parameters = parameters or {}
-
- def __repr__(self):
- return (
- '{name}(deployment_id={self.deployment_id}, '
- 'workflow_id={self.workflow_id}, '
- 'execution_id={self.execution_id})'.format(
- name=self.__class__.__name__, self=self))
-
- def operation(
- self,
- name,
- operation_details,
- node_instance,
- inputs=None):
- """
- Called during workflow creation, return an operation context. This object should be added to
- the task graph.
- """
- return OperationContext(
- name=name,
- operation_details=operation_details,
- workflow_context=self,
- node_instance=node_instance,
- inputs=inputs or {})
-
- @property
- def task_graph(self):
- """
- The task graph class
- """
- return TaskGraph
-
- @property
- def blueprint_id(self):
- """
- The blueprint id
- """
- return self.deployment.blueprint_id
-
- @property
- @lru_cache()
- def blueprint(self):
- """
- The blueprint model
- """
- return self.model.blueprint.get(self.blueprint_id)
-
- @property
- @lru_cache()
- def deployment(self):
- """
- The deployment model
- """
- return self.model.deployment.get(self.deployment_id)
-
- @property
- def nodes(self):
- """
- Iterator over nodes
- """
- return self.model.node.iter(
- filters={'blueprint_id': self.blueprint_id})
-
- @property
- def node_instances(self):
- """
- Iterator over node instances
- """
- return self.model.node_instance.iter(filters={'deployment_id': self.deployment_id})
-
- @property
- def execution(self):
- """
- The execution model
- """
- return self.model.execution.get(self.execution_id)
-
- @execution.setter
- def execution(self, value):
- """
- Store the execution in the model storage
- """
- self.model.execution.store(value)
-
- def download_blueprint_resource(self, destination, path=None):
- """
- Download a blueprint resource from the resource storage
- """
- return self.resource.blueprint.download(
- entry_id=self.blueprint_id,
- destination=destination,
- path=path)
-
- def download_deployment_resource(self, destination, path=None):
- """
- Download a deployment resource from the resource storage
- """
- return self.resource.deployment.download(
- entry_id=self.deployment_id,
- destination=destination,
- path=path)
-
- @lru_cache()
- def get_deployment_resource_data(self, path=None):
- """
- Read a deployment resource as string from the resource storage
- """
- return self.resource.deployment.data(entry_id=self.deployment_id, path=path)
-
- @lru_cache()
- def get_blueprint_resource_data(self, path=None):
- """
- Read a blueprint resource as string from the resource storage
- """
- return self.resource.blueprint.data(entry_id=self.blueprint_id, path=path)
-
-
-class OperationContext(LoggerMixin):
- """
- Context object used during operation creation and execution
- """
-
- def __init__(
- self,
- name,
- operation_details,
- workflow_context,
- node_instance,
- inputs=None):
- super(OperationContext, self).__init__()
- self.name = name
- self.id = str(uuid4())
- self.operation_details = operation_details
- self.workflow_context = workflow_context
- self.node_instance = node_instance
- self.inputs = inputs or {}
-
- def __repr__(self):
- details = ', '.join(
- '{0}={1}'.format(key, value)
- for key, value in self.operation_details.items())
- return '{name}({0})'.format(details, name=self.name)
-
- def __getattr__(self, attr):
- try:
- return getattr(self.workflow_context, attr)
- except AttributeError:
- return super(OperationContext, self).__getattribute__(attr)
-
- @property
- def operation(self):
- """
- The model operation
- """
- return self.storage.operation.get(self.id)
-
- @operation.setter
- def operation(self, value):
- """
- Store the operation in the model storage
- """
- self.storage.operation.store(value)
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/aria/decorators.py
----------------------------------------------------------------------
diff --git a/aria/decorators.py b/aria/decorators.py
index 7bc41b3..5713246 100644
--- a/aria/decorators.py
+++ b/aria/decorators.py
@@ -20,12 +20,13 @@ Workflow and operation decorators
from uuid import uuid4
from functools import partial, wraps
-from aria.tools.validation import validate_function_arguments
+from . import context
+from .workflows.api import task_graph
+from .tools.validation import validate_function_arguments
def workflow(
func=None,
- workflow_context=True,
simple_workflow=True,
suffix_template=''):
"""
@@ -34,31 +35,29 @@ def workflow(
if func is None:
return partial(
workflow,
- workflow_context=workflow_context,
simple_workflow=simple_workflow,
suffix_template=suffix_template)
@wraps(func)
- def _wrapper(context, **custom_kwargs):
+ def _wrapper(ctx, **custom_kwargs):
+
workflow_name = _generate_workflow_name(
func_name=func.__name__,
suffix_template=suffix_template,
- context=context,
+ ctx=ctx,
**custom_kwargs)
- func_kwargs = _create_func_kwargs(
- custom_kwargs,
- context,
- add_context=workflow_context,
- workflow_name=workflow_name)
- validate_function_arguments(func, func_kwargs)
- func(**func_kwargs)
- return func_kwargs['graph']
+
+ custom_kwargs.setdefault('ctx', ctx)
+ custom_kwargs.setdefault('graph', task_graph.TaskGraph(workflow_name))
+ validate_function_arguments(func, custom_kwargs)
+ with context.workflow.current.push(ctx):
+ func(**custom_kwargs)
+ return custom_kwargs['graph']
return _wrapper
def operation(
- func=None,
- operation_context=True):
+ func=None):
"""
Operation decorator
"""
@@ -66,29 +65,25 @@ def operation(
return partial(operation)
@wraps(func)
- def _wrapper(context, **custom_kwargs):
+ def _wrapper(ctx, **custom_kwargs):
func_kwargs = _create_func_kwargs(
custom_kwargs,
- context,
- add_context=operation_context)
+ ctx)
validate_function_arguments(func, func_kwargs)
- context.description = func.__doc__
+ ctx.description = func.__doc__
return func(**func_kwargs)
return _wrapper
-def _generate_workflow_name(func_name, context, suffix_template, **custom_kwargs):
+def _generate_workflow_name(func_name, ctx, suffix_template, **custom_kwargs):
return '{func_name}.{suffix}'.format(
func_name=func_name,
- suffix=suffix_template.format(context=context, **custom_kwargs) or str(uuid4()))
+ suffix=suffix_template.format(ctx=ctx, **custom_kwargs) or str(uuid4()))
def _create_func_kwargs(
kwargs,
- context,
- add_context=True,
+ ctx,
workflow_name=None):
- if add_context:
- kwargs['context'] = context
- kwargs.setdefault('graph', context.task_graph(workflow_name))
+ kwargs.setdefault('graph', ctx.task_graph(workflow_name))
return kwargs
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/aria/logger.py
----------------------------------------------------------------------
diff --git a/aria/logger.py b/aria/logger.py
index 4959126..0002cb5 100644
--- a/aria/logger.py
+++ b/aria/logger.py
@@ -37,7 +37,7 @@ class LoggerMixin(object):
def __init__(self, *args, **kwargs):
self.logger_name = self.logger_name or self.__class__.__name__
- self.logger = _base_logger.getChild(self.logger_name)
+ self.logger = logging.getLogger('{0}.{1}'.format(_base_logger.name, self.logger_name))
self.logger.setLevel(self.logger_level)
super(LoggerMixin, self).__init__(*args, **kwargs)
@@ -63,7 +63,7 @@ class LoggerMixin(object):
def __setstate__(self, obj_dict):
vars(self).update(
- logger=_base_logger.getChild(obj_dict['logger_name']),
+ logger=logging.getLogger('{0}.{1}'.format(_base_logger.name, obj_dict['logger_name'])),
**obj_dict)
@@ -112,7 +112,7 @@ class _DefaultConsoleFormat(logging.Formatter):
self._fmt = '%(levelname)s: %(message)s'
except AttributeError:
return record.message
- return super(_DefaultConsoleFormat, self).format(record)
+ return logging.Formatter.format(self, record)
def create_file_log_handler(
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/aria/storage/models.py
----------------------------------------------------------------------
diff --git a/aria/storage/models.py b/aria/storage/models.py
index 32403ed..64c00bc 100644
--- a/aria/storage/models.py
+++ b/aria/storage/models.py
@@ -216,32 +216,6 @@ class Execution(Model):
is_system_workflow = Field(type=bool, default=False)
-class Operation(Model):
- """
- A Model which represents an operation
- """
- PENDING = 'pending'
- STARTED = 'started'
- SUCCESS = 'success'
- FAILED = 'failed'
- STATES = (
- PENDING,
- STARTED,
- SUCCESS,
- FAILED,
- )
- END_STATES = [SUCCESS, FAILED]
-
- id = Field(type=basestring, default=uuid_generator)
- status = Field(type=basestring, choices=STATES, default=STARTED)
- execution_id = Field(type=basestring)
- eta = Field(type=datetime, default=0)
- started_at = Field(type=datetime, default=None)
- ended_at = Field(type=datetime, default=None)
- max_retries = Field(type=int, default=0)
- retry_count = Field(type=int, default=0)
-
-
class Relationship(Model):
"""
A Model which represents a relationship
@@ -395,3 +369,35 @@ class Plugin(Model):
excluded_wheels = Field()
supported_py_versions = Field(type=list)
uploaded_at = Field(type=datetime)
+
+
+class Task(Model):
+ """
+ A Model which represents an task
+ """
+ PENDING = 'pending'
+ STARTED = 'started'
+ SUCCESS = 'success'
+ FAILED = 'failed'
+ STATES = (
+ PENDING,
+ STARTED,
+ SUCCESS,
+ FAILED,
+ )
+ END_STATES = [SUCCESS, FAILED]
+
+ id = Field(type=basestring, default=uuid_generator)
+ status = Field(type=basestring, choices=STATES, default=STARTED)
+ execution_id = Field(type=basestring)
+ eta = Field(type=datetime, default=0)
+ started_at = Field(type=datetime, default=None)
+ ended_at = Field(type=datetime, default=None)
+ max_retries = Field(type=int, default=0)
+ retry_count = Field(type=int, default=0)
+
+ # Operation specific fields
+ name = Field(type=basestring)
+ operation_details = Field(type=dict)
+ node_instance = PointerField(type=NodeInstance)
+ inputs = Field(type=dict, default=lambda: {})
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/aria/tools/lru_cache.py
----------------------------------------------------------------------
diff --git a/aria/tools/lru_cache.py b/aria/tools/lru_cache.py
index 5863376..bb39b90 100755
--- a/aria/tools/lru_cache.py
+++ b/aria/tools/lru_cache.py
@@ -21,7 +21,11 @@ Function lru_cache implementation for python 2.7
from time import time
from functools import partial, wraps
from itertools import imap
-from collections import OrderedDict
+
+try:
+ from collections import OrderedDict
+except ImportError:
+ from ordereddict import OrderedDict
class _LRUCache(object):
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/aria/workflows/api/__init__.py
----------------------------------------------------------------------
diff --git a/aria/workflows/api/__init__.py b/aria/workflows/api/__init__.py
index ae1e83e..b7f4b0c 100644
--- a/aria/workflows/api/__init__.py
+++ b/aria/workflows/api/__init__.py
@@ -12,3 +12,5 @@
# 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 . import task, task_graph
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/aria/workflows/api/task.py
----------------------------------------------------------------------
diff --git a/aria/workflows/api/task.py b/aria/workflows/api/task.py
new file mode 100644
index 0000000..a24fdc4
--- /dev/null
+++ b/aria/workflows/api/task.py
@@ -0,0 +1,90 @@
+# 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.
+
+"""
+Provides the tasks to be enterd into the task graph
+"""
+from uuid import uuid4
+
+from aria import context
+
+
+class BaseTask(object):
+ """
+ Abstract task_graph task
+ """
+ pass
+
+
+class OperationTask(BaseTask):
+ """
+ Represents an operation task in the task_graph
+ """
+
+ def __init__(self,
+ name,
+ operation_details,
+ node_instance,
+ **operation_kwargs):
+ """
+ Creates an operation task using the name, details, node instance and any additional kwargs.
+ :param name: the operation of the name.
+ :param operation_details: the details for the operation.
+ :param node_instance: the node instnace on which this operation is registered.
+ :param operation_kwargs: any additional kwargs for the operation.
+ """
+ self.id = uuid4()
+ self.workflow_context = context.workflow.current.get()
+ self.name = name
+ self.operation_details = operation_details
+ self.node_instance = node_instance
+ self.operation_kwargs = operation_kwargs
+
+ def __getattr__(self, item):
+ try:
+ return self.__getattribute__(item)
+ except AttributeError:
+ return getattr(self._workflow_context.model.task.get(self._task_model.id), item)
+
+
+class WorkflowTask(BaseTask):
+ """
+ Represents an workflow task in the task_graph
+ """
+ def __init__(self, workflow_func, **kwargs):
+ """
+ Creates a workflow task using the workflow_func provided and its kwargs
+ :param workflow_func: the function to run
+ :param kwargs: the kwargs that would be passed to the workflow_func
+ """
+ kwargs.setdefault('ctx', context.workflow.current.get())
+ self._graph = workflow_func(**kwargs)
+
+ @property
+ def graph(self):
+ """
+ The graph constructed by the sub workflow
+ :return:
+ """
+ return self._graph
+
+ def __getattr__(self, item):
+ try:
+ return self.__getattribute__(item)
+ except AttributeError:
+ try:
+ return getattr(self._graph, item)
+ except AttributeError:
+ return getattr(context.workflow.current.get(), item)
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/aria/workflows/api/task_graph.py
----------------------------------------------------------------------
diff --git a/aria/workflows/api/task_graph.py b/aria/workflows/api/task_graph.py
new file mode 100644
index 0000000..42cf8b6
--- /dev/null
+++ b/aria/workflows/api/task_graph.py
@@ -0,0 +1,294 @@
+# 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.
+
+"""
+Task graph. Used by users to build workflows
+"""
+
+from uuid import uuid4
+from collections import Iterable
+
+from networkx import DiGraph, topological_sort
+
+from . import task
+
+
+class TaskNotInGraphError(Exception):
+ """
+ An error representing a scenario where a given task is not in the graph as expected
+ """
+ pass
+
+
+def _filter_out_empty_tasks(func=None):
+ if func is None:
+ return lambda f: _filter_out_empty_tasks(func=f)
+
+ def _wrapper(task, *tasks, **kwargs):
+ tasks = filter(lambda t: bool(t), [task] + list(tasks))
+ return func(*tasks, **kwargs)
+ return _wrapper
+
+
+class TaskGraph(object):
+ """
+ A tasks graph builder.
+ Build an operations flow graph
+ """
+
+ def __init__(self, name):
+ self.name = name
+ self._id = str(uuid4())
+ self._graph = DiGraph()
+
+ def __repr__(self):
+ return '{name}(id={self._id}, name={self.name}, graph={self._graph!r})'.format(
+ name=self.__class__.__name__, self=self)
+
+ @property
+ def id(self):
+ """
+ Represents the id of the graph
+ :return: graph id
+ """
+ return self._id
+
+ # graph traversal methods
+
+ @property
+ def tasks(self):
+ """
+ An iterator on tasks added to the graph
+ :yields: Iterator over all tasks in the graph
+ """
+ for _, data in self._graph.nodes_iter(data=True):
+ yield data['task']
+
+ def topological_order(self, reverse=False):
+ """
+ Returns topological sort on the graph
+ :param reverse: whether to reverse the sort
+ :return: a list which represents the topological sort
+ """
+ for task_id in topological_sort(self._graph, reverse=reverse):
+ yield self.get_task(task_id)
+
+ def get_dependencies(self, dependent_task):
+ """
+ Iterates over the task's dependencies
+ :param BaseTask dependent_task: The task whose dependencies are requested
+ :yields: Iterator over all tasks which dependency_task depends on
+ :raise: TaskNotInGraphError if dependent_task is not in the graph
+ """
+ if not self.has_tasks(dependent_task):
+ raise TaskNotInGraphError('Task id: {0}'.format(dependent_task.id))
+ for _, dependency_id in self._graph.out_edges_iter(dependent_task.id):
+ yield self.get_task(dependency_id)
+
+ def get_dependents(self, dependency_task):
+ """
+ Iterates over the task's dependents
+ :param BaseTask dependency_task: The task whose dependents are requested
+ :yields: Iterator over all tasks which depend on dependency_task
+ :raise: TaskNotInGraphError if dependency_task is not in the graph
+ """
+ if not self.has_tasks(dependency_task):
+ raise TaskNotInGraphError('Task id: {0}'.format(dependency_task.id))
+ for dependent_id, _ in self._graph.in_edges_iter(dependency_task.id):
+ yield self.get_task(dependent_id)
+
+ # task methods
+
+ def get_task(self, task_id):
+ """
+ Get a task instance that's been inserted to the graph by the task's id
+ :param basestring task_id: The task's id
+ :return: Requested task
+ :rtype: BaseTask
+ :raise: TaskNotInGraphError if no task found in the graph with the given id
+ """
+ if not self._graph.has_node(task_id):
+ raise TaskNotInGraphError('Task id: {0}'.format(task_id))
+ data = self._graph.node[task_id]
+ return data['task']
+
+ @_filter_out_empty_tasks
+ def add_tasks(self, *tasks):
+ """
+ Add a task to the graph
+ :param BaseTask task: The task
+ :return: A list of added tasks
+ :rtype: list
+ """
+ # assert all([isinstance(t, (task.BaseTask, Iterable)) for t in tasks])
+ return_tasks = []
+
+ for t in tasks:
+ if isinstance(t, Iterable):
+ return_tasks += self.add_tasks(*t)
+ elif not self.has_tasks(t):
+ self._graph.add_node(t.id, task=t)
+ return_tasks.append(t)
+
+ return return_tasks
+
+ @_filter_out_empty_tasks
+ def remove_tasks(self, *tasks):
+ """
+ Remove the provided task from the graph
+ :param BaseTask task: The task
+ :return: A list of removed tasks
+ :rtype: list
+ """
+ return_tasks = []
+
+ if len(tasks) > 1:
+ for sub_task in tasks:
+ return_tasks += self.remove_tasks(*sub_task)
+ elif self.has_tasks(tasks[0]):
+ self._graph.remove_node(tasks[0].id)
+ return_tasks.append(tasks[0])
+
+ return return_tasks
+
+ @_filter_out_empty_tasks
+ def has_tasks(self, *tasks):
+ """
+ Check whether a task is in the graph or not
+ :param BaseTask task: The task
+ :return: True if all tasks are in the graph, otherwise True
+ :rtype: list
+ """
+ # assert all(isinstance(t, (task.BaseTask, Iterable)) for t in tasks)
+ return_value = True
+
+ for t in tasks:
+ if isinstance(t, Iterable):
+ return_value &= self.has_tasks(*t)
+ else:
+ return_value &= self._graph.has_node(t.id)
+
+ return return_value
+
+ def add_dependency(self, dependent, dependency):
+ """
+ Add a dependency for one item (task, sequence or parallel) on another
+ The dependent will only be executed after the dependency terminates
+ If either of the items is either a sequence or a parallel,
+ multiple dependencies may be added
+ :param BaseTask|_TasksArrangement dependent: The dependent (task, sequence or parallel)
+ :param BaseTask|_TasksArrangement dependency: The dependency (task, sequence or parallel)
+ :return: True if the dependency between the two hadn't already existed, otherwise False
+ :rtype: bool
+ :raise TaskNotInGraphError if either the dependent or dependency are tasks which
+ are not in the graph
+ """
+ if not (self.has_tasks(dependent) and self.has_tasks(dependency)):
+ raise TaskNotInGraphError()
+
+ if self.has_dependency(dependent, dependency):
+ return False
+
+ if isinstance(dependent, Iterable):
+ for dependent_task in dependent:
+ self.add_dependency(dependent_task, dependency)
+ else:
+ if isinstance(dependency, Iterable):
+ for dependency_task in dependency:
+ self.add_dependency(dependent, dependency_task)
+ else:
+ self._graph.add_edge(dependent.id, dependency.id)
+
+ return True
+
+ def has_dependency(self, dependent, dependency):
+ """
+ Check whether one item (task, sequence or parallel) depends on another
+
+ Note that if either of the items is either a sequence or a parallel,
+ and some of the dependencies exist in the graph but not all of them,
+ this method will return False
+
+ :param BaseTask|_TasksArrangement dependent: The dependent (task, sequence or parallel)
+ :param BaseTask|_TasksArrangement dependency: The dependency (task, sequence or parallel)
+ :return: True if the dependency between the two exists, otherwise False
+ :rtype: bool
+ :raise TaskNotInGraphError if either the dependent or dependency are tasks
+ which are not in the graph
+ """
+ if not (self.has_tasks(dependent) and self.has_tasks(dependency)):
+ raise TaskNotInGraphError()
+
+ return_value = True
+
+ if isinstance(dependent, Iterable):
+ for dependent_task in dependent:
+ return_value &= self.has_dependency(dependent_task, dependency)
+ else:
+ if isinstance(dependency, Iterable):
+ for dependency_task in dependency:
+ return_value &= self.has_dependency(dependent, dependency_task)
+ else:
+ return_value &= self._graph.has_edge(dependent.id, dependency.id)
+
+ return return_value
+
+ def remove_dependency(self, dependent, dependency):
+ """
+ Remove a dependency for one item (task, sequence or parallel) on another
+
+ Note that if either of the items is either a sequence or a parallel, and some of
+ the dependencies exist in the graph but not all of them, this method will not remove
+ any of the dependencies and return False
+
+ :param BaseTask|_TasksArrangement dependent: The dependent (task, sequence or parallel)
+ :param BaseTask|_TasksArrangement dependency: The dependency (task, sequence or parallel)
+ :return: False if the dependency between the two hadn't existed, otherwise True
+ :rtype: bool
+ :raise TaskNotInGraphError if either the dependent or dependency are tasks
+ which are not in the graph
+ """
+ if not (self.has_tasks(dependent) and self.has_tasks(dependency)):
+ raise TaskNotInGraphError()
+
+ if not self.has_dependency(dependent, dependency):
+ return False
+
+ if isinstance(dependent, Iterable):
+ for dependent_task in dependent:
+ self.remove_dependency(dependent_task, dependency)
+ else:
+ if isinstance(dependency, Iterable):
+ for dependency_task in dependency:
+ self.remove_dependency(dependent, dependency_task)
+ else:
+ self._graph.remove_edge(dependent.id, dependency.id)
+
+ return True
+
+ @_filter_out_empty_tasks
+ def sequence(self, *tasks):
+ """
+ Create and insert a sequence into the graph, effectively each task i depends on i-1
+ :param tasks: an iterable of dependencies
+ :return: the provided tasks
+ """
+ if tasks:
+ self.add_tasks(*tasks)
+
+ for i in xrange(1, len(tasks)):
+ self.add_dependency(tasks[i], tasks[i-1])
+
+ return tasks
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/aria/workflows/api/tasks_graph.py
----------------------------------------------------------------------
diff --git a/aria/workflows/api/tasks_graph.py b/aria/workflows/api/tasks_graph.py
deleted file mode 100644
index 5160345..0000000
--- a/aria/workflows/api/tasks_graph.py
+++ /dev/null
@@ -1,203 +0,0 @@
-# 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 uuid import uuid4
-
-from networkx import DiGraph, topological_sort
-
-from aria.tools.validation import ValidatorMixin
-
-
-class TaskNotFoundError(Exception):
- pass
-
-
-class TaskNotInGraphError(Exception):
- pass
-
-
-class TaskGraph(ValidatorMixin):
- """
- A task graph builder.
- Build a operations flow graph
- """
-
- def __init__(self, name):
- self.name = name
- self.id = str(uuid4())
- self.graph = DiGraph()
-
- def __getattr__(self, attr):
- try:
- return getattr(self.graph, attr)
- except AttributeError:
- return super(TaskGraph, self).__getattribute__(attr)
-
- def __repr__(self):
- return '{name}(id={self.id}, name={self.name}, graph={self.graph!r})'.format(
- name=self.__class__.__name__, self=self)
-
- @property
- def tasks(self):
- """
- An iterator on tasks added to the graph
- """
- for _, data in self.graph.nodes_iter(data=True):
- yield data['task']
-
- @property
- def leaf_tasks(self):
- for task in self.tasks_in_order():
- if not self.graph.predecessors(task.id):
- yield task
-
- def task_tree(self, reverse=False):
- """
- Iterates over the tasks to be executed in topological order and their dependencies.
- :param reverse: reverse the order
- """
- for task in self.tasks_in_order(reverse=reverse):
- yield task, self.task_dependencies(task)
-
- def tasks_in_order(self, reverse=False):
- """
- Iterates over the tasks to be executed in topological order
- :param reverse: reverse the order
- """
- for task_id in topological_sort(self.graph, reverse=reverse):
- yield self.graph.node[task_id]['task']
-
- def has_dependencies(self, task):
- return len(self.task_dependencies(task)) > 0
-
- def task_dependencies(self, task):
- """
- Iterates over the task dependencies
- """
- for task_ids in self.graph.edges_iter(task.id):
- for task_id in task_ids:
- if task.id != task_id:
- yield self.get_task(task_id)
-
- def add_task(self, task):
- """
- Add a task to this graph
- :param WorkflowTask|TaskGraph task: The task
- """
- self.graph.add_node(task.id, task=task)
-
- def get_task(self, task_id):
- """
- Get a task instance that was inserted to this graph by its id
-
- :param basestring task_id: the task id
- :return: requested task
- :rtype: WorkflowTask|TaskGraph
- :raise: TaskNotFoundError if no task found with given id
- """
- try:
- data = self.graph.node[task_id]
- return data['task']
- except KeyError:
- raise TaskNotFoundError('Task id: {0}'.format(task_id))
-
- def remove_task(self, task):
- """
- Remove the provided task from the graph
- :param WorkflowTask|graph task: The task
- """
- self.graph.remove_node(task.id)
-
- def dependency(self, source_task, after):
- """
- Add a dependency between tasks.
- The source task will only be executed after the target task terminates.
- A source task may depend on several tasks,
- In which case it will only be executed after all its target tasks will terminate.
-
- tasks flow order:
- after -> source_task
-
- :param WorkflowTask|TaskGraph source_task: The source task
- :type source_task: WorkflowTask
- :param list after: The target task
- :raise TaskNotInGraphError
- """
- if not self.graph.has_node(source_task.id):
- raise TaskNotInGraphError(
- 'source task {0!r} is not in graph (task id: {0.id})'.format(source_task))
- for target_task in after:
- if not self.graph.has_node(target_task.id):
- raise TaskNotInGraphError(
- 'target task {0!r} is not in graph (task id: {0.id})'.format(target_task))
- self.graph.add_edge(source_task.id, target_task.id)
-
- # workflow creation helper methods
- def chain(self, tasks, after=()):
- """
- create a chain of tasks.
- tasks will be added to the graph with a dependency between
- the tasks by order.
-
- tasks flow order:
- if tasks = (task0, task1, ..., taskn)
- after -> task0 -> task1 -> ... -> taskn
-
- :param tasks: list of WorkflowTask instances.
- :param after: target to the sequence
- """
- for source_task in tasks:
- self.add_task(source_task)
- self.dependency(source_task, after=after)
- after = (source_task,)
-
- def fan_out(self, tasks, after=()):
- """
- create a fan-out.
- tasks will be added to the graph with a dependency to
- the target task.
-
- tasks flow order:
- if tasks = (task0, task1, ..., taskn)
- after -> task0
- |-> task1
- |...
- \-> taskn
-
- :param tasks: list of WorkflowTask instances.
- :param after: target to the tasks
- """
- for source_task in tasks:
- self.add_task(source_task)
- self.dependency(source_task, after=after)
-
- def fan_in(self, source_task, after=None):
- """
- create a fan-in.
- source task will be added to the graph with a dependency to
- the tasks.
-
- tasks flow order:
- if after = (task0, task1, ..., taskn)
- task0\
- task1|-> source_task
- ... |
- taskn/
-
- :param source_task: source to the tasks
- :param after: list of WorkflowTask instances.
- """
- self.add_task(source_task)
- self.dependency(source_task, after=after)
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/aria/workflows/builtin/execute_operation.py
----------------------------------------------------------------------
diff --git a/aria/workflows/builtin/execute_operation.py b/aria/workflows/builtin/execute_operation.py
index 9e87c30..ddbb8e7 100644
--- a/aria/workflows/builtin/execute_operation.py
+++ b/aria/workflows/builtin/execute_operation.py
@@ -24,7 +24,7 @@ from .workflows import execute_operation_on_instance
@workflow
def execute_operation(
- context,
+ ctx,
graph,
operation,
operation_kwargs,
@@ -37,7 +37,7 @@ def execute_operation(
"""
The execute_operation workflow
- :param WorkflowContext context: the workflow context
+ :param WorkflowContext workflow_context: the workflow context
:param TaskGraph graph: the graph which will describe the workflow.
:param basestring operation: the operation name to execute
:param dict operation_kwargs:
@@ -52,7 +52,7 @@ def execute_operation(
subgraphs = {}
# filtering node instances
filtered_node_instances = list(_filter_node_instances(
- context=context,
+ context=ctx,
node_ids=node_ids,
node_instance_ids=node_instance_ids,
type_names=type_names))
@@ -60,31 +60,31 @@ def execute_operation(
if run_by_dependency_order:
filtered_node_instances_ids = set(node_instance.id
for node_instance in filtered_node_instances)
- for node_instance in context.node_instances:
+ for node_instance in ctx.node_instances:
if node_instance.id not in filtered_node_instances_ids:
- subgraphs[node_instance.id] = context.task_graph(
+ subgraphs[node_instance.id] = ctx.task_graph(
name='execute_operation_stub_{0}'.format(node_instance.id))
# registering actual tasks to sequences
for node_instance in filtered_node_instances:
- node_instance_sub_workflow = execute_operation_on_instance(
- context=context,
- graph=graph,
- node_instance=node_instance,
- operation=operation,
- operation_kwargs=operation_kwargs,
- allow_kwargs_override=allow_kwargs_override)
- subgraphs[node_instance.id] = node_instance_sub_workflow
+ graph.add_tasks(
+ execute_operation_on_instance(
+ node_instance=node_instance,
+ operation=operation,
+ operation_kwargs=operation_kwargs,
+ allow_kwargs_override=allow_kwargs_override
+ )
+ )
for _, node_instance_sub_workflow in subgraphs.items():
- graph.add_task(node_instance_sub_workflow)
+ graph.add_tasks(node_instance_sub_workflow)
# adding tasks dependencies if required
if run_by_dependency_order:
- for node_instance in context.node_instances:
+ for node_instance in ctx.node_instances:
for relationship_instance in node_instance.relationship_instances:
- graph.dependency(source_task=subgraphs[node_instance.id],
- after=[subgraphs[relationship_instance.target_id]])
+ graph.add_dependency(source_task=subgraphs[node_instance.id],
+ after=[subgraphs[relationship_instance.target_id]])
def _filter_node_instances(context, node_ids=(), node_instance_ids=(), type_names=()):
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/aria/workflows/builtin/heal.py
----------------------------------------------------------------------
diff --git a/aria/workflows/builtin/heal.py b/aria/workflows/builtin/heal.py
index cab2e6e..2f3d538 100644
--- a/aria/workflows/builtin/heal.py
+++ b/aria/workflows/builtin/heal.py
@@ -19,50 +19,49 @@ Builtin heal workflow
from aria import workflow
-from .uninstall import uninstall
-from .install import install
-from .workflows import relationship_tasks
+from .workflows import relationship_tasks, install_node_instance, uninstall_node_instance
+from ..api import task, task_graph
@workflow
-def heal(context, graph, node_instance_id):
+def heal(ctx, graph, node_instance_id):
"""
The heal workflow
- :param WorkflowContext context: the workflow context
+ :param WorkflowContext ctx: the workflow context
:param TaskGraph graph: the graph which will describe the workflow.
:param node_instance_id: the id of the node instance to heal
:return:
"""
- failing_node = context.storage.node_instance.get(node_instance_id)
- host_node = context.storage.node_instance.get(failing_node.host_id)
- failed_node_instance_subgraph = _get_contained_subgraph(context, host_node)
+ failing_node = ctx.model.node_instance.get(node_instance_id)
+ host_node = ctx.model.node_instance.get(failing_node.host_id)
+ failed_node_instance_subgraph = _get_contained_subgraph(ctx, host_node)
failed_node_instance_ids = list(n.id for n in failed_node_instance_subgraph)
- targeted_node_instances = [
- context.storage.node_instance.get(relationship_instance.target_id)
- for node_instance in failed_node_instance_subgraph
- for relationship_instance in node_instance.relationship_instances
- if relationship_instance.target_id not in failed_node_instance_ids
- ]
-
- graph.chain([
- heal_uninstall(
- context=context,
- failing_node_instances=failed_node_instance_subgraph,
- targeted_node_instances=targeted_node_instances),
- heal_install(
- context=context,
- failing_node_instances=failed_node_instance_subgraph,
- targeted_node_instances=targeted_node_instances)
- ])
+ targeted_node_instances = [node_instance for node_instance in ctx.node_instances
+ if node_instance.id not in failed_node_instance_ids]
+
+ uninstall_subgraph = task.WorkflowTask(
+ heal_uninstall,
+ failing_node_instances=failed_node_instance_subgraph,
+ targeted_node_instances=targeted_node_instances
+ )
+
+ install_subgraph = task.WorkflowTask(
+ heal_install,
+ failing_node_instances=failed_node_instance_subgraph,
+ targeted_node_instances=targeted_node_instances)
+
+ graph.add_tasks(uninstall_subgraph)
+ graph.add_tasks(install_subgraph)
+ graph.add_dependency(install_subgraph, uninstall_subgraph)
@workflow(suffix_template='{failing_node_instances}')
-def heal_uninstall(context, graph, failing_node_instances, targeted_node_instances):
+def heal_uninstall(ctx, graph, failing_node_instances, targeted_node_instances):
"""
the uninstall part of the heal mechanism
- :param WorkflowContext context: the workflow context
+ :param WorkflowContext ctx: the workflow context
:param TaskGraph graph: the task graph to edit.
:param failing_node_instances: the failing nodes to heal.
:param targeted_node_instances: the targets of the relationships where the failing node are
@@ -73,46 +72,53 @@ def heal_uninstall(context, graph, failing_node_instances, targeted_node_instanc
# Create install stub workflow for each unaffected node instance
for node_instance in targeted_node_instances:
- node_instance_sub_workflow = context.task_graph(
+ node_instance_sub_workflow = task_graph.TaskGraph(
name='uninstall_stub_{0}'.format(node_instance.id))
node_instance_sub_workflows[node_instance.id] = node_instance_sub_workflow
- graph.add_task(node_instance_sub_workflow)
+ graph.add_tasks(node_instance_sub_workflow)
- # Create install sub workflow for each failing node
- uninstall(
- context=context,
- graph=graph,
- node_instances=failing_node_instances,
- node_instance_sub_workflows=node_instance_sub_workflows)
+ # Create install sub workflow for each failing node on the same graph
+ node_instance_sub_workflows = node_instance_sub_workflows or {}
- # Add operations for intact nodes depending on a node instance
- # belonging to node_instances
+ # create install sub workflow for every node instance
+ for node_instance in failing_node_instances:
+ node_instance_sub_workflow = task.WorkflowTask(uninstall_node_instance,
+ node_instance=node_instance)
+ node_instance_sub_workflows[node_instance.id] = node_instance_sub_workflow
+ graph.add_tasks(node_instance_sub_workflow)
+
+ # create dependencies between the node instance sub workflow
+ for node_instance in failing_node_instances:
+ node_instance_sub_workflow = node_instance_sub_workflows[node_instance.id]
+ for relationship_instance in reversed(node_instance.relationship_instances):
+ graph.add_dependency(node_instance_sub_workflows[relationship_instance.target_id],
+ node_instance_sub_workflow)
+
+ # Add operations for intact nodes depending on a node instance belonging to node_instances
for node_instance in targeted_node_instances:
node_instance_sub_workflow = node_instance_sub_workflows[node_instance.id]
for relationship_instance in reversed(node_instance.relationship_instances):
- target_node_instance = context.storage.node_instance.get(
- relationship_instance.target_id)
- if target_node_instance in failing_node_instances:
- after_tasks = [node_instance_sub_workflows[relationship.target_id]
- for relationship in node_instance.relationship_instances]
+ target_node_instance = ctx.model.node_instance.get(relationship_instance.target_id)
+ target_node_instance_subgraph = node_instance_sub_workflows[target_node_instance.id]
+ graph.add_dependency(target_node_instance_subgraph, node_instance_sub_workflow)
- elif target_node_instance in targeted_node_instances:
- after_tasks = [relationship_tasks(
+ if target_node_instance in failing_node_instances:
+ dependency = relationship_tasks(
+ graph=graph,
node_instance=node_instance,
relationship_instance=relationship_instance,
- context=context,
- operation_name='aria.interfaces.relationship_lifecycle.unlink')]
+ context=ctx,
+ operation_name='aria.interfaces.relationship_lifecycle.unlink')
- if after_tasks:
- graph.dependency(source_task=node_instance_sub_workflow, after=after_tasks)
+ graph.add_dependency(node_instance_sub_workflow, dependency)
@workflow(suffix_template='{failing_node_instances}')
-def heal_install(context, graph, failing_node_instances, targeted_node_instances):
+def heal_install(ctx, graph, failing_node_instances, targeted_node_instances):
"""
the install part of the heal mechanism
- :param WorkflowContext context: the workflow context
+ :param WorkflowContext ctx: the workflow context
:param TaskGraph graph: the task graph to edit.
:param failing_node_instances: the failing nodes to heal.
:param targeted_node_instances: the targets of the relationships where the failing node are
@@ -123,17 +129,28 @@ def heal_install(context, graph, failing_node_instances, targeted_node_instances
# Create install sub workflow for each unaffected
for node_instance in targeted_node_instances:
- node_instance_sub_workflow = context.task_graph(
+ node_instance_sub_workflow = task_graph.TaskGraph(
name='install_stub_{0}'.format(node_instance.id))
node_instance_sub_workflows[node_instance.id] = node_instance_sub_workflow
- graph.add_task(node_instance_sub_workflow)
+ graph.add_tasks(node_instance_sub_workflow)
+
+ # create install sub workflow for each failing node on the same graph
+ node_instance_sub_workflows = node_instance_sub_workflows or {}
# create install sub workflow for every node instance
- install(
- context=context,
- graph=graph,
- node_instances=failing_node_instances,
- node_instance_sub_workflows=node_instance_sub_workflows)
+ for node_instance in failing_node_instances:
+ node_instance_sub_workflow = task.WorkflowTask(install_node_instance,
+ node_instance=node_instance)
+ node_instance_sub_workflows[node_instance.id] = node_instance_sub_workflow
+ graph.add_tasks(node_instance_sub_workflow)
+
+ # create dependencies between the node instance sub workflow
+ for node_instance in failing_node_instances:
+ node_instance_sub_workflow = node_instance_sub_workflows[node_instance.id]
+ if node_instance.relationship_instances:
+ dependencies = [node_instance_sub_workflows[relationship_instance.target_id]
+ for relationship_instance in node_instance.relationship_instances]
+ graph.add_dependency(node_instance_sub_workflow, dependencies)
# Add operations for intact nodes depending on a node instance
# belonging to node_instances
@@ -141,35 +158,33 @@ def heal_install(context, graph, failing_node_instances, targeted_node_instances
node_instance_sub_workflow = node_instance_sub_workflows[node_instance.id]
for relationship_instance in node_instance.relationship_instances:
- target_node_instance = context.storage.node_instance.get(
- relationship_instance.target_id)
- if target_node_instance in failing_node_instances:
- after_tasks = [node_instance_sub_workflows[relationship.target_id]
- for relationship in node_instance.relationship_instances]
+ target_node_instance = ctx.model.node_instance.get(relationship_instance.target_id)
+ target_node_instance_subworkflow = node_instance_sub_workflows[target_node_instance.id]
+ graph.add_dependency(node_instance_sub_workflow, target_node_instance_subworkflow)
- elif target_node_instance in targeted_node_instances:
- after_tasks = [relationship_tasks(
+ if target_node_instance in failing_node_instances:
+ dependent = relationship_tasks(
+ graph=graph,
node_instance=node_instance,
relationship_instance=relationship_instance,
- context=context,
- operation_name='aria.interfaces.relationship_lifecycle.establish')]
+ context=ctx,
+ operation_name='aria.interfaces.relationship_lifecycle.establish')
- if after_tasks:
- graph.dependency(source_task=node_instance_sub_workflow, after=after_tasks)
+ graph.add_dependency(dependent, node_instance_sub_workflow)
def _get_contained_subgraph(context, host_node_instance):
- contained_instances = set(node_instance
- for node_instance in context.node_instances
- if node_instance.host_id == host_node_instance.id and
- node_instance.id != node_instance.host_id)
- result = {host_node_instance}
+ contained_instances = [node_instance
+ for node_instance in context.node_instances
+ if node_instance.host_id == host_node_instance.id and
+ node_instance.id != node_instance.host_id]
+ result = [host_node_instance]
if not contained_instances:
return result
- result.update(contained_instances)
+ result.extend(contained_instances)
for node_instance in contained_instances:
- result.update(_get_contained_subgraph(context, node_instance))
+ result.extend(_get_contained_subgraph(context, node_instance))
- return result
+ return set(result)
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/aria/workflows/builtin/install.py
----------------------------------------------------------------------
diff --git a/aria/workflows/builtin/install.py b/aria/workflows/builtin/install.py
index 35d2968..0ab3ad6 100644
--- a/aria/workflows/builtin/install.py
+++ b/aria/workflows/builtin/install.py
@@ -20,13 +20,14 @@ Builtin install workflow
from aria import workflow
from .workflows import install_node_instance
+from ..api import task
@workflow
-def install(context, graph, node_instances=(), node_instance_sub_workflows=None):
+def install(ctx, graph, node_instances=(), node_instance_sub_workflows=None):
"""
The install workflow
- :param WorkflowContext context: the workflow context
+ :param WorkflowContext ctx: the workflow context
:param TaskGraph graph: the graph which will describe the workflow.
:param node_instances: the node instances on which to run the workflow
:param dict node_instance_sub_workflows: a dictionary of subworkflows with id as key and
@@ -34,21 +35,19 @@ def install(context, graph, node_instances=(), node_instance_sub_workflows=None)
:return:
"""
node_instance_sub_workflows = node_instance_sub_workflows or {}
- node_instances = node_instances or list(context.node_instances)
+ node_instances = node_instances or list(ctx.node_instances)
# create install sub workflow for every node instance
for node_instance in node_instances:
- node_instance_sub_workflow = install_node_instance(
- context=context,
- node_instance=node_instance)
+ node_instance_sub_workflow = task.WorkflowTask(install_node_instance,
+ node_instance=node_instance)
node_instance_sub_workflows[node_instance.id] = node_instance_sub_workflow
- graph.add_task(node_instance_sub_workflow)
+ graph.add_tasks(node_instance_sub_workflow)
# create dependencies between the node instance sub workflow
for node_instance in node_instances:
node_instance_sub_workflow = node_instance_sub_workflows[node_instance.id]
- graph.dependency(
- source_task=node_instance_sub_workflow,
- after=[
- node_instance_sub_workflows[relationship.target_id]
- for relationship in node_instance.relationship_instances])
+ if node_instance.relationship_instances:
+ dependencies = [node_instance_sub_workflows[relationship_instance.target_id]
+ for relationship_instance in node_instance.relationship_instances]
+ graph.add_dependency(node_instance_sub_workflow, dependencies)
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/aria/workflows/builtin/uninstall.py
----------------------------------------------------------------------
diff --git a/aria/workflows/builtin/uninstall.py b/aria/workflows/builtin/uninstall.py
index 47c8259..f4e965c 100644
--- a/aria/workflows/builtin/uninstall.py
+++ b/aria/workflows/builtin/uninstall.py
@@ -20,10 +20,11 @@ Builtin uninstall workflow
from aria import workflow
from .workflows import uninstall_node_instance
+from ..api import task
@workflow
-def uninstall(context, graph, node_instances=(), node_instance_sub_workflows=None):
+def uninstall(ctx, graph, node_instances=(), node_instance_sub_workflows=None):
"""
The uninstall workflow
:param WorkflowContext context: the workflow context
@@ -34,20 +35,18 @@ def uninstall(context, graph, node_instances=(), node_instance_sub_workflows=Non
:return:
"""
node_instance_sub_workflows = node_instance_sub_workflows or {}
- node_instances = node_instances or list(context.node_instances)
+ node_instances = node_instances or list(ctx.node_instances)
# create install sub workflow for every node instance
for node_instance in node_instances:
- node_instance_sub_workflow = uninstall_node_instance(
- context=context,
- node_instance=node_instance)
+ node_instance_sub_workflow = task.WorkflowTask(uninstall_node_instance,
+ node_instance=node_instance)
node_instance_sub_workflows[node_instance.id] = node_instance_sub_workflow
- graph.add_task(node_instance_sub_workflow)
+ graph.add_tasks(node_instance_sub_workflow)
# create dependencies between the node instance sub workflow
for node_instance in node_instances:
node_instance_sub_workflow = node_instance_sub_workflows[node_instance.id]
for relationship_instance in reversed(node_instance.relationship_instances):
- graph.dependency(
- source_task=node_instance_sub_workflows[relationship_instance.target_id],
- after=[node_instance_sub_workflow])
+ graph.add_dependency(node_instance_sub_workflows[relationship_instance.target_id],
+ node_instance_sub_workflow)
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/aria/workflows/builtin/workflows.py
----------------------------------------------------------------------
diff --git a/aria/workflows/builtin/workflows.py b/aria/workflows/builtin/workflows.py
index b6fbb94..4bff7cd 100644
--- a/aria/workflows/builtin/workflows.py
+++ b/aria/workflows/builtin/workflows.py
@@ -21,6 +21,8 @@ from itertools import groupby
from aria import workflow
+from ..api import task
+
__all__ = (
'install_node_instance',
@@ -32,43 +34,43 @@ __all__ = (
# Install node instance workflow and sub workflows
@workflow(suffix_template='{node_instance.id}')
-def install_node_instance(context, graph, node_instance):
+def install_node_instance(ctx, graph, node_instance):
"""
A workflow which installs a node instance.
- :param WorkflowContext context: the workflow context
+ :param WorkflowContext ctx: the workflow context
:param TaskGraph graph: the tasks graph of which to edit
:param node_instance: the node instance to install
:return:
"""
- create_node_instance = context.operation(
+ create_node_instance = task.OperationTask(
name='aria.interfaces.lifecycle.create.{0}'.format(node_instance.id),
- operation_details=node_instance.node.operations[
- 'aria.interfaces.lifecycle.create'],
+ operation_details=node_instance.node.operations['aria.interfaces.lifecycle.create'],
node_instance=node_instance
)
- configure_node_instance = context.operation(
+ configure_node_instance = task.OperationTask(
name='aria.interfaces.lifecycle.configure.{0}'.format(node_instance.id),
operation_details=node_instance.node.operations['aria.interfaces.lifecycle.configure'],
node_instance=node_instance
)
- start_node_instance = context.operation(
+ start_node_instance = task.OperationTask(
name='aria.interfaces.lifecycle.start.{0}'.format(node_instance.id),
- operation_details=node_instance.node.operations[
- 'aria.interfaces.lifecycle.start'],
+ operation_details=node_instance.node.operations['aria.interfaces.lifecycle.start'],
node_instance=node_instance
)
- graph.chain(tasks=[
+
+ graph.sequence(
create_node_instance,
- preconfigure_relationship(context=context, node_instance=node_instance),
+ preconfigure_relationship(graph, ctx, node_instance),
configure_node_instance,
- postconfigure_relationship(context=context, node_instance=node_instance),
+ postconfigure_relationship(graph, ctx, node_instance),
start_node_instance,
- establish_relationship(context=context, node_instance=node_instance),
- ])
+ establish_relationship(graph, ctx, node_instance)
+ )
+ return graph
-@workflow(suffix_template='{node_instance.id}')
-def preconfigure_relationship(context, graph, node_instance):
+
+def preconfigure_relationship(graph, ctx, node_instance):
"""
:param context:
@@ -76,14 +78,14 @@ def preconfigure_relationship(context, graph, node_instance):
:param node_instance:
:return:
"""
- graph.chain(tasks=relationships_tasks(
+ return relationships_tasks(
+ graph=graph,
operation_name='aria.interfaces.relationship_lifecycle.preconfigure',
- context=context,
- node_instance=node_instance))
+ context=ctx,
+ node_instance=node_instance)
-@workflow(suffix_template='{node_instance.id}')
-def postconfigure_relationship(context, graph, node_instance):
+def postconfigure_relationship(graph, ctx, node_instance):
"""
:param context:
@@ -91,14 +93,14 @@ def postconfigure_relationship(context, graph, node_instance):
:param node_instance:
:return:
"""
- graph.chain(tasks=relationships_tasks(
+ return relationships_tasks(
+ graph=graph,
operation_name='aria.interfaces.relationship_lifecycle.postconfigure',
- context=context,
- node_instance=node_instance))
+ context=ctx,
+ node_instance=node_instance)
-@workflow(suffix_template='{node_instance.id}')
-def establish_relationship(context, graph, node_instance):
+def establish_relationship(graph, ctx, node_instance):
"""
:param context:
@@ -106,16 +108,17 @@ def establish_relationship(context, graph, node_instance):
:param node_instance:
:return:
"""
- graph.chain(tasks=relationships_tasks(
+ return relationships_tasks(
+ graph=graph,
operation_name='aria.interfaces.relationship_lifecycle.establish',
- context=context,
- node_instance=node_instance))
+ context=ctx,
+ node_instance=node_instance)
# Uninstall node instance workflow and subworkflows
@workflow(suffix_template='{node_instance.id}')
-def uninstall_node_instance(graph, context, node_instance):
+def uninstall_node_instance(ctx, graph, node_instance):
"""
A workflow which uninstalls a node instance.
:param WorkflowContext context: the workflow context
@@ -123,28 +126,25 @@ def uninstall_node_instance(graph, context, node_instance):
:param node_instance: the node instance to uninstall
:return:
"""
- stop_node_instance = context.operation(
+ stop_node_instance = task.OperationTask(
name='aria.interfaces.lifecycle.stop.{0}'.format(node_instance.id),
- operation_details=node_instance.node.operations[
- 'aria.interfaces.lifecycle.stop'],
+ operation_details=node_instance.node.operations['aria.interfaces.lifecycle.stop'],
node_instance=node_instance
)
- delete_node_instance = context.operation(
+ delete_node_instance = task.OperationTask(
name='aria.interfaces.lifecycle.delete.{0}'.format(node_instance.id),
- operation_details=node_instance.node.operations[
- 'aria.interfaces.lifecycle.delete'],
+ operation_details=node_instance.node.operations['aria.interfaces.lifecycle.delete'],
node_instance=node_instance
)
- graph.chain(tasks=[
+ graph.sequence(
stop_node_instance,
- unlink_relationship(context=context, node_instance=node_instance),
- delete_node_instance,
- ])
+ unlink_relationship(graph, ctx, node_instance),
+ delete_node_instance
+ )
-@workflow(suffix_template='{node_instance.id}')
-def unlink_relationship(context, graph, node_instance):
+def unlink_relationship(graph, ctx, node_instance):
"""
:param context:
@@ -152,19 +152,15 @@ def unlink_relationship(context, graph, node_instance):
:param node_instance:
:return:
"""
- tasks = relationships_tasks(
+ return relationships_tasks(
+ graph=graph,
operation_name='aria.interfaces.relationship_lifecycle.unlink',
- context=context,
+ context=ctx,
node_instance=node_instance
)
- graph.chain(tasks=tasks)
- return tasks
-@workflow(suffix_template='{node_instance.id}.{operation}')
def execute_operation_on_instance(
- context,
- graph,
node_instance,
operation,
operation_kwargs,
@@ -187,16 +183,18 @@ def execute_operation_on_instance(
node_instance=node_instance,
operation_name=operation)
- graph.add_task(
- context.operation(
- name=task_name,
- operation_details=node_instance.node.operations[operation],
- node_instance=node_instance,
- parameters=operation_kwargs)
- )
+ return task.OperationTask(
+ name=task_name,
+ operation_details=node_instance.node.operations[operation],
+ node_instance=node_instance,
+ inputs=operation_kwargs)
-def relationships_tasks(operation_name, context, node_instance):
+
+def relationships_tasks(graph,
+ operation_name,
+ context,
+ node_instance):
"""
Creates a relationship task (source and target) for all of a node_instance relationships.
:param basestring operation_name: the relationship operation name.
@@ -211,17 +209,24 @@ def relationships_tasks(operation_name, context, node_instance):
sub_tasks = []
for index, (_, relationship_group) in enumerate(relationships_groups):
for relationship_instance in relationship_group:
- relationship_subgraph = relationship_tasks(
+ relationship_operations = relationship_tasks(
+ graph=graph,
node_instance=node_instance,
relationship_instance=relationship_instance,
context=context,
operation_name=operation_name,
index=index)
- sub_tasks.append(relationship_subgraph)
- return sub_tasks
+ sub_tasks.append(relationship_operations)
+
+ return graph.sequence(*sub_tasks) if sub_tasks else sub_tasks
-def relationship_tasks(node_instance, relationship_instance, context, operation_name, index=None):
+def relationship_tasks(graph,
+ node_instance,
+ relationship_instance,
+ context,
+ operation_name,
+ index=None):
"""
Creates a relationship task source and target.
:param NodeInstance node_instance: the node instance of the relationship
@@ -232,29 +237,20 @@ def relationship_tasks(node_instance, relationship_instance, context, operation_
:return:
"""
index = index or node_instance.relationship_instances.index(relationship_instance)
- sub_workflow_name = '{name}.{index}.{node_instance.id}'.format(
- name=operation_name,
- index=index,
- node_instance=node_instance,
- )
operation_name_template = '{name}.{index}.{{0}}.<{source_id}, {target_id}>'.format(
name=operation_name,
index=index,
source_id=node_instance.id,
target_id=relationship_instance.target_id,
)
- source_operation = context.operation(
+ source_operation = task.OperationTask(
name=operation_name_template.format('source'),
node_instance=node_instance,
operation_details=relationship_instance.relationship.source_operations[
operation_name])
- target_operation = context.operation(
+ target_operation = task.OperationTask(
name=operation_name_template.format('target'),
- node_instance=context.storage.node_instance.get(
- relationship_instance.target_id),
+ node_instance=context.model.node_instance.get(relationship_instance.target_id),
operation_details=relationship_instance.relationship.target_operations[
operation_name])
- sub_graph = context.task_graph(name=sub_workflow_name)
- sub_graph.add_task(source_operation)
- sub_graph.add_task(target_operation)
- return sub_graph
+ return graph.add_tasks(source_operation, target_operation)
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/aria/workflows/core/__init__.py
----------------------------------------------------------------------
diff --git a/aria/workflows/core/__init__.py b/aria/workflows/core/__init__.py
index ae1e83e..08ccff7 100644
--- a/aria/workflows/core/__init__.py
+++ b/aria/workflows/core/__init__.py
@@ -12,3 +12,5 @@
# 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 . import translation, task
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/aria/workflows/core/executor.py
----------------------------------------------------------------------
diff --git a/aria/workflows/core/executor.py b/aria/workflows/core/executor.py
index ace445a..0a5cd67 100644
--- a/aria/workflows/core/executor.py
+++ b/aria/workflows/core/executor.py
@@ -134,8 +134,7 @@ class MultiprocessExecutor(BaseExecutor):
self._listener_thread = threading.Thread(target=self._listener)
self._listener_thread.daemon = True
self._listener_thread.start()
- self._pool = multiprocessing.Pool(processes=pool_size,
- maxtasksperchild=1)
+ self._pool = multiprocessing.Pool(processes=pool_size)
def execute(self, task):
self._tasks[task.id] = task
[2/3] incubator-ariatosca git commit: ARIA-4 Create an API for
creating workflows
Posted by mx...@apache.org.
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/aria/workflows/core/task.py
----------------------------------------------------------------------
diff --git a/aria/workflows/core/task.py b/aria/workflows/core/task.py
new file mode 100644
index 0000000..de8a561
--- /dev/null
+++ b/aria/workflows/core/task.py
@@ -0,0 +1,106 @@
+# 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.
+
+"""
+Workflow tasks
+"""
+
+
+class BaseTask(object):
+ """
+ Base class for Task objects
+ """
+
+ def __init__(self, id):
+ self._id = id
+
+ @property
+ def id(self):
+ """
+ :return: the task's id
+ """
+ return self._id
+
+
+class StartWorkflowTask(BaseTask):
+ """
+ Tasks marking a workflow start
+ """
+ pass
+
+
+class EndWorkflowTask(BaseTask):
+ """
+ Tasks marking a workflow end
+ """
+ pass
+
+
+class StartSubWorkflowTask(BaseTask):
+ """
+ Tasks marking a subworkflow start
+ """
+ pass
+
+
+class EndSubWorkflowTask(BaseTask):
+ """
+ Tasks marking a subworkflow end
+ """
+ pass
+
+
+class OperationTask(BaseTask):
+ """
+ Operation tasks
+ """
+
+ def __init__(self, api_task, *args, **kwargs):
+ self._api_task = api_task
+ self._ctx = api_task.workflow_context
+ super(OperationTask, self).__init__(id=api_task.id)
+ self._create_operation_in_storage()
+
+ @property
+ def ctx(self):
+ """
+ :return: the task's name
+ """
+ return self._ctx
+
+ @property
+ def api_task(self):
+ """
+ :return: the task's context
+ """
+ return self._api_task
+
+ def _create_operation_in_storage(self):
+ task = self.ctx.model.task.model_cls(
+ name=self.api_task.name,
+ operation_details=self.api_task.operation_details,
+ node_instance=self.api_task.node_instance,
+ inputs=self.api_task.operation_kwargs,
+
+ execution_id=self.ctx.execution_id,
+ max_retries=self.ctx.parameters.get('max_retries', 1),
+ )
+ self.ctx.model.task.store(task)
+
+ def __getattr__(self, attr):
+ try:
+ return super(OperationTask, self).__getattribute__(attr)
+ except AttributeError:
+ return getattr(self.context, attr)
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/aria/workflows/core/tasks.py
----------------------------------------------------------------------
diff --git a/aria/workflows/core/tasks.py b/aria/workflows/core/tasks.py
deleted file mode 100644
index 76ae609..0000000
--- a/aria/workflows/core/tasks.py
+++ /dev/null
@@ -1,104 +0,0 @@
-# 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.
-
-"""
-Workflow tasks
-"""
-
-
-class BaseTask(object):
- """
- Base class for Task objects
- """
-
- def __init__(self, id, name, context):
- self._id = id
- self._name = name
- self._context = context
-
- @property
- def id(self):
- """
- :return: the task's id
- """
- return self._id
-
- @property
- def name(self):
- """
- :return: the task's name
- """
- return self._name
-
- @property
- def context(self):
- """
- :return: the task's context
- """
- return self._context
-
-
-class StartWorkflowTask(BaseTask):
- """
- Tasks marking a workflow start
- """
- pass
-
-
-class EndWorkflowTask(BaseTask):
- """
- Tasks marking a workflow end
- """
- pass
-
-
-class StartSubWorkflowTask(BaseTask):
- """
- Tasks marking a subworkflow start
- """
- pass
-
-
-class EndSubWorkflowTask(BaseTask):
- """
- Tasks marking a subworkflow end
- """
- pass
-
-
-class OperationTask(BaseTask):
- """
- Operation tasks
- """
-
- def __init__(self, *args, **kwargs):
- super(OperationTask, self).__init__(*args, **kwargs)
- self._create_operation_in_storage()
-
- def _create_operation_in_storage(self):
- operation_cls = self.context.storage.operation.model_cls
- operation = operation_cls(
- id=self.context.id,
- execution_id=self.context.execution_id,
- max_retries=self.context.parameters.get('max_retries', 1),
- status=operation_cls.PENDING,
- )
- self.context.operation = operation
-
- def __getattr__(self, attr):
- try:
- return getattr(self.context, attr)
- except AttributeError:
- return super(OperationTask, self).__getattribute__(attr)
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/aria/workflows/core/translation.py
----------------------------------------------------------------------
diff --git a/aria/workflows/core/translation.py b/aria/workflows/core/translation.py
index dc483c6..ec44f48 100644
--- a/aria/workflows/core/translation.py
+++ b/aria/workflows/core/translation.py
@@ -17,17 +17,15 @@
Translation of user graph's API to the execution graph
"""
-from aria import contexts
-
-from . import tasks
+from . import task
+from .. import api
def build_execution_graph(
task_graph,
- workflow_context,
execution_graph,
- start_cls=tasks.StartWorkflowTask,
- end_cls=tasks.EndWorkflowTask,
+ start_cls=task.StartWorkflowTask,
+ end_cls=task.EndWorkflowTask,
depends_on=()):
"""
Translates the user graph to the execution graph
@@ -39,43 +37,36 @@ def build_execution_graph(
:param depends_on: internal use
"""
# Insert start marker
- start_task = start_cls(id=_start_graph_suffix(task_graph.id),
- name=_start_graph_suffix(task_graph.name),
- context=workflow_context)
+ start_task = start_cls(id=_start_graph_suffix(task_graph.id))
_add_task_and_dependencies(execution_graph, start_task, depends_on)
- for operation_or_workflow, dependencies in task_graph.task_tree(reverse=True):
+ for api_task in task_graph.topological_order(reverse=True):
+ dependencies = task_graph.get_dependencies(api_task)
operation_dependencies = _get_tasks_from_dependencies(
execution_graph,
dependencies,
default=[start_task])
- if _is_operation(operation_or_workflow):
+ if _is_operation(api_task):
# Add the task an the dependencies
- operation_task = tasks.OperationTask(id=operation_or_workflow.id,
- name=operation_or_workflow.name,
- context=operation_or_workflow)
+ operation_task = task.OperationTask(api_task)
_add_task_and_dependencies(execution_graph, operation_task, operation_dependencies)
else:
# Built the graph recursively while adding start and end markers
build_execution_graph(
- task_graph=operation_or_workflow,
- workflow_context=workflow_context,
+ task_graph=api_task,
execution_graph=execution_graph,
- start_cls=tasks.StartSubWorkflowTask,
- end_cls=tasks.EndSubWorkflowTask,
+ start_cls=task.StartSubWorkflowTask,
+ end_cls=task.EndSubWorkflowTask,
depends_on=operation_dependencies
)
# Insert end marker
workflow_dependencies = _get_tasks_from_dependencies(
execution_graph,
- task_graph.leaf_tasks,
+ _get_non_dependency_tasks(task_graph),
default=[start_task])
- end_task = end_cls(
- id=_end_graph_suffix(task_graph.id),
- name=_end_graph_suffix(task_graph.name),
- context=workflow_context)
+ end_task = end_cls(id=_end_graph_suffix(task_graph.id))
_add_task_and_dependencies(execution_graph, end_task, workflow_dependencies)
@@ -95,7 +86,7 @@ def _get_tasks_from_dependencies(execution_graph, dependencies, default=()):
def _is_operation(task):
- return isinstance(task, contexts.OperationContext)
+ return isinstance(task, api.task.OperationTask)
def _start_graph_suffix(id):
@@ -104,3 +95,9 @@ def _start_graph_suffix(id):
def _end_graph_suffix(id):
return '{0}-End'.format(id)
+
+
+def _get_non_dependency_tasks(graph):
+ for task in graph.tasks:
+ if len(list(graph.get_dependents(task))) == 0:
+ yield task
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/requirements.txt
----------------------------------------------------------------------
diff --git a/requirements.txt b/requirements.txt
index 1240b72..b585039 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -11,6 +11,7 @@
# limitations under the License.
PyYAML==3.10
+billiard==3.3.0.20
networkx==1.9
requests==2.7.0
retrying==1.3.3
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/setup.py
----------------------------------------------------------------------
diff --git a/setup.py b/setup.py
index 811810b..d2c60bb 100644
--- a/setup.py
+++ b/setup.py
@@ -41,6 +41,11 @@ try:
except ImportError:
install_requires.append('importlib')
+try:
+ from collections import OrderedDict
+except ImportError:
+ install_requires.append('ordereddict==1.1')
+
setup(
name=_PACKAGE_NAME,
version=version,
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/tests/.pylintrc
----------------------------------------------------------------------
diff --git a/tests/.pylintrc b/tests/.pylintrc
index f6cfd7a..59cca90 100644
--- a/tests/.pylintrc
+++ b/tests/.pylintrc
@@ -62,7 +62,7 @@ confidence=
# --enable=similarities". If you want to run only the classes checker, but have
# no Warning level messages displayed, use"--disable=all --enable=classes
# --disable=W"
-disable=import-star-module-level,old-octal-literal,oct-method,print-statement,unpacking-in-except,parameter-unpacking,backtick,old-raise-syntax,old-ne-operator,long-suffix,dict-view-method,dict-iter-method,metaclass-assignment,next-method-called,raising-string,indexing-exception,raw_input-builtin,long-builtin,file-builtin,execfile-builtin,coerce-builtin,cmp-builtin,buffer-builtin,basestring-builtin,apply-builtin,filter-builtin-not-iterating,using-cmp-argument,useless-suppression,range-builtin-not-iterating,suppressed-message,no-absolute-import,old-division,cmp-method,reload-builtin,zip-builtin-not-iterating,intern-builtin,unichr-builtin,reduce-builtin,standarderror-builtin,unicode-builtin,xrange-builtin,coerce-method,delslice-method,getslice-method,setslice-method,input-builtin,round-builtin,hex-method,nonzero-method,map-builtin-not-iterating,redefined-builtin,no-self-use,missing-docstring,attribute-defined-outside-init
+disable=import-star-module-level,old-octal-literal,oct-method,print-statement,unpacking-in-except,parameter-unpacking,backtick,old-raise-syntax,old-ne-operator,long-suffix,dict-view-method,dict-iter-method,metaclass-assignment,next-method-called,raising-string,indexing-exception,raw_input-builtin,long-builtin,file-builtin,execfile-builtin,coerce-builtin,cmp-builtin,buffer-builtin,basestring-builtin,apply-builtin,filter-builtin-not-iterating,using-cmp-argument,useless-suppression,range-builtin-not-iterating,suppressed-message,no-absolute-import,old-division,cmp-method,reload-builtin,zip-builtin-not-iterating,intern-builtin,unichr-builtin,reduce-builtin,standarderror-builtin,unicode-builtin,xrange-builtin,coerce-method,delslice-method,getslice-method,setslice-method,input-builtin,round-builtin,hex-method,nonzero-method,map-builtin-not-iterating,redefined-builtin,no-self-use,missing-docstring,attribute-defined-outside-init,redefined-outer-name,import-error
[REPORTS]
@@ -339,7 +339,7 @@ max-args=15
ignored-argument-names=_.*
# Maximum number of locals for function / method body
-max-locals=15
+max-locals=30
# Maximum number of return / yield for function / method body
max-returns=6
@@ -360,7 +360,7 @@ max-attributes=15
min-public-methods=0
# Maximum number of public methods for a class (see R0904).
-max-public-methods=20
+max-public-methods=50
# Maximum number of boolean expressions in a if statement
max-bool-expr=5
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/tests/mock/__init__.py
----------------------------------------------------------------------
diff --git a/tests/mock/__init__.py b/tests/mock/__init__.py
new file mode 100644
index 0000000..14541d0
--- /dev/null
+++ b/tests/mock/__init__.py
@@ -0,0 +1,16 @@
+# 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 . import models, context, operations
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/tests/mock/context.py
----------------------------------------------------------------------
diff --git a/tests/mock/context.py b/tests/mock/context.py
new file mode 100644
index 0000000..71df67c
--- /dev/null
+++ b/tests/mock/context.py
@@ -0,0 +1,36 @@
+# 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 import context, application_model_storage
+
+from ..storage import InMemoryModelDriver
+from . import models
+
+
+def simple():
+ """
+ Create the following graph in storage:
+ dependency_node <------ dependent_node
+ :return:
+ """
+ storage = application_model_storage(InMemoryModelDriver())
+ storage.setup()
+ return context.workflow.WorkflowContext(
+ model_storage=storage,
+ resource_storage=None,
+ deployment_id=models.DEPLOYMENT_ID,
+ workflow_id=models.WORKFLOW_ID,
+ execution_id=models.EXECUTION_ID
+ )
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/tests/mock/models.py
----------------------------------------------------------------------
diff --git a/tests/mock/models.py b/tests/mock/models.py
new file mode 100644
index 0000000..633adbb
--- /dev/null
+++ b/tests/mock/models.py
@@ -0,0 +1,132 @@
+# 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 datetime import datetime
+
+from aria.storage import models
+
+from . import operations
+
+DEPLOYMENT_ID = 'test_deployment_id'
+BLUEPRINT_ID = 'test_blueprint_id'
+WORKFLOW_ID = 'test_workflow_id'
+EXECUTION_ID = 'test_execution_id'
+
+
+def get_dependency_node():
+ return models.Node(
+ id='dependency_node',
+ host_id='dependency_node',
+ blueprint_id=BLUEPRINT_ID,
+ type='test_node_type',
+ type_hierarchy=[],
+ number_of_instances=1,
+ planned_number_of_instances=1,
+ deploy_number_of_instances=1,
+ properties={},
+ operations=dict((key, {}) for key in operations.NODE_OPERATIONS),
+ relationships=[],
+ min_number_of_instances=1,
+ max_number_of_instances=1,
+ )
+
+
+def get_dependency_node_instance(dependency_node=None):
+ return models.NodeInstance(
+ id='dependency_node_instance',
+ host_id='dependency_node_instance',
+ deployment_id=DEPLOYMENT_ID,
+ runtime_properties={},
+ version=None,
+ relationship_instances=[],
+ node=dependency_node or get_dependency_node()
+ )
+
+
+def get_relationship(target=None):
+ return models.Relationship(
+ target_id=target.id or get_dependency_node().id,
+ source_interfaces={},
+ source_operations=dict((key, {}) for key in operations.RELATIONSHIP_OPERATIONS),
+ target_interfaces={},
+ target_operations=dict((key, {}) for key in operations.RELATIONSHIP_OPERATIONS),
+ type='rel_type',
+ type_hierarchy=[],
+ properties={},
+ )
+
+
+def get_relationship_instance(target_instance=None, relationship=None):
+ return models.RelationshipInstance(
+ target_id=target_instance.id or get_dependency_node_instance().id,
+ target_name='test_target_name',
+ type='some_type',
+ relationship=relationship or get_relationship(target_instance.node
+ if target_instance else None)
+ )
+
+
+def get_dependent_node(relationship=None):
+ return models.Node(
+ id='dependent_node',
+ host_id='dependent_node',
+ blueprint_id=BLUEPRINT_ID,
+ type='test_node_type',
+ type_hierarchy=[],
+ number_of_instances=1,
+ planned_number_of_instances=1,
+ deploy_number_of_instances=1,
+ properties={},
+ operations=dict((key, {}) for key in operations.NODE_OPERATIONS),
+ relationships=[relationship or get_relationship()],
+ min_number_of_instances=1,
+ max_number_of_instances=1,
+ )
+
+
+def get_dependent_node_instance(relationship_instance, dependent_node=None):
+ return models.NodeInstance(
+ id='dependent_node_instance',
+ host_id='dependent_node_instance',
+ deployment_id=DEPLOYMENT_ID,
+ runtime_properties={},
+ version=None,
+ relationship_instances=[relationship_instance or get_relationship_instance()],
+ node=dependent_node or get_dependency_node()
+ )
+
+
+def get_execution():
+ return models.Execution(
+ id=EXECUTION_ID,
+ status=models.Execution.STARTED,
+ deployment_id=DEPLOYMENT_ID,
+ workflow_id=WORKFLOW_ID,
+ blueprint_id=BLUEPRINT_ID,
+ started_at=datetime.now(),
+ parameters=None
+ )
+
+
+def get_deployment():
+ now = datetime.now()
+ return models.Deployment(
+ id=DEPLOYMENT_ID,
+ description=None,
+ created_at=now,
+ updated_at=now,
+ blueprint_id=BLUEPRINT_ID,
+ workflows={}
+ )
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/tests/mock/operations.py
----------------------------------------------------------------------
diff --git a/tests/mock/operations.py b/tests/mock/operations.py
new file mode 100644
index 0000000..407061f
--- /dev/null
+++ b/tests/mock/operations.py
@@ -0,0 +1,33 @@
+# 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.
+
+NODE_OPERATIONS_INSTALL = [
+ 'aria.interfaces.lifecycle.create',
+ 'aria.interfaces.lifecycle.configure',
+ 'aria.interfaces.lifecycle.start',
+ ]
+NODE_OPERATIONS_UNINSTALL = [
+ 'aria.interfaces.lifecycle.stop',
+ 'aria.interfaces.lifecycle.delete',
+]
+NODE_OPERATIONS = NODE_OPERATIONS_INSTALL + NODE_OPERATIONS_UNINSTALL
+
+RELATIONSHIP_OPERATIONS_INSTALL = [
+ 'aria.interfaces.relationship_lifecycle.preconfigure',
+ 'aria.interfaces.relationship_lifecycle.postconfigure',
+ 'aria.interfaces.relationship_lifecycle.establish',
+ ]
+RELATIONSHIP_OPERATIONS_UNINSTALL = ['aria.interfaces.relationship_lifecycle.unlink']
+RELATIONSHIP_OPERATIONS = RELATIONSHIP_OPERATIONS_INSTALL + RELATIONSHIP_OPERATIONS_UNINSTALL
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/tests/test_logger.py
----------------------------------------------------------------------
diff --git a/tests/test_logger.py b/tests/test_logger.py
index 7891dd6..37731bb 100644
--- a/tests/test_logger.py
+++ b/tests/test_logger.py
@@ -80,7 +80,7 @@ def test_create_file_log_handler():
assert handler.baseFilename == temp_file.name
assert handler.maxBytes == 5 * 1000 * 1024
assert handler.backupCount == 10
- assert handler.delay is True
+ assert handler.stream is None
assert handler.level == logging.DEBUG
assert handler.formatter == _default_file_formatter
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/tests/workflows/__init__.py
----------------------------------------------------------------------
diff --git a/tests/workflows/__init__.py b/tests/workflows/__init__.py
index ae1e83e..fe04b2f 100644
--- a/tests/workflows/__init__.py
+++ b/tests/workflows/__init__.py
@@ -12,3 +12,5 @@
# 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 . import api, builtin, core
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/tests/workflows/api/__init__.py
----------------------------------------------------------------------
diff --git a/tests/workflows/api/__init__.py b/tests/workflows/api/__init__.py
new file mode 100644
index 0000000..e69de29
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/tests/workflows/api/test_task.py
----------------------------------------------------------------------
diff --git a/tests/workflows/api/test_task.py b/tests/workflows/api/test_task.py
new file mode 100644
index 0000000..0edacf7
--- /dev/null
+++ b/tests/workflows/api/test_task.py
@@ -0,0 +1,120 @@
+# 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 datetime import datetime
+
+import pytest
+
+from aria import context
+from aria.storage import models
+from aria.workflows.api import task
+
+from ... import mock
+
+
+@pytest.fixture()
+def ctx():
+ simple_context = mock.context.simple()
+ dependency_node = mock.models.get_dependency_node()
+ dependency_node_instance = mock.models.get_dependency_node_instance(
+ dependency_node=dependency_node)
+
+ relationship = mock.models.get_relationship(dependency_node)
+ relationship_instance = mock.models.get_relationship_instance(
+ relationship=relationship,
+ target_instance=dependency_node_instance
+ )
+
+ dependent_node = mock.models.get_dependent_node(relationship)
+ dependent_node_instance = mock.models.get_dependent_node_instance(
+ dependent_node=dependent_node,
+ relationship_instance=relationship_instance
+ )
+
+ simple_context.model.node.store(dependent_node)
+ simple_context.model.node.store(dependency_node)
+ simple_context.model.node_instance.store(dependent_node_instance)
+ simple_context.model.node_instance.store(dependency_node_instance)
+ simple_context.model.relationship.store(relationship)
+ simple_context.model.relationship_instance.store(relationship_instance)
+ simple_context.model.execution.store(mock.models.get_execution())
+ simple_context.model.deployment.store(mock.models.get_deployment())
+
+ return simple_context
+
+
+class TestOperationTask(object):
+
+ def test_operation_task_creation(self):
+ workflow_context = mock.context.simple()
+
+ now = datetime.now()
+ name = 'task_name'
+ op_details = {'operation_details': True}
+ node_instance = mock.models.get_dependency_node_instance()
+ inputs = {'inputs': True}
+ task_id = 'task_id'
+ status = models.Task.PENDING
+ eta = now
+ started_at = now
+ ended_at = now
+ max_retries = 3
+ retry_count = 1
+
+ with context.workflow.current.push(workflow_context):
+ model_task = task.OperationTask(name=name,
+ operation_details=op_details,
+ node_instance=node_instance,
+ inputs=inputs,
+ id=task_id,
+ status=status,
+ eta=eta,
+ started_at=started_at,
+ ended_at=ended_at,
+ max_retries=max_retries,
+ retry_count=retry_count)
+
+ assert model_task.name == name
+ assert model_task.operation_details == op_details
+ assert model_task.node_instance == node_instance
+ assert model_task.operation_kwargs['inputs'] == inputs
+ assert model_task.operation_kwargs['id'] == task_id
+ assert model_task.operation_kwargs['status'] == status
+ assert model_task.operation_kwargs['eta'] == eta
+ assert model_task.operation_kwargs['started_at'] == started_at
+ assert model_task.operation_kwargs['ended_at'] == ended_at
+ assert model_task.operation_kwargs['max_retries'] == max_retries
+ assert model_task.operation_kwargs['retry_count'] == retry_count
+
+
+class TestWorkflowTask(object):
+
+ def test_workflow_task_creation(self, ctx):
+
+ workspace = {}
+
+ def sub_workflow(**kwargs):
+ workspace.update(kwargs)
+ return True
+
+ with context.workflow.current.push(ctx):
+ workflow_task = task.WorkflowTask(sub_workflow, kwarg='workflow_kwarg')
+ assert workflow_task.blueprint_id == ctx.blueprint_id
+ assert workflow_task.deployment == ctx.deployment
+ assert list(workflow_task.nodes) == list(ctx.nodes)
+ assert list(workflow_task.node_instances) == list(ctx.node_instances)
+ assert workflow_task.execution == ctx.execution
+ assert workspace == {'kwarg': 'workflow_kwarg', 'ctx': ctx}
+ assert workflow_task.graph is True
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/tests/workflows/api/test_task_graph.py
----------------------------------------------------------------------
diff --git a/tests/workflows/api/test_task_graph.py b/tests/workflows/api/test_task_graph.py
new file mode 100644
index 0000000..7d409cf
--- /dev/null
+++ b/tests/workflows/api/test_task_graph.py
@@ -0,0 +1,714 @@
+# 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 uuid import uuid4
+
+import pytest
+
+from aria.workflows.api import task_graph, task
+
+
+class MockTask(task.BaseTask):
+ def __init__(self):
+ self.id = str(uuid4())
+
+
+@pytest.fixture
+def graph():
+ return task_graph.TaskGraph(name='mock-graph')
+
+
+class TestTaskGraphTasks(object):
+
+ def test_add_task(self, graph):
+ task = MockTask()
+ add_result = graph.add_tasks(task)
+ assert add_result == [task]
+ tasks = [t for t in graph.tasks]
+ assert len(tasks) == 1
+ assert tasks[0] == task
+
+ def test_add_existing_task(self, graph):
+ task = MockTask()
+ graph.add_tasks(task)
+ # adding a task already in graph - should have no effect, and return False
+ add_result = graph.add_tasks(task)
+ assert add_result == []
+ tasks = [t for t in graph.tasks]
+ assert len(tasks) == 1
+ assert tasks[0] == task
+
+ def test_remove_task(self, graph):
+ task = MockTask()
+ other_task = MockTask()
+ graph.add_tasks(task)
+ graph.add_tasks(other_task)
+ graph.remove_tasks(other_task)
+ tasks = [t for t in graph.tasks]
+ assert len(tasks) == 1
+ assert tasks[0] == task
+
+ def test_remove_tasks_with_dependency(self, graph):
+ task = MockTask()
+ dependent_task = MockTask()
+ graph.add_tasks(task)
+ graph.add_tasks(dependent_task)
+ graph.add_dependency(dependent_task, task)
+ remove_result = graph.remove_tasks(dependent_task)
+ assert remove_result == [dependent_task]
+ tasks = [t for t in graph.tasks]
+ assert len(tasks) == 1
+ assert tasks[0] == task
+ # asserting no dependencies are left for the dependent task
+ assert len(list(graph.get_dependencies(task))) == 0
+
+ def test_remove_nonexistent_task(self, graph):
+ task = MockTask()
+ task_not_in_graph = MockTask()
+ graph.add_tasks(task)
+ # removing a task not in graph - should have no effect, and return False
+ remove_result = graph.remove_tasks(task_not_in_graph)
+ assert remove_result == []
+ tasks = [t for t in graph.tasks]
+ assert len(tasks) == 1
+ assert tasks[0] == task
+
+ def test_has_task(self, graph):
+ task = MockTask()
+ graph.add_tasks(task)
+ assert graph.has_tasks(task) is True
+
+ def test_has_nonexistent_task(self, graph):
+ task = MockTask()
+ task_not_in_graph = MockTask()
+ graph.add_tasks(task)
+ assert graph.has_tasks(task_not_in_graph) is False
+
+ def test_get_task(self, graph):
+ task = MockTask()
+ graph.add_tasks(task)
+ assert graph.get_task(task.id) == task
+
+ def test_get_nonexistent_task(self, graph):
+ task = MockTask()
+ task_not_in_graph = MockTask()
+ graph.add_tasks(task)
+ with pytest.raises(task_graph.TaskNotInGraphError):
+ graph.get_task(task_not_in_graph.id)
+
+
+class TestTaskGraphGraphTraversal(object):
+
+ def test_tasks_iteration(self, graph):
+ task = MockTask()
+ other_task = MockTask()
+ graph.add_tasks(task)
+ graph.add_tasks(other_task)
+ tasks = [t for t in graph.tasks]
+ assert set(tasks) == set([task, other_task])
+
+ def test_get_dependents(self, graph):
+ task = MockTask()
+ dependent_task_1 = MockTask()
+ dependent_task_2 = MockTask()
+ transitively_dependent_task = MockTask()
+
+ graph.add_tasks(task)
+ graph.add_tasks(dependent_task_1)
+ graph.add_tasks(dependent_task_2)
+ graph.add_tasks(transitively_dependent_task)
+
+ graph.add_dependency(dependent_task_1, task)
+ graph.add_dependency(dependent_task_2, task)
+ graph.add_dependency(transitively_dependent_task, dependent_task_2)
+
+ dependent_tasks = list(graph.get_dependents(task))
+ # transitively_dependent_task not expected to appear in the result
+ assert set(dependent_tasks) == set([dependent_task_1, dependent_task_2])
+
+ def test_get_task_empty_dependents(self, graph):
+ task = MockTask()
+ other_task = MockTask()
+ graph.add_tasks(task)
+ graph.add_tasks(other_task)
+ dependent_tasks = list(graph.get_dependents(task))
+ assert len(dependent_tasks) == 0
+
+ def test_get_nonexistent_task_dependents(self, graph):
+ task = MockTask()
+ task_not_in_graph = MockTask()
+ graph.add_tasks(task)
+ with pytest.raises(task_graph.TaskNotInGraphError):
+ list(graph.get_dependents(task_not_in_graph))
+
+ def test_get_dependencies(self, graph):
+ task = MockTask()
+ dependency_task_1 = MockTask()
+ dependency_task_2 = MockTask()
+ transitively_dependency_task = MockTask()
+
+ graph.add_tasks(task)
+ graph.add_tasks(dependency_task_1)
+ graph.add_tasks(dependency_task_2)
+ graph.add_tasks(transitively_dependency_task)
+
+ graph.add_dependency(task, dependency_task_1)
+ graph.add_dependency(task, dependency_task_2)
+ graph.add_dependency(dependency_task_2, transitively_dependency_task)
+
+ dependency_tasks = list(graph.get_dependencies(task))
+ # transitively_dependency_task not expected to appear in the result
+ assert set(dependency_tasks) == set([dependency_task_1, dependency_task_2])
+
+ def test_get_task_empty_dependencies(self, graph):
+ task = MockTask()
+ other_task = MockTask()
+ graph.add_tasks(task)
+ graph.add_tasks(other_task)
+ dependency_tasks = list(graph.get_dependencies(task))
+ assert len(dependency_tasks) == 0
+
+ def test_get_nonexistent_task_dependencies(self, graph):
+ task = MockTask()
+ task_not_in_graph = MockTask()
+ graph.add_tasks(task)
+ with pytest.raises(task_graph.TaskNotInGraphError):
+ list(graph.get_dependencies(task_not_in_graph))
+
+
+class TestTaskGraphDependencies(object):
+
+ def test_add_dependency(self, graph):
+ task = MockTask()
+ dependency_task = MockTask()
+ unrelated_task = MockTask()
+ graph.add_tasks(task)
+ graph.add_tasks(dependency_task)
+ graph.add_tasks(unrelated_task)
+ add_result = graph.add_dependency(task, dependency_task)
+ assert add_result is True
+ dependency_tasks = list(graph.get_dependencies(task))
+ assert len(dependency_tasks) == 1
+ assert dependency_tasks[0] == dependency_task
+
+ def test_add_existing_dependency(self, graph):
+ task = MockTask()
+ dependency_task = MockTask()
+ graph.add_tasks(task)
+ graph.add_tasks(dependency_task)
+ graph.add_dependency(task, dependency_task)
+ # adding a dependency already in graph - should have no effect, and return False
+ add_result = graph.add_dependency(task, dependency_task)
+ assert add_result is False
+ dependency_tasks = list(graph.get_dependencies(task))
+ assert len(dependency_tasks) == 1
+ assert dependency_tasks[0] == dependency_task
+
+ def test_add_dependency_nonexistent_dependent(self, graph):
+ task = MockTask()
+ task_not_in_graph = MockTask()
+ graph.add_tasks(task)
+ with pytest.raises(task_graph.TaskNotInGraphError):
+ graph.add_dependency(task_not_in_graph, task)
+
+ def test_add_dependency_nonexistent_dependency(self, graph):
+ task = MockTask()
+ task_not_in_graph = MockTask()
+ graph.add_tasks(task)
+ with pytest.raises(task_graph.TaskNotInGraphError):
+ graph.add_dependency(task, task_not_in_graph)
+
+ def test_has_dependency(self, graph):
+ task = MockTask()
+ dependency_task = MockTask()
+ graph.add_tasks(task)
+ graph.add_tasks(dependency_task)
+ graph.add_dependency(task, dependency_task)
+ assert graph.has_dependency(task, dependency_task) is True
+
+ def test_has_nonexistent_dependency(self, graph):
+ task = MockTask()
+ other_task = MockTask()
+ graph.add_tasks(task)
+ graph.add_tasks(other_task)
+ assert graph.has_dependency(task, other_task) is False
+
+ def test_has_dependency_nonexistent_dependent(self, graph):
+ task = MockTask()
+ task_not_in_graph = MockTask()
+ graph.add_tasks(task)
+ with pytest.raises(task_graph.TaskNotInGraphError):
+ graph.has_dependency(task_not_in_graph, task)
+
+ def test_has_dependency_nonexistent_dependency(self, graph):
+ task = MockTask()
+ task_not_in_graph = MockTask()
+ graph.add_tasks(task)
+ with pytest.raises(task_graph.TaskNotInGraphError):
+ graph.has_dependency(task, task_not_in_graph)
+
+ def test_remove_dependency(self, graph):
+ task = MockTask()
+ dependency_task = MockTask()
+ another_dependent_task = MockTask()
+ graph.add_tasks(task)
+ graph.add_tasks(dependency_task)
+ graph.add_tasks(another_dependent_task)
+ graph.add_dependency(task, dependency_task)
+ graph.add_dependency(another_dependent_task, dependency_task)
+
+ remove_result = graph.remove_dependency(task, dependency_task)
+ assert remove_result is True
+ assert graph.has_dependency(task, dependency_task) is False
+ assert graph.has_dependency(another_dependent_task, dependency_task) is True
+
+ def test_remove_nonexistent_dependency(self, graph):
+ task = MockTask()
+ dependency_task = MockTask()
+ graph.add_tasks(task)
+ graph.add_tasks(dependency_task)
+ # removing a dependency not in graph - should have no effect, and return False
+ remove_result = graph.remove_dependency(task, dependency_task)
+ assert remove_result is False
+ tasks = [t for t in graph.tasks]
+ assert set(tasks) == set([task, dependency_task])
+
+ def test_remove_dependency_nonexistent_dependent(self, graph):
+ task = MockTask()
+ task_not_in_graph = MockTask()
+ graph.add_tasks(task)
+ with pytest.raises(task_graph.TaskNotInGraphError):
+ graph.remove_dependency(task_not_in_graph, task)
+
+ def test_remove_dependency_nonexistent_dependency(self, graph):
+ # in this test the dependency *task* is not in the graph, not just the dependency itself
+ task = MockTask()
+ task_not_in_graph = MockTask()
+ graph.add_tasks(task)
+ with pytest.raises(task_graph.TaskNotInGraphError):
+ graph.remove_dependency(task, task_not_in_graph)
+
+
+class TestTaskGraphArrangements(object):
+
+ def test_sequence(self, graph):
+ tasks = [MockTask(), MockTask(), MockTask()]
+ graph.sequence(*tasks)
+ graph_tasks = [t for t in graph.tasks]
+ assert set(graph_tasks) == set(tasks)
+ assert len(list(graph.get_dependencies(tasks[0]))) == 0
+ assert set(graph.get_dependencies(tasks[1])) == set([tasks[0]])
+ assert set(graph.get_dependencies(tasks[2])) == set([tasks[1]])
+
+ def test_sequence_with_some_tasks_and_dependencies_already_in_graph(self, graph):
+ tasks = [MockTask(), MockTask(), MockTask()]
+ # insert some tasks and dependencies to the graph
+ graph.add_tasks(tasks[1])
+ graph.add_tasks(tasks[2])
+ graph.add_dependency(tasks[2], tasks[1])
+
+ graph.sequence(*tasks)
+ graph_tasks = [t for t in graph.tasks]
+ assert set(graph_tasks) == set(tasks)
+ assert len(list(graph.get_dependencies(tasks[0]))) == 0
+ assert set(graph.get_dependencies(tasks[1])) == set([tasks[0]])
+ assert set(graph.get_dependencies(tasks[2])) == set([tasks[1]])
+
+ def test_sequence_with_nested_sequence(self, graph):
+ tasks = [MockTask() for _ in xrange(5)]
+ graph.sequence(tasks[0], graph.sequence(tasks[1], tasks[2], tasks[3]), tasks[4])
+ graph_tasks = [t for t in graph.tasks]
+ assert set(graph_tasks) == set(tasks)
+ # first task should have no dependencies
+ assert len(list(graph.get_dependencies(tasks[0]))) == 0
+ assert len(list(graph.get_dependencies(tasks[1]))) == 1
+ assert len(list(graph.get_dependencies(tasks[2]))) == 2
+ assert len(list(graph.get_dependencies(tasks[3]))) == 2
+ assert len(list(graph.get_dependencies(tasks[4]))) == 3
+
+ def test_sequence_with_nested_parallel(self, graph):
+ tasks = [MockTask() for _ in xrange(5)]
+ graph.sequence(tasks[0], (tasks[1], tasks[2], tasks[3]), tasks[4])
+ graph_tasks = [t for t in graph.tasks]
+ assert set(graph_tasks) == set(tasks)
+ # first task should have no dependencies
+ assert len(list(graph.get_dependencies(tasks[0]))) == 0
+ # rest of the tasks (except last) should have a single dependency - the first task
+ for i in xrange(1, 4):
+ assert set(graph.get_dependencies(tasks[i])) == set([tasks[0]])
+ # last task should have have a dependency on all tasks except for the first one
+ assert set(graph.get_dependencies(tasks[4])) == set([tasks[1], tasks[2], tasks[3]])
+
+ def test_parallel(self, graph):
+ tasks = [MockTask(), MockTask(), MockTask()]
+ graph.add_tasks(tasks)
+ graph_tasks = [t for t in graph.tasks]
+ assert set(graph_tasks) == set(tasks)
+ for i in xrange(len(tasks)):
+ assert len(list(graph.get_dependencies(tasks[i]))) == 0
+
+ def test_parallel_with_some_tasks_and_dependencies_already_in_graph(self, graph):
+ tasks = [MockTask(), MockTask(), MockTask()]
+
+ # insert some tasks and dependencies to the graph
+ graph.add_tasks(tasks[1])
+ graph.add_tasks(tasks[2])
+ graph.add_tasks(*tasks)
+
+ assert set(graph.tasks) == set(tasks)
+ for i in xrange(len(tasks)):
+ assert len(list(graph.get_dependencies(tasks[i]))) == 0
+
+ def test_parallel_with_nested_sequence(self, graph):
+ all_tasks = [MockTask() for _ in xrange(5)]
+ graph.add_tasks(all_tasks[0],
+ graph.sequence(all_tasks[1], all_tasks[2], all_tasks[3]),
+ all_tasks[4])
+ assert set(graph.tasks) == set(all_tasks)
+
+ # tasks[2] and tasks[3] should each have a single dependency; the rest should have none
+ assert len(list(graph.get_dependencies(all_tasks[0]))) == 0
+ assert len(list(graph.get_dependencies(all_tasks[1]))) == 0
+ assert set(graph.get_dependencies(all_tasks[2])) == set([all_tasks[1]])
+ assert set(graph.get_dependencies(all_tasks[3])) == set([all_tasks[2]])
+ assert len(list(graph.get_dependencies(all_tasks[4]))) == 0
+
+ def test_parallel_with_nested_parallel(self, graph):
+ tasks = [MockTask() for _ in xrange(5)]
+ graph.add_tasks(tasks[0], (tasks[1], tasks[2], tasks[3]), tasks[4])
+ graph_tasks = [t for t in graph.tasks]
+ assert set(graph_tasks) == set(tasks)
+ # none of the tasks should have any dependencies
+ for i in xrange(len(tasks)):
+ assert len(list(graph.get_dependencies(tasks[i]))) == 0
+
+ def test_group_with_twice_nested_group(self, graph):
+ twice_nested_tasks = [MockTask(), MockTask(), MockTask()]
+ once_nested_tasks = [MockTask(), MockTask(), MockTask(), twice_nested_tasks]
+ tasks = [MockTask(), MockTask(), MockTask(), once_nested_tasks]
+ graph.add_tasks(*tasks)
+
+ assert set(graph.tasks) == set(tasks[:3] + once_nested_tasks[:3] + twice_nested_tasks)
+ for tasks_list in [tasks, once_nested_tasks, twice_nested_tasks]:
+ for i in xrange(len(tasks_list[:3])):
+ assert len(list(graph.get_dependencies(tasks_list[i]))) == 0
+
+
+class TestTaskGraphDependenciesWithArrangements(object):
+
+ def test_add_dependency_dependent_sequence(self, graph):
+ task = MockTask()
+ sequence_tasks = [MockTask() for _ in xrange(3)]
+ graph.add_tasks(task)
+ sequence = graph.sequence(*sequence_tasks)
+ graph.add_dependency(sequence, task)
+ assert graph.has_dependency(sequence_tasks[0], task) is True
+ assert graph.has_dependency(sequence_tasks[1], task) is True
+ assert graph.has_dependency(sequence_tasks[2], task) is True
+
+ def test_add_dependency_dependency_sequence(self, graph):
+ task = MockTask()
+ sequence_tasks = [MockTask() for _ in xrange(3)]
+ graph.add_tasks(task)
+ sequence = graph.sequence(*sequence_tasks)
+ graph.add_dependency(task, sequence)
+ assert graph.has_dependency(task, sequence_tasks[0]) is True
+ assert graph.has_dependency(task, sequence_tasks[1]) is True
+ assert graph.has_dependency(task, sequence_tasks[2]) is True
+
+ def test_add_dependency_dependent_parallel(self, graph):
+ task = MockTask()
+ parallel_tasks = [MockTask() for _ in xrange(3)]
+ graph.add_tasks(task)
+ parallel = graph.add_tasks(*parallel_tasks)
+ graph.add_dependency(parallel, task)
+ assert graph.has_dependency(parallel_tasks[0], task) is True
+ assert graph.has_dependency(parallel_tasks[1], task) is True
+ assert graph.has_dependency(parallel_tasks[2], task) is True
+
+ def test_add_dependency_dependency_parallel(self, graph):
+ task = MockTask()
+ parallel_tasks = [MockTask() for _ in xrange(3)]
+ graph.add_tasks(task)
+ parallel = graph.add_tasks(*parallel_tasks)
+ graph.add_dependency(task, parallel)
+ assert graph.has_dependency(task, parallel_tasks[0]) is True
+ assert graph.has_dependency(task, parallel_tasks[1]) is True
+ assert graph.has_dependency(task, parallel_tasks[2]) is True
+
+ def test_add_dependency_between_sequence_and_parallel(self, graph):
+ sequence_tasks = [MockTask() for _ in xrange(3)]
+ parallel_tasks = [MockTask() for _ in xrange(3)]
+ sequence = graph.sequence(*sequence_tasks)
+ parallel = graph.add_tasks(*parallel_tasks)
+ graph.add_dependency(sequence, parallel)
+ for parallel_task in parallel_tasks:
+ assert graph.has_dependency(sequence_tasks[0], parallel_task) is True
+ assert graph.has_dependency(sequence_tasks[1], parallel_task) is True
+ assert graph.has_dependency(sequence_tasks[2], parallel_task) is True
+
+ def test_add_dependency_between_parallel_and_sequence(self, graph):
+ sequence_tasks = [MockTask() for _ in xrange(3)]
+ parallel_tasks = [MockTask() for _ in xrange(3)]
+ sequence = graph.sequence(*sequence_tasks)
+ parallel = graph.add_tasks(*parallel_tasks)
+ graph.add_dependency(parallel, sequence)
+ for parallel_task in parallel_tasks:
+ assert graph.has_dependency(parallel_task, sequence_tasks[0]) is True
+ assert graph.has_dependency(parallel_task, sequence_tasks[1]) is True
+ assert graph.has_dependency(parallel_task, sequence_tasks[2]) is True
+
+ def test_add_dependency_dependency_parallel_with_some_existing_dependencies(self, graph):
+ task = MockTask()
+ parallel_tasks = [MockTask() for _ in xrange(3)]
+ graph.add_tasks(task)
+ parallel = graph.add_tasks(*parallel_tasks)
+ # adding a dependency on a specific task manually,
+ # before adding a dependency on the whole parallel
+ graph.add_dependency(task, parallel_tasks[1])
+ graph.add_dependency(task, parallel)
+ assert graph.has_dependency(task, parallel_tasks[0]) is True
+ assert graph.has_dependency(task, parallel_tasks[1]) is True
+ assert graph.has_dependency(task, parallel_tasks[2]) is True
+
+ def test_add_existing_dependency_between_sequence_and_parallel(self, graph):
+ sequence_tasks = [MockTask() for _ in xrange(3)]
+ parallel_tasks = [MockTask() for _ in xrange(3)]
+ sequence = graph.sequence(*sequence_tasks)
+ parallel = graph.add_tasks(*parallel_tasks)
+ add_result = graph.add_dependency(sequence, parallel)
+ assert add_result is True
+ # adding a dependency already in graph - should have no effect, and return False
+ add_result = graph.add_dependency(sequence, parallel)
+ assert add_result is False
+ for parallel_task in parallel_tasks:
+ assert graph.has_dependency(sequence_tasks[0], parallel_task) is True
+ assert graph.has_dependency(sequence_tasks[1], parallel_task) is True
+ assert graph.has_dependency(sequence_tasks[2], parallel_task) is True
+
+ def test_has_dependency_dependent_sequence(self, graph):
+ task = MockTask()
+ sequence_tasks = [MockTask() for _ in xrange(3)]
+ graph.add_tasks(task)
+ sequence = graph.sequence(*sequence_tasks)
+ assert graph.has_dependency(sequence, task) is False
+ graph.add_dependency(sequence, task)
+ assert graph.has_dependency(sequence, task) is True
+
+ def test_has_dependency_dependency_sequence(self, graph):
+ task = MockTask()
+ sequence_tasks = [MockTask() for _ in xrange(3)]
+ graph.add_tasks(task)
+ sequence = graph.sequence(*sequence_tasks)
+ assert graph.has_dependency(task, sequence) is False
+ graph.add_dependency(task, sequence)
+ assert graph.has_dependency(task, sequence) is True
+
+ def test_has_dependency_dependent_parallel(self, graph):
+ task = MockTask()
+ parallel_tasks = [MockTask() for _ in xrange(3)]
+ graph.add_tasks(task)
+ parallel = graph.add_tasks(*parallel_tasks)
+ assert graph.has_dependency(parallel, task) is False
+ graph.add_dependency(parallel, task)
+ assert graph.has_dependency(parallel, task) is True
+
+ def test_has_dependency_dependency_parallel(self, graph):
+ task = MockTask()
+ parallel_tasks = [MockTask() for _ in xrange(3)]
+ graph.add_tasks(task)
+ parallel = graph.add_tasks(*parallel_tasks)
+ assert graph.has_dependency(task, parallel) is False
+ graph.add_dependency(task, parallel)
+ assert graph.has_dependency(task, parallel) is True
+
+ def test_has_dependency_between_sequence_and_parallel(self, graph):
+ sequence_tasks = [MockTask() for _ in xrange(3)]
+ parallel_tasks = [MockTask() for _ in xrange(3)]
+ sequence = graph.sequence(*sequence_tasks)
+ parallel = graph.add_tasks(*parallel_tasks)
+ assert graph.has_dependency(sequence, parallel) is False
+ graph.add_dependency(sequence, parallel)
+ assert graph.has_dependency(sequence, parallel) is True
+
+ def test_has_dependency_between_parallel_and_sequence(self, graph):
+ sequence_tasks = [MockTask() for _ in xrange(3)]
+ parallel_tasks = [MockTask() for _ in xrange(3)]
+ sequence = graph.sequence(*sequence_tasks)
+ parallel = graph.add_tasks(*parallel_tasks)
+ assert graph.has_dependency(parallel, sequence) is False
+ graph.add_dependency(parallel, sequence)
+ assert graph.has_dependency(parallel, sequence) is True
+
+ def test_has_dependency_dependency_parallel_with_some_existing_dependencies(self, graph):
+ task = MockTask()
+ parallel_tasks = [MockTask() for _ in xrange(3)]
+ graph.add_tasks(task)
+ parallel = graph.add_tasks(*parallel_tasks)
+ graph.add_dependency(task, parallel_tasks[1])
+ # only a partial dependency exists - has_dependency is expected to return False
+ assert graph.has_dependency(task, parallel) is False
+
+ def test_has_nonexistent_dependency_between_sequence_and_parallel(self, graph):
+ sequence_tasks = [MockTask() for _ in xrange(3)]
+ parallel_tasks = [MockTask() for _ in xrange(3)]
+ sequence = graph.sequence(*sequence_tasks)
+ parallel = graph.add_tasks(*parallel_tasks)
+ assert graph.has_dependency(sequence, parallel) is False
+
+ def test_remove_dependency_dependent_sequence(self, graph):
+ task = MockTask()
+ sequence_tasks = [MockTask() for _ in xrange(3)]
+ graph.add_tasks(task)
+ sequence = graph.sequence(*sequence_tasks)
+ graph.add_dependency(sequence, task)
+ remove_result = graph.remove_dependency(sequence, task)
+ assert remove_result is True
+ assert graph.has_dependency(sequence_tasks[0], task) is False
+ assert graph.has_dependency(sequence_tasks[1], task) is False
+ assert graph.has_dependency(sequence_tasks[2], task) is False
+
+ def test_remove_dependency_dependency_sequence(self, graph):
+ task = MockTask()
+ sequence_tasks = [MockTask() for _ in xrange(3)]
+ graph.add_tasks(task)
+ sequence = graph.sequence(*sequence_tasks)
+ graph.add_dependency(task, sequence)
+ remove_result = graph.remove_dependency(task, sequence)
+ assert remove_result is True
+ assert graph.has_dependency(task, sequence_tasks[0]) is False
+ assert graph.has_dependency(task, sequence_tasks[1]) is False
+ assert graph.has_dependency(task, sequence_tasks[2]) is False
+
+ def test_remove_dependency_dependent_parallel(self, graph):
+ task = MockTask()
+ parallel_tasks = [MockTask() for _ in xrange(3)]
+ graph.add_tasks(task)
+ parallel = graph.add_tasks(*parallel_tasks)
+ graph.add_dependency(parallel, task)
+ remove_result = graph.remove_dependency(parallel, task)
+ assert remove_result is True
+ assert graph.has_dependency(parallel_tasks[0], task) is False
+ assert graph.has_dependency(parallel_tasks[1], task) is False
+ assert graph.has_dependency(parallel_tasks[2], task) is False
+
+ def test_remove_dependency_dependency_parallel(self, graph):
+ task = MockTask()
+ parallel_tasks = [MockTask() for _ in xrange(3)]
+ graph.add_tasks(task)
+ parallel = graph.add_tasks(*parallel_tasks)
+ graph.add_dependency(task, parallel)
+ remove_result = graph.remove_dependency(task, parallel)
+ assert remove_result is True
+ assert graph.has_dependency(task, parallel_tasks[0]) is False
+ assert graph.has_dependency(task, parallel_tasks[1]) is False
+ assert graph.has_dependency(task, parallel_tasks[2]) is False
+
+ def test_remove_dependency_between_sequence_and_parallel(self, graph):
+ sequence_tasks = [MockTask() for _ in xrange(3)]
+ parallel_tasks = [MockTask() for _ in xrange(3)]
+ sequence = graph.sequence(*sequence_tasks)
+ parallel = graph.add_tasks(*parallel_tasks)
+ graph.add_dependency(sequence, parallel)
+ remove_result = graph.remove_dependency(sequence, parallel)
+ assert remove_result is True
+ for parallel_task in parallel_tasks:
+ assert graph.has_dependency(sequence_tasks[0], parallel_task) is False
+ assert graph.has_dependency(sequence_tasks[1], parallel_task) is False
+ assert graph.has_dependency(sequence_tasks[2], parallel_task) is False
+
+ def test_remove_dependency_between_parallel_and_sequence(self, graph):
+ sequence_tasks = [MockTask() for _ in xrange(3)]
+ parallel_tasks = [MockTask() for _ in xrange(3)]
+ sequence = graph.sequence(*sequence_tasks)
+ parallel = graph.add_tasks(*parallel_tasks)
+ graph.add_dependency(parallel, sequence)
+ remove_result = graph.remove_dependency(parallel, sequence)
+ assert remove_result is True
+ for parallel_task in parallel_tasks:
+ assert graph.has_dependency(parallel_task, sequence_tasks[0]) is False
+ assert graph.has_dependency(parallel_task, sequence_tasks[1]) is False
+ assert graph.has_dependency(parallel_task, sequence_tasks[2]) is False
+
+ def test_remove_dependency_dependency_parallel_with_some_existing_dependencies(self, graph):
+ task = MockTask()
+ parallel_tasks = [MockTask() for _ in xrange(3)]
+ graph.add_tasks(task)
+ parallel = graph.add_tasks(*parallel_tasks)
+ graph.add_dependency(task, parallel_tasks[1])
+ remove_result = graph.remove_dependency(task, parallel)
+ # only a partial dependency exists - remove_dependency is expected to return False
+ assert remove_result is False
+ # no dependencies are expected to have changed
+ assert graph.has_dependency(task, parallel_tasks[0]) is False
+ assert graph.has_dependency(task, parallel_tasks[1]) is True
+ assert graph.has_dependency(task, parallel_tasks[2]) is False
+
+ def test_remove_nonexistent_dependency_between_sequence_and_parallel(self, graph):
+ sequence_tasks = [MockTask() for _ in xrange(3)]
+ parallel_tasks = [MockTask() for _ in xrange(3)]
+ sequence = graph.sequence(*sequence_tasks)
+ parallel = graph.add_tasks(*parallel_tasks)
+ # removing a dependency not in graph - should have no effect, and return False
+ remove_result = graph.remove_dependency(sequence, parallel)
+ assert remove_result is False
+
+ def test_sequence_with_twice_nested_sequence_and_interdependencies(self, graph):
+ twice_nested_tasks = list(graph.sequence(MockTask(), MockTask(), MockTask()))
+ once_nested_tasks = list(graph.sequence(MockTask(),
+ MockTask(),
+ MockTask(),
+ twice_nested_tasks))
+ tasks = [MockTask(), MockTask(), MockTask(), once_nested_tasks]
+ graph.sequence(*tasks)
+
+ assert set(graph.tasks) == set(tasks[:3] + once_nested_tasks[:3] + twice_nested_tasks)
+ assert set(graph.get_dependencies(tasks[0])) == set()
+ for i in xrange(1, len(tasks[:-1])):
+ assert set(graph.get_dependencies(tasks[i])) == set([tasks[i - 1]])
+
+ assert set(graph.get_dependencies(once_nested_tasks[0])) == set([tasks[2]])
+ for i in xrange(1, len(once_nested_tasks[:-1])):
+ assert set(graph.get_dependencies(once_nested_tasks[i])) == \
+ set([tasks[2], once_nested_tasks[i-1]])
+
+ assert set(graph.get_dependencies(twice_nested_tasks[0])) == \
+ set([tasks[2], once_nested_tasks[2]])
+ for i in xrange(1, len(twice_nested_tasks[:-1])):
+ assert set(graph.get_dependencies(twice_nested_tasks[i])) == \
+ set([tasks[2], once_nested_tasks[2], twice_nested_tasks[i-1]])
+
+ def test_group_with_twice_nested_group_and_interdependencies(self, graph):
+ twice_nested_tasks = [MockTask(), MockTask(), MockTask()]
+ once_nested_tasks = [MockTask(), MockTask(), MockTask(), twice_nested_tasks]
+ tasks = [MockTask(), MockTask(), MockTask(), once_nested_tasks]
+ graph.add_tasks(*tasks)
+
+ graph.add_dependency(tasks[2], once_nested_tasks[2])
+ graph.add_dependency(once_nested_tasks[1], twice_nested_tasks[0])
+ graph.add_dependency(twice_nested_tasks[1], tasks[0])
+
+ assert set(graph.tasks) == set(tasks[:3] + once_nested_tasks[:3] + twice_nested_tasks)
+ assert set(graph.get_dependencies(tasks[0])) == set()
+ assert set(graph.get_dependencies(tasks[1])) == set()
+ assert set(graph.get_dependencies(tasks[2])) == set([once_nested_tasks[2]])
+
+ assert set((graph.get_dependencies(once_nested_tasks[0]))) == set()
+ assert set(graph.get_dependencies(once_nested_tasks[1])) == set([twice_nested_tasks[0]])
+ assert set(graph.get_dependencies(once_nested_tasks[2])) == set()
+
+ assert set((graph.get_dependencies(twice_nested_tasks[0]))) == set()
+ assert set(graph.get_dependencies(twice_nested_tasks[1])) == set([tasks[0]])
+ assert set(graph.get_dependencies(twice_nested_tasks[2])) == set()
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/tests/workflows/builtin/__init__.py
----------------------------------------------------------------------
diff --git a/tests/workflows/builtin/__init__.py b/tests/workflows/builtin/__init__.py
new file mode 100644
index 0000000..6d58900
--- /dev/null
+++ b/tests/workflows/builtin/__init__.py
@@ -0,0 +1,53 @@
+# 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 ... import mock
+
+def assert_node_install_operations(operations, with_relationships=False):
+ if with_relationships:
+ all_operations = [
+ 'aria.interfaces.lifecycle.create',
+ 'aria.interfaces.relationship_lifecycle.preconfigure',
+ 'aria.interfaces.relationship_lifecycle.preconfigure',
+ 'aria.interfaces.lifecycle.configure',
+ 'aria.interfaces.relationship_lifecycle.postconfigure',
+ 'aria.interfaces.relationship_lifecycle.postconfigure',
+ 'aria.interfaces.lifecycle.start',
+ 'aria.interfaces.relationship_lifecycle.establish',
+ 'aria.interfaces.relationship_lifecycle.establish',
+ ]
+
+ for i, operation in enumerate(operations):
+ assert operation.name.startswith(all_operations[i])
+ else:
+ for i, operation in enumerate(operations):
+ assert operation.name.startswith(mock.operations.NODE_OPERATIONS_INSTALL[i])
+
+
+def assert_node_uninstall_operations(operations, with_relationships=False):
+ if with_relationships:
+ all_operations = [
+ 'aria.interfaces.lifecycle.stop',
+ 'aria.interfaces.relationship_lifecycle.unlink',
+ 'aria.interfaces.relationship_lifecycle.unlink',
+ 'aria.interfaces.lifecycle.delete',
+ ]
+
+ for i, operation in enumerate(operations):
+ assert operation.name.startswith(all_operations[i])
+ else:
+ for i, operation in enumerate(operations):
+ assert operation.name.startswith(mock.operations.NODE_OPERATIONS_UNINSTALL[i])
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/tests/workflows/builtin/test_execute_operation.py
----------------------------------------------------------------------
diff --git a/tests/workflows/builtin/test_execute_operation.py b/tests/workflows/builtin/test_execute_operation.py
new file mode 100644
index 0000000..82010c8
--- /dev/null
+++ b/tests/workflows/builtin/test_execute_operation.py
@@ -0,0 +1,74 @@
+# 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.
+
+import pytest
+
+from aria.workflows.api import task
+from aria.workflows.builtin.execute_operation import execute_operation
+
+from ... import mock
+
+
+@pytest.fixture()
+def ctx():
+ simple_context = mock.context.simple()
+ dependency_node = mock.models.get_dependency_node()
+ dependency_node_instance = mock.models.get_dependency_node_instance(
+ dependency_node=dependency_node)
+
+ relationship = mock.models.get_relationship(dependency_node)
+ relationship_instance = mock.models.get_relationship_instance(
+ relationship=relationship,
+ target_instance=dependency_node_instance
+ )
+
+ dependent_node = mock.models.get_dependent_node(relationship)
+ dependent_node_instance = mock.models.get_dependent_node_instance(
+ dependent_node=dependent_node,
+ relationship_instance=relationship_instance
+ )
+
+ simple_context.model.node.store(dependent_node)
+ simple_context.model.node.store(dependency_node)
+ simple_context.model.node_instance.store(dependent_node_instance)
+ simple_context.model.node_instance.store(dependency_node_instance)
+ simple_context.model.relationship.store(relationship)
+ simple_context.model.relationship_instance.store(relationship_instance)
+
+ return simple_context
+
+
+def test_execute_operation(ctx):
+ operation_name = mock.operations.NODE_OPERATIONS_INSTALL[0]
+ node_instance_id = 'dependency_node_instance'
+
+ execute_tasks = list(
+ task.WorkflowTask(
+ execute_operation,
+ ctx=ctx,
+ operation=operation_name,
+ operation_kwargs={},
+ allow_kwargs_override=False,
+ run_by_dependency_order=False,
+ type_names=[],
+ node_ids=[],
+ node_instance_ids=[node_instance_id]
+ ).topological_order()
+ )
+
+ assert len(execute_tasks) == 1
+ assert execute_tasks[0].name == '{0}.{1}'.format(node_instance_id, operation_name)
+
+# TODO: add more scenarios
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/tests/workflows/builtin/test_heal.py
----------------------------------------------------------------------
diff --git a/tests/workflows/builtin/test_heal.py b/tests/workflows/builtin/test_heal.py
new file mode 100644
index 0000000..4b9ae09
--- /dev/null
+++ b/tests/workflows/builtin/test_heal.py
@@ -0,0 +1,117 @@
+# 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.
+
+import pytest
+
+from aria.workflows.api import task
+from aria.workflows.builtin.heal import heal
+
+from . import (assert_node_install_operations,
+ assert_node_uninstall_operations)
+from ... import mock
+
+
+@pytest.fixture()
+def ctx():
+ simple_context = mock.context.simple()
+ dependency_node = mock.models.get_dependency_node()
+ dependency_node_instance = mock.models.get_dependency_node_instance(
+ dependency_node=dependency_node)
+
+ relationship = mock.models.get_relationship(dependency_node)
+ relationship_instance = mock.models.get_relationship_instance(
+ relationship=relationship,
+ target_instance=dependency_node_instance
+ )
+
+ dependent_node = mock.models.get_dependent_node(relationship)
+ dependent_node_instance = mock.models.get_dependent_node_instance(
+ dependent_node=dependent_node,
+ relationship_instance=relationship_instance
+ )
+
+ simple_context.model.node.store(dependent_node)
+ simple_context.model.node.store(dependency_node)
+ simple_context.model.node_instance.store(dependent_node_instance)
+ simple_context.model.node_instance.store(dependency_node_instance)
+ simple_context.model.relationship.store(relationship)
+ simple_context.model.relationship_instance.store(relationship_instance)
+
+ return simple_context
+
+
+def test_heal_dependent_node(ctx):
+ heal_graph = task.WorkflowTask(heal, ctx=ctx, node_instance_id='dependent_node_instance')
+
+ assert len(list(heal_graph.tasks)) == 2
+ uninstall_subgraph, install_subgraph = list(heal_graph.topological_order(reverse=True))
+
+ assert len(list(uninstall_subgraph.tasks)) == 2
+ dependent_node_subgraph_uninstall, dependency_node_subgraph_uninstall = \
+ list(uninstall_subgraph.topological_order(reverse=True))
+
+ assert len(list(install_subgraph.tasks)) == 2
+ dependency_node_subgraph_install, dependent_node_subgraph_install = \
+ list(install_subgraph.topological_order(reverse=True))
+
+ dependent_node_uninstall_tasks = \
+ list(dependent_node_subgraph_uninstall.topological_order(reverse=True))
+ dependency_node_uninstall_tasks = \
+ list(dependency_node_subgraph_uninstall.topological_order(reverse=True))
+ dependent_node_install_tasks = \
+ list(dependent_node_subgraph_install.topological_order(reverse=True))
+ dependency_node_install_tasks = \
+ list(dependency_node_subgraph_install.topological_order(reverse=True))
+
+ assert_node_uninstall_operations(dependent_node_uninstall_tasks, with_relationships=True)
+ assert len(dependency_node_uninstall_tasks) == 0
+
+ assert len(dependency_node_install_tasks) == 0
+ assert_node_install_operations(dependent_node_install_tasks, with_relationships=True)
+
+
+def test_heal_dependency_node(ctx):
+ heal_graph = task.WorkflowTask(heal, ctx=ctx, node_instance_id='dependency_node_instance')
+ # both subgraphs should contain un\install for both the dependent and the dependency
+ assert len(list(heal_graph.tasks)) == 2
+ uninstall_subgraph, install_subgraph = list(heal_graph.topological_order(reverse=True))
+
+ uninstall_tasks = list(uninstall_subgraph.topological_order(reverse=True))
+ assert len(uninstall_tasks) == 4
+ unlink_source, unlink_target = uninstall_tasks[:2]
+ dependent_node_subgraph_uninstall, dependency_node_subgraph_uninstall = uninstall_tasks[2:]
+
+ install_tasks = list(install_subgraph.topological_order(reverse=True))
+ assert len(install_tasks) == 4
+ dependency_node_subgraph_install, dependent_node_subgraph_install = install_tasks[:2]
+ establish_source, establish_target = install_tasks[2:]
+
+ assert len(list(dependent_node_subgraph_uninstall.tasks)) == 0
+ dependency_node_uninstall_tasks = \
+ list(dependency_node_subgraph_uninstall.topological_order(reverse=True))
+ assert len(list(dependent_node_subgraph_install.tasks)) == 0
+ dependency_node_install_tasks = \
+ list(dependency_node_subgraph_install.topological_order(reverse=True))
+
+ assert unlink_source.name.startswith('aria.interfaces.relationship_lifecycle.unlink')
+ assert unlink_target.name.startswith('aria.interfaces.relationship_lifecycle.unlink')
+ assert_node_uninstall_operations(dependency_node_uninstall_tasks)
+
+ assert_node_install_operations(dependency_node_install_tasks)
+ assert establish_source.name.startswith('aria.interfaces.relationship_lifecycle.establish')
+ assert establish_target.name.startswith('aria.interfaces.relationship_lifecycle.establish')
+
+
+# TODO: add tests for contained in scenario
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/tests/workflows/builtin/test_install.py
----------------------------------------------------------------------
diff --git a/tests/workflows/builtin/test_install.py b/tests/workflows/builtin/test_install.py
new file mode 100644
index 0000000..d27ece4
--- /dev/null
+++ b/tests/workflows/builtin/test_install.py
@@ -0,0 +1,63 @@
+# 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.
+
+import pytest
+
+from aria.workflows.builtin.install import install
+from aria.workflows.api import task
+
+from . import assert_node_install_operations
+from ... import mock
+
+
+@pytest.fixture()
+def ctx():
+ simple_context = mock.context.simple()
+ dependency_node = mock.models.get_dependency_node()
+ dependency_node_instance = mock.models.get_dependency_node_instance(
+ dependency_node=dependency_node)
+
+ relationship = mock.models.get_relationship(dependency_node)
+ relationship_instance = mock.models.get_relationship_instance(
+ relationship=relationship,
+ target_instance=dependency_node_instance
+ )
+
+ dependent_node = mock.models.get_dependent_node(relationship)
+ dependent_node_instance = mock.models.get_dependent_node_instance(
+ dependent_node=dependent_node,
+ relationship_instance=relationship_instance
+ )
+
+ simple_context.model.node.store(dependent_node)
+ simple_context.model.node.store(dependency_node)
+ simple_context.model.node_instance.store(dependent_node_instance)
+ simple_context.model.node_instance.store(dependency_node_instance)
+ simple_context.model.relationship.store(relationship)
+ simple_context.model.relationship_instance.store(relationship_instance)
+
+ return simple_context
+
+
+def test_install(ctx):
+ install_tasks = list(task.WorkflowTask(install, ctx=ctx).topological_order(True))
+
+ assert len(install_tasks) == 2
+ dependency_node_subgraph, dependent_node_subgraph = install_tasks
+ dependent_node_tasks = list(dependent_node_subgraph.topological_order(reverse=True))
+ dependency_node_tasks = list(dependency_node_subgraph.topological_order(reverse=True))
+
+ assert_node_install_operations(dependency_node_tasks)
+ assert_node_install_operations(dependent_node_tasks, with_relationships=True)
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/tests/workflows/builtin/test_uninstall.py
----------------------------------------------------------------------
diff --git a/tests/workflows/builtin/test_uninstall.py b/tests/workflows/builtin/test_uninstall.py
new file mode 100644
index 0000000..cee7998
--- /dev/null
+++ b/tests/workflows/builtin/test_uninstall.py
@@ -0,0 +1,63 @@
+# 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.
+
+import pytest
+
+from aria.workflows.api import task
+from aria.workflows.builtin.uninstall import uninstall
+
+from . import assert_node_uninstall_operations
+from ... import mock
+
+
+@pytest.fixture()
+def ctx():
+ simple_context = mock.context.simple()
+ dependency_node = mock.models.get_dependency_node()
+ dependency_node_instance = mock.models.get_dependency_node_instance(
+ dependency_node=dependency_node)
+
+ relationship = mock.models.get_relationship(dependency_node)
+ relationship_instance = mock.models.get_relationship_instance(
+ relationship=relationship,
+ target_instance=dependency_node_instance
+ )
+
+ dependent_node = mock.models.get_dependent_node(relationship)
+ dependent_node_instance = mock.models.get_dependent_node_instance(
+ dependent_node=dependent_node,
+ relationship_instance=relationship_instance
+ )
+
+ simple_context.model.node.store(dependent_node)
+ simple_context.model.node.store(dependency_node)
+ simple_context.model.node_instance.store(dependent_node_instance)
+ simple_context.model.node_instance.store(dependency_node_instance)
+ simple_context.model.relationship.store(relationship)
+ simple_context.model.relationship_instance.store(relationship_instance)
+
+ return simple_context
+
+
+def test_uninstall(ctx):
+ uninstall_tasks = list(task.WorkflowTask(uninstall, ctx=ctx).topological_order(True))
+
+ assert len(uninstall_tasks) == 2
+ dependent_node_subgraph, dependency_node_subgraph = uninstall_tasks
+ dependent_node_tasks = list(dependent_node_subgraph.topological_order(reverse=True))
+ dependency_node_tasks = list(dependency_node_subgraph.topological_order(reverse=True))
+
+ assert_node_uninstall_operations(operations=dependency_node_tasks)
+ assert_node_uninstall_operations(operations=dependent_node_tasks, with_relationships=True)
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/tests/workflows/core/__init__.py
----------------------------------------------------------------------
diff --git a/tests/workflows/core/__init__.py b/tests/workflows/core/__init__.py
new file mode 100644
index 0000000..e69de29
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/017779dd/tests/workflows/core/test_executor.py
----------------------------------------------------------------------
diff --git a/tests/workflows/core/test_executor.py b/tests/workflows/core/test_executor.py
new file mode 100644
index 0000000..b31f49e
--- /dev/null
+++ b/tests/workflows/core/test_executor.py
@@ -0,0 +1,122 @@
+# 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.
+
+import logging
+import uuid
+
+import pytest
+import retrying
+
+from aria import events
+from aria.storage import models
+from aria.workflows.core import executor
+
+from ... import mock
+
+
+class TestExecutor(object):
+
+ @pytest.mark.parametrize('pool_size,executor_cls', [
+ (1, executor.ThreadExecutor),
+ (2, executor.ThreadExecutor),
+ (1, executor.MultiprocessExecutor),
+ (2, executor.MultiprocessExecutor),
+ (0, executor.CurrentThreadBlockingExecutor)
+ ])
+ def test_execute(self, pool_size, executor_cls):
+ self.executor = executor_cls(pool_size)
+ expected_value = 'value'
+ successful_task = MockTask(mock_successful_task)
+ failing_task = MockTask(mock_failing_task)
+ task_with_inputs = MockTask(mock_task_with_input, inputs={'input': expected_value})
+
+ for task in [successful_task, failing_task, task_with_inputs]:
+ self.executor.execute(task)
+
+ @retrying.retry(stop_max_delay=10000, wait_fixed=100)
+ def assertion():
+ assert successful_task.states == ['start', 'success']
+ assert failing_task.states == ['start', 'failure']
+ assert task_with_inputs.states == ['start', 'failure']
+ assert isinstance(failing_task.exception, TestException)
+ assert isinstance(task_with_inputs.exception, TestException)
+ assert str(task_with_inputs.exception) == expected_value
+ assertion()
+
+ def setup_method(self):
+ events.start_task_signal.connect(start_handler)
+ events.on_success_task_signal.connect(success_handler)
+ events.on_failure_task_signal.connect(failure_handler)
+
+ def teardown_method(self):
+ events.start_task_signal.disconnect(start_handler)
+ events.on_success_task_signal.disconnect(success_handler)
+ events.on_failure_task_signal.disconnect(failure_handler)
+ if hasattr(self, 'executor'):
+ self.executor.close()
+
+
+def mock_successful_task():
+ pass
+
+
+def mock_failing_task():
+ raise TestException
+
+
+def mock_task_with_input(input):
+ raise TestException(input)
+
+
+class TestException(Exception):
+ pass
+
+
+class MockContext(object):
+
+ def __init__(self, operation_details, inputs):
+ self.operation_details = operation_details
+ self.inputs = inputs
+ self.operation = models.Task(execution_id='',
+ name='mock_context',
+ operation_details={},
+ node_instance=mock.models.get_dependency_node_instance())
+
+
+class MockTask(object):
+
+ def __init__(self, func, inputs=None):
+ self.states = []
+ self.exception = None
+ self.id = str(uuid.uuid4())
+ name = func.__name__
+ operation = 'tests.workflows.core.test_executor.{name}'.format(name=name)
+ self.context = MockContext(operation_details={'operation': operation},
+ inputs=inputs or {})
+ self.logger = logging.getLogger()
+ self.name = name
+
+
+def start_handler(task, *args, **kwargs):
+ task.states.append('start')
+
+
+def success_handler(task, *args, **kwargs):
+ task.states.append('success')
+
+
+def failure_handler(task, exception, *args, **kwargs):
+ task.states.append('failure')
+ task.exception = exception