You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@ariatosca.apache.org by da...@apache.org on 2016/11/08 12:29:14 UTC
[06/11] incubator-ariatosca git commit: move things around
move things around
Project: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/commit/109a0ecb
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/tree/109a0ecb
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/diff/109a0ecb
Branch: refs/heads/ARIA-10-task-retries
Commit: 109a0ecb4e8e44cbdc8a9f8449d3265fc92ea784
Parents: 4870401
Author: Dan Kilman <da...@gigaspaces.com>
Authored: Tue Nov 8 12:23:44 2016 +0200
Committer: Dan Kilman <da...@gigaspaces.com>
Committed: Tue Nov 8 12:23:44 2016 +0200
----------------------------------------------------------------------
tests/workflows/core/test_engine.py | 107 +++++++++----------
tests/workflows/core/test_executor.py | 136 -------------------------
tests/workflows/executor/__init__.py | 14 +++
tests/workflows/executor/test_executor.py | 136 +++++++++++++++++++++++++
4 files changed, 205 insertions(+), 188 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/109a0ecb/tests/workflows/core/test_engine.py
----------------------------------------------------------------------
diff --git a/tests/workflows/core/test_engine.py b/tests/workflows/core/test_engine.py
index f3f72cb..e518f77 100644
--- a/tests/workflows/core/test_engine.py
+++ b/tests/workflows/core/test_engine.py
@@ -35,58 +35,7 @@ import tests.storage
global_test_holder = {}
-class TestEngine(object):
-
- def test_empty_graph_execution(self, workflow_context, executor):
- @workflow
- def mock_workflow(**_):
- pass
- self._execute(workflow_func=mock_workflow,
- workflow_context=workflow_context,
- executor=executor)
- assert workflow_context.states == ['start', 'success']
- assert workflow_context.exception is None
- assert 'sent_task_signal_calls' not in global_test_holder
-
- def test_single_task_successful_execution(self, workflow_context, executor):
- @workflow
- def mock_workflow(ctx, graph):
- graph.add_tasks(self._op(mock_success_task, ctx))
- self._execute(
- workflow_func=mock_workflow,
- workflow_context=workflow_context,
- executor=executor)
- assert workflow_context.states == ['start', 'success']
- assert workflow_context.exception is None
- assert global_test_holder.get('sent_task_signal_calls') == 1
-
- def test_single_task_failed_execution(self, workflow_context, executor):
- @workflow
- def mock_workflow(ctx, graph):
- graph.add_tasks(self._op(mock_failed_task, ctx))
- with pytest.raises(exceptions.ExecutorException):
- self._execute(
- workflow_func=mock_workflow,
- workflow_context=workflow_context,
- executor=executor)
- assert workflow_context.states == ['start', 'failure']
- assert isinstance(workflow_context.exception, exceptions.ExecutorException)
- assert global_test_holder.get('sent_task_signal_calls') == 1
-
- def test_two_tasks_execution_order(self, workflow_context, executor):
- @workflow
- def mock_workflow(ctx, graph):
- op1 = self._op(mock_ordered_task, ctx, inputs={'counter': 1})
- op2 = self._op(mock_ordered_task, ctx, inputs={'counter': 2})
- graph.sequence(op1, op2)
- self._execute(
- workflow_func=mock_workflow,
- workflow_context=workflow_context,
- executor=executor)
- assert workflow_context.states == ['start', 'success']
- assert workflow_context.exception is None
- assert global_test_holder.get('invocations') == [1, 2]
- assert global_test_holder.get('sent_task_signal_calls') == 2
+class BaseTest(object):
@staticmethod
def _execute(workflow_func, workflow_context, executor):
@@ -174,6 +123,60 @@ class TestEngine(object):
return result
+class TestEngine(BaseTest):
+
+ def test_empty_graph_execution(self, workflow_context, executor):
+ @workflow
+ def mock_workflow(**_):
+ pass
+ self._execute(workflow_func=mock_workflow,
+ workflow_context=workflow_context,
+ executor=executor)
+ assert workflow_context.states == ['start', 'success']
+ assert workflow_context.exception is None
+ assert 'sent_task_signal_calls' not in global_test_holder
+
+ def test_single_task_successful_execution(self, workflow_context, executor):
+ @workflow
+ def mock_workflow(ctx, graph):
+ graph.add_tasks(self._op(mock_success_task, ctx))
+ self._execute(
+ workflow_func=mock_workflow,
+ workflow_context=workflow_context,
+ executor=executor)
+ assert workflow_context.states == ['start', 'success']
+ assert workflow_context.exception is None
+ assert global_test_holder.get('sent_task_signal_calls') == 1
+
+ def test_single_task_failed_execution(self, workflow_context, executor):
+ @workflow
+ def mock_workflow(ctx, graph):
+ graph.add_tasks(self._op(mock_failed_task, ctx))
+ with pytest.raises(exceptions.ExecutorException):
+ self._execute(
+ workflow_func=mock_workflow,
+ workflow_context=workflow_context,
+ executor=executor)
+ assert workflow_context.states == ['start', 'failure']
+ assert isinstance(workflow_context.exception, exceptions.ExecutorException)
+ assert global_test_holder.get('sent_task_signal_calls') == 1
+
+ def test_two_tasks_execution_order(self, workflow_context, executor):
+ @workflow
+ def mock_workflow(ctx, graph):
+ op1 = self._op(mock_ordered_task, ctx, inputs={'counter': 1})
+ op2 = self._op(mock_ordered_task, ctx, inputs={'counter': 2})
+ graph.sequence(op1, op2)
+ self._execute(
+ workflow_func=mock_workflow,
+ workflow_context=workflow_context,
+ executor=executor)
+ assert workflow_context.states == ['start', 'success']
+ assert workflow_context.exception is None
+ assert global_test_holder.get('invocations') == [1, 2]
+ assert global_test_holder.get('sent_task_signal_calls') == 2
+
+
def mock_success_task():
pass
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/109a0ecb/tests/workflows/core/test_executor.py
----------------------------------------------------------------------
diff --git a/tests/workflows/core/test_executor.py b/tests/workflows/core/test_executor.py
deleted file mode 100644
index 8ec0303..0000000
--- a/tests/workflows/core/test_executor.py
+++ /dev/null
@@ -1,136 +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
-from contextlib import contextmanager
-
-import pytest
-import retrying
-
-from aria import events
-from aria.storage import models
-from aria.workflows.executor import (
- thread,
- multiprocess,
- blocking,
- # celery
-)
-
-try:
- import celery as _celery
- app = _celery.Celery()
- app.conf.update(CELERY_RESULT_BACKEND='amqp://')
-except ImportError:
- _celery = None
- app = None
-
-
-class TestExecutor(object):
-
- @pytest.mark.parametrize('executor_cls,executor_kwargs', [
- (thread.ThreadExecutor, {'pool_size': 1}),
- (thread.ThreadExecutor, {'pool_size': 2}),
- (multiprocess.MultiprocessExecutor, {'pool_size': 1}),
- (multiprocess.MultiprocessExecutor, {'pool_size': 2}),
- (blocking.CurrentThreadBlockingExecutor, {}),
- # (celery.CeleryExecutor, {'app': app})
- ])
- def test_execute(self, executor_cls, executor_kwargs):
- self.executor = executor_cls(**executor_kwargs)
- 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, MockException)
- assert isinstance(task_with_inputs.exception, MockException)
- 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 hasattr(self, 'executor'):
- self.executor.close()
-
-
-def mock_successful_task():
- pass
-
-
-def mock_failing_task():
- raise MockException
-
-
-def mock_task_with_input(input):
- raise MockException(input)
-
-if app:
- mock_successful_task = app.task(mock_successful_task)
- mock_failing_task = app.task(mock_failing_task)
- mock_task_with_input = app.task(mock_task_with_input)
-
-
-class MockException(Exception):
- pass
-
-
-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.operation_details = {'operation': operation}
- self.logger = logging.getLogger()
- self.name = name
- self.inputs = inputs or {}
-
- for state in models.Task.STATES:
- setattr(self, state.upper(), state)
-
- @contextmanager
- def update(self):
- yield self
-
-
-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/109a0ecb/tests/workflows/executor/__init__.py
----------------------------------------------------------------------
diff --git a/tests/workflows/executor/__init__.py b/tests/workflows/executor/__init__.py
new file mode 100644
index 0000000..ae1e83e
--- /dev/null
+++ b/tests/workflows/executor/__init__.py
@@ -0,0 +1,14 @@
+# 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.
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/109a0ecb/tests/workflows/executor/test_executor.py
----------------------------------------------------------------------
diff --git a/tests/workflows/executor/test_executor.py b/tests/workflows/executor/test_executor.py
new file mode 100644
index 0000000..8ec0303
--- /dev/null
+++ b/tests/workflows/executor/test_executor.py
@@ -0,0 +1,136 @@
+# 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
+from contextlib import contextmanager
+
+import pytest
+import retrying
+
+from aria import events
+from aria.storage import models
+from aria.workflows.executor import (
+ thread,
+ multiprocess,
+ blocking,
+ # celery
+)
+
+try:
+ import celery as _celery
+ app = _celery.Celery()
+ app.conf.update(CELERY_RESULT_BACKEND='amqp://')
+except ImportError:
+ _celery = None
+ app = None
+
+
+class TestExecutor(object):
+
+ @pytest.mark.parametrize('executor_cls,executor_kwargs', [
+ (thread.ThreadExecutor, {'pool_size': 1}),
+ (thread.ThreadExecutor, {'pool_size': 2}),
+ (multiprocess.MultiprocessExecutor, {'pool_size': 1}),
+ (multiprocess.MultiprocessExecutor, {'pool_size': 2}),
+ (blocking.CurrentThreadBlockingExecutor, {}),
+ # (celery.CeleryExecutor, {'app': app})
+ ])
+ def test_execute(self, executor_cls, executor_kwargs):
+ self.executor = executor_cls(**executor_kwargs)
+ 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, MockException)
+ assert isinstance(task_with_inputs.exception, MockException)
+ 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 hasattr(self, 'executor'):
+ self.executor.close()
+
+
+def mock_successful_task():
+ pass
+
+
+def mock_failing_task():
+ raise MockException
+
+
+def mock_task_with_input(input):
+ raise MockException(input)
+
+if app:
+ mock_successful_task = app.task(mock_successful_task)
+ mock_failing_task = app.task(mock_failing_task)
+ mock_task_with_input = app.task(mock_task_with_input)
+
+
+class MockException(Exception):
+ pass
+
+
+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.operation_details = {'operation': operation}
+ self.logger = logging.getLogger()
+ self.name = name
+ self.inputs = inputs or {}
+
+ for state in models.Task.STATES:
+ setattr(self, state.upper(), state)
+
+ @contextmanager
+ def update(self):
+ yield self
+
+
+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