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/10/30 15:50:50 UTC
incubator-ariatosca git commit: reworked the task_graph api and added
some nested constructs tests
Repository: incubator-ariatosca
Updated Branches:
refs/heads/ARIA-3-api-for-creating-workflows 936e2f97e -> 94bbcf6f8
reworked the task_graph api and added some nested constructs tests
Project: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/commit/94bbcf6f
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/tree/94bbcf6f
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/diff/94bbcf6f
Branch: refs/heads/ARIA-3-api-for-creating-workflows
Commit: 94bbcf6f8c0e347160c447a029fbc18826861b1f
Parents: 936e2f9
Author: mxmrlv <mx...@gmail.com>
Authored: Sun Oct 30 17:50:37 2016 +0200
Committer: mxmrlv <mx...@gmail.com>
Committed: Sun Oct 30 17:50:37 2016 +0200
----------------------------------------------------------------------
aria/workflows/api/task_graph.py | 140 ++++++++++++++++------------
tests/workflows/api/test_task_graph.py | 117 ++++++++++++++++-------
2 files changed, 164 insertions(+), 93 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/94bbcf6f/aria/workflows/api/task_graph.py
----------------------------------------------------------------------
diff --git a/aria/workflows/api/task_graph.py b/aria/workflows/api/task_graph.py
index 2d8ee54..626ed72 100644
--- a/aria/workflows/api/task_graph.py
+++ b/aria/workflows/api/task_graph.py
@@ -18,6 +18,7 @@ Task graph. Used by users to build workflows
"""
from uuid import uuid4
+from collections import Iterable
from networkx import DiGraph, topological_sort
@@ -144,6 +145,41 @@ class TaskGraph(object):
"""
return self._graph.has_node(task.id)
+ def add_tasks(self, *tasks):
+ return_tasks = []
+
+ for task in tasks:
+ if isinstance(task, Iterable):
+ return_tasks += self.add_tasks(*task)
+ else:
+ self.add_task(task)
+ return_tasks.append(task)
+
+ return all(return_tasks) and return_tasks
+
+ def remove_tasks(self, *tasks):
+
+ return_tasks = []
+
+ if len(tasks) > 1:
+ for sub_task in tasks:
+ return_tasks += self.remove_tasks(*sub_task)
+ else:
+ return self.remove_task(tasks[0]) and tasks[0]
+
+ return all(return_tasks) and return_tasks
+
+ def has_tasks(self, *tasks):
+ return_value = True
+
+ for task in tasks:
+ if isinstance(task, Iterable):
+ return_value &= self.has_tasks(*task)
+ else:
+ return_value &= self.has_task(task)
+
+ return return_value
+
def add_dependency(self, dependent, dependency):
"""
Add a dependency for one item (task, sequence or parallel) on another
@@ -157,14 +193,22 @@ class TaskGraph(object):
: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
- dependent_tasks, dependency_tasks = \
- self._extract_dependent_and_dependency_tasks(dependent, dependency)
- for dependent_task in dependent_tasks:
- for dependency_task in dependency_tasks:
- self._graph.add_edge(dependent_task.id, dependency_task.id)
+ 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):
@@ -182,13 +226,22 @@ class TaskGraph(object):
:raise TaskNotInGraphError if either the dependent or dependency are tasks
which are not in the graph
"""
- dependent_tasks, dependency_tasks = \
- self._extract_dependent_and_dependency_tasks(dependent, dependency)
- for dependent_task in dependent_tasks:
- for dependency_task in dependency_tasks:
- if not self._graph.has_edge(dependent_task.id, dependency_task.id):
- return False
- return True
+ 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):
"""
@@ -205,61 +258,28 @@ class TaskGraph(object):
: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
- dependent_tasks, dependency_tasks = \
- self._extract_dependent_and_dependency_tasks(dependent, dependency)
- for dependent_task in dependent_tasks:
- for dependency_task in dependency_tasks:
- self._graph.remove_edge(dependent_task.id, dependency_task.id)
+ 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
def sequence(self, *items):
- group = self.parallel(*items)
+ self.add_tasks(*items)
for i in xrange(1, len(items)):
self.add_dependency(items[i], items[i-1])
- return group
-
- def parallel(self, *items):
- for item in items:
- if not isinstance(item, _TasksGroup):
- self.add_task(item)
-
- return _TasksGroup(items)
-
- def _extract_dependent_and_dependency_tasks(self, dependent, dependency):
- if isinstance(dependent, _TasksGroup):
- dependent_tasks = dependent.tasks
- else:
- if not self._graph.has_node(dependent.id):
- raise TaskNotInGraphError(
- 'dependent task {0!r} is not in graph (task id: {0.id})'.format(dependent))
- dependent_tasks = [dependent]
-
- if isinstance(dependency, _TasksGroup):
- dependency_tasks = dependency.tasks
- else:
- if not self._graph.has_node(dependency.id):
- raise TaskNotInGraphError(
- 'dependency task {0!r} is not in graph (task id: {0.id})'.format(dependency))
- dependency_tasks = [dependency]
-
- return dependent_tasks, dependency_tasks
-
-
-class _TasksGroup(object):
-
- def __init__(self, tasks):
- self._tasks = map(lambda t: t.tasks if isinstance(t, _TasksGroup) else t, tasks)
-
- @property
- def tasks(self):
- """
- The tasks in this group
- :rtype list
- :return: tasks list
- """
- return self._tasks
+ return items
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/94bbcf6f/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
index efa1e9a..96763c6 100644
--- a/tests/workflows/api/test_task_graph.py
+++ b/tests/workflows/api/test_task_graph.py
@@ -60,7 +60,7 @@ class TestTaskGraphTasks(object):
assert len(tasks) == 1
assert tasks[0] == task
- def test_remove_task_with_dependency(self, graph):
+ def test_remove_tasks_with_dependency(self, graph):
task = MockTask()
dependent_task = MockTask()
graph.add_task(task)
@@ -338,10 +338,9 @@ class TestTaskGraphArrangements(object):
assert len(list(graph.get_task_dependencies(tasks[3]))) == 2
assert len(list(graph.get_task_dependencies(tasks[4]))) == 3
-
def test_sequence_with_nested_parallel(self, graph):
tasks = [MockTask() for _ in xrange(5)]
- graph.sequence(tasks[0], graph.parallel(tasks[1], tasks[2], tasks[3]), tasks[4])
+ 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
@@ -354,7 +353,7 @@ class TestTaskGraphArrangements(object):
def test_parallel(self, graph):
tasks = [MockTask(), MockTask(), MockTask()]
- graph.parallel(*tasks)
+ graph.add_tasks(tasks)
graph_tasks = [t for t in graph.tasks]
assert set(graph_tasks) == set(tasks)
for i in xrange(len(tasks)):
@@ -366,18 +365,18 @@ class TestTaskGraphArrangements(object):
# insert some tasks and dependencies to the graph
graph.add_task(tasks[1])
graph.add_task(tasks[2])
+ graph.add_tasks(*tasks)
- graph.parallel(*tasks)
- graph_tasks = [t for t in graph.tasks]
- assert set(graph_tasks) == set(tasks)
+ assert set(graph.tasks) == set(tasks)
for i in xrange(len(tasks)):
assert len(list(graph.get_task_dependencies(tasks[i]))) == 0
def test_parallel_with_nested_sequence(self, graph):
- tasks = [MockTask() for _ in xrange(5)]
- graph.parallel(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)
+ all_tasks = [MockTask() for _ in xrange(5)]
+ task_group = 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(task_group)
# tasks[2] and tasks[3] should each have a single dependency; the rest should have none
assert len(list(graph.get_task_dependencies(tasks[0]))) == 0
@@ -388,16 +387,23 @@ class TestTaskGraphArrangements(object):
def test_parallel_with_nested_parallel(self, graph):
tasks = [MockTask() for _ in xrange(5)]
- graph.parallel(tasks[0], graph.parallel(tasks[1], tasks[2], tasks[3]), tasks[4])
+ 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_task_dependencies(tasks[i]))) == 0
- def test_sequence_and_parallel_deep_nesting(self, graph):
- # including connecting seq to parallel
- pass
+ 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_task_dependencies(tasks_list[i]))) == 0
class TestTaskGraphDependenciesWithArrangements(object):
@@ -426,7 +432,7 @@ class TestTaskGraphDependenciesWithArrangements(object):
task = MockTask()
parallel_tasks = [MockTask() for _ in xrange(3)]
graph.add_task(task)
- parallel = graph.parallel(*parallel_tasks)
+ 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
@@ -436,7 +442,7 @@ class TestTaskGraphDependenciesWithArrangements(object):
task = MockTask()
parallel_tasks = [MockTask() for _ in xrange(3)]
graph.add_task(task)
- parallel = graph.parallel(*parallel_tasks)
+ 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
@@ -446,7 +452,7 @@ class TestTaskGraphDependenciesWithArrangements(object):
sequence_tasks = [MockTask() for _ in xrange(3)]
parallel_tasks = [MockTask() for _ in xrange(3)]
sequence = graph.sequence(*sequence_tasks)
- parallel = graph.parallel(*parallel_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
@@ -457,7 +463,7 @@ class TestTaskGraphDependenciesWithArrangements(object):
sequence_tasks = [MockTask() for _ in xrange(3)]
parallel_tasks = [MockTask() for _ in xrange(3)]
sequence = graph.sequence(*sequence_tasks)
- parallel = graph.parallel(*parallel_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
@@ -468,7 +474,7 @@ class TestTaskGraphDependenciesWithArrangements(object):
task = MockTask()
parallel_tasks = [MockTask() for _ in xrange(3)]
graph.add_task(task)
- parallel = graph.parallel(*parallel_tasks)
+ 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])
@@ -481,7 +487,7 @@ class TestTaskGraphDependenciesWithArrangements(object):
sequence_tasks = [MockTask() for _ in xrange(3)]
parallel_tasks = [MockTask() for _ in xrange(3)]
sequence = graph.sequence(*sequence_tasks)
- parallel = graph.parallel(*parallel_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
@@ -514,7 +520,7 @@ class TestTaskGraphDependenciesWithArrangements(object):
task = MockTask()
parallel_tasks = [MockTask() for _ in xrange(3)]
graph.add_task(task)
- parallel = graph.parallel(*parallel_tasks)
+ 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
@@ -523,7 +529,7 @@ class TestTaskGraphDependenciesWithArrangements(object):
task = MockTask()
parallel_tasks = [MockTask() for _ in xrange(3)]
graph.add_task(task)
- parallel = graph.parallel(*parallel_tasks)
+ 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
@@ -532,7 +538,7 @@ class TestTaskGraphDependenciesWithArrangements(object):
sequence_tasks = [MockTask() for _ in xrange(3)]
parallel_tasks = [MockTask() for _ in xrange(3)]
sequence = graph.sequence(*sequence_tasks)
- parallel = graph.parallel(*parallel_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
@@ -541,7 +547,7 @@ class TestTaskGraphDependenciesWithArrangements(object):
sequence_tasks = [MockTask() for _ in xrange(3)]
parallel_tasks = [MockTask() for _ in xrange(3)]
sequence = graph.sequence(*sequence_tasks)
- parallel = graph.parallel(*parallel_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
@@ -550,7 +556,7 @@ class TestTaskGraphDependenciesWithArrangements(object):
task = MockTask()
parallel_tasks = [MockTask() for _ in xrange(3)]
graph.add_task(task)
- parallel = graph.parallel(*parallel_tasks)
+ 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
@@ -559,7 +565,7 @@ class TestTaskGraphDependenciesWithArrangements(object):
sequence_tasks = [MockTask() for _ in xrange(3)]
parallel_tasks = [MockTask() for _ in xrange(3)]
sequence = graph.sequence(*sequence_tasks)
- parallel = graph.parallel(*parallel_tasks)
+ parallel = graph.add_tasks(*parallel_tasks)
assert graph.has_dependency(sequence, parallel) is False
def test_remove_dependency_dependent_sequence(self, graph):
@@ -590,7 +596,7 @@ class TestTaskGraphDependenciesWithArrangements(object):
task = MockTask()
parallel_tasks = [MockTask() for _ in xrange(3)]
graph.add_task(task)
- parallel = graph.parallel(*parallel_tasks)
+ parallel = graph.add_tasks(*parallel_tasks)
graph.add_dependency(parallel, task)
remove_result = graph.remove_dependency(parallel, task)
assert remove_result is True
@@ -602,7 +608,7 @@ class TestTaskGraphDependenciesWithArrangements(object):
task = MockTask()
parallel_tasks = [MockTask() for _ in xrange(3)]
graph.add_task(task)
- parallel = graph.parallel(*parallel_tasks)
+ parallel = graph.add_tasks(*parallel_tasks)
graph.add_dependency(task, parallel)
remove_result = graph.remove_dependency(task, parallel)
assert remove_result is True
@@ -614,7 +620,7 @@ class TestTaskGraphDependenciesWithArrangements(object):
sequence_tasks = [MockTask() for _ in xrange(3)]
parallel_tasks = [MockTask() for _ in xrange(3)]
sequence = graph.sequence(*sequence_tasks)
- parallel = graph.parallel(*parallel_tasks)
+ parallel = graph.add_tasks(*parallel_tasks)
graph.add_dependency(sequence, parallel)
remove_result = graph.remove_dependency(sequence, parallel)
assert remove_result is True
@@ -627,7 +633,7 @@ class TestTaskGraphDependenciesWithArrangements(object):
sequence_tasks = [MockTask() for _ in xrange(3)]
parallel_tasks = [MockTask() for _ in xrange(3)]
sequence = graph.sequence(*sequence_tasks)
- parallel = graph.parallel(*parallel_tasks)
+ parallel = graph.add_tasks(*parallel_tasks)
graph.add_dependency(parallel, sequence)
remove_result = graph.remove_dependency(parallel, sequence)
assert remove_result is True
@@ -640,7 +646,7 @@ class TestTaskGraphDependenciesWithArrangements(object):
task = MockTask()
parallel_tasks = [MockTask() for _ in xrange(3)]
graph.add_task(task)
- parallel = graph.parallel(*parallel_tasks)
+ 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
@@ -654,7 +660,52 @@ class TestTaskGraphDependenciesWithArrangements(object):
sequence_tasks = [MockTask() for _ in xrange(3)]
parallel_tasks = [MockTask() for _ in xrange(3)]
sequence = graph.sequence(*sequence_tasks)
- parallel = graph.parallel(*parallel_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_task_dependencies(tasks[0])) == set()
+ for i in xrange(1, len(tasks[:-1])):
+ assert set(graph.get_task_dependencies(tasks[i])) == {tasks[i-1]}
+
+ assert set(graph.get_task_dependencies(once_nested_tasks[0])) == {tasks[2]}
+ for i in xrange(1, len(once_nested_tasks[:-1])):
+ assert set(graph.get_task_dependencies(once_nested_tasks[i])) == \
+ {tasks[2], once_nested_tasks[i-1]}
+
+ assert set(graph.get_task_dependencies(twice_nested_tasks[0])) == \
+ {tasks[2], once_nested_tasks[2]}
+ for i in xrange(1, len(twice_nested_tasks[:-1])):
+ assert set(graph.get_task_dependencies(twice_nested_tasks[i])) == \
+ {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_task_dependencies(tasks[0])) == set()
+ assert set(graph.get_task_dependencies(tasks[1])) == set()
+ assert set(graph.get_task_dependencies(tasks[2])) == {once_nested_tasks[2]}
+
+ assert set((graph.get_task_dependencies(once_nested_tasks[0]))) == set()
+ assert set(graph.get_task_dependencies(once_nested_tasks[1])) == {twice_nested_tasks[0]}
+ assert set(graph.get_task_dependencies(once_nested_tasks[2])) == set()
+
+ assert set((graph.get_task_dependencies(twice_nested_tasks[0]))) == set()
+ assert set(graph.get_task_dependencies(twice_nested_tasks[1])) == {tasks[0]}
+ assert set(graph.get_task_dependencies(twice_nested_tasks[2])) == set()