--- /dev/null
+# 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 json
+import time
+from threading import Thread, Event
+from datetime import datetime
+
+import mock
+import pytest
+
+from aria.modeling import exceptions as modeling_exceptions
+from aria.modeling import models
+from aria.orchestrator import exceptions
+from aria.orchestrator import events
+from aria.orchestrator.workflow_runner import WorkflowRunner
+from aria.orchestrator.workflows.executor.process import ProcessExecutor
+from aria.orchestrator.workflows import api
+from aria.orchestrator.workflows.core import engine, graph_compiler
+from aria.orchestrator.workflows.executor import thread
+from aria.orchestrator import (
+ workflow,
+ operation,
+)
+
+from tests import (
+ mock as tests_mock,
+ storage
+)
+
+from ..fixtures import ( # pylint: disable=unused-import
+ plugins_dir,
+ plugin_manager,
+ fs_model as model,
+ resource_storage as resource
+)
+
+custom_events = {
+ 'is_resumed': Event(),
+ 'is_active': Event(),
+ 'execution_cancelled': Event(),
+ 'execution_failed': Event(),
+}
+
+
+class TimeoutError(BaseException):
+ pass
+
+
+class FailingTask(BaseException):
+ pass
+
+
+def test_undeclared_workflow(request):
+ # validating a proper error is raised when the workflow is not declared in the service
+ with pytest.raises(exceptions.UndeclaredWorkflowError):
+ _create_workflow_runner(request, 'undeclared_workflow')
+
+
+def test_missing_workflow_implementation(service, request):
+ # validating a proper error is raised when the workflow code path does not exist
+ workflow = models.Operation(
+ name='test_workflow',
+ service=service,
+ function='nonexistent.workflow.implementation')
+ service.workflows['test_workflow'] = workflow
+
+ with pytest.raises(exceptions.WorkflowImplementationNotFoundError):
+ _create_workflow_runner(request, 'test_workflow')
+
+
+def test_builtin_workflow_instantiation(request):
+ # validates the workflow runner instantiates properly when provided with a builtin workflow
+ # (expecting no errors to be raised on undeclared workflow or missing workflow implementation)
+ workflow_runner = _create_workflow_runner(request, 'install')
+ tasks = list(workflow_runner.execution.tasks)
+ assert len(tasks) == 18 # expecting 18 tasks for 2 node topology
+
+
+def test_custom_workflow_instantiation(request):
+ # validates the workflow runner instantiates properly when provided with a custom workflow
+ # (expecting no errors to be raised on undeclared workflow or missing workflow implementation)
+ mock_workflow = _setup_mock_workflow_in_service(request)
+ workflow_runner = _create_workflow_runner(request, mock_workflow)
+ tasks = list(workflow_runner.execution.tasks)
+ assert len(tasks) == 2 # mock workflow creates only start workflow and end workflow task
+
+
+def test_existing_active_executions(request, service, model):
+ existing_active_execution = models.Execution(
+ service=service,
+ status=models.Execution.STARTED,
+ workflow_name='uninstall')
+ model.execution.put(existing_active_execution)
+ with pytest.raises(exceptions.ActiveExecutionsError):
+ _create_workflow_runner(request, 'install')
+
+
+def test_existing_executions_but_no_active_ones(request, service, model):
+ existing_terminated_execution = models.Execution(
+ service=service,
+ status=models.Execution.SUCCEEDED,
+ workflow_name='uninstall')
+ model.execution.put(existing_terminated_execution)
+ # no active executions exist, so no error should be raised
+ _create_workflow_runner(request, 'install')
+
+
+def test_default_executor(request):
+ # validates the ProcessExecutor is used by the workflow runner by default
+ mock_workflow = _setup_mock_workflow_in_service(request)
+
+ with mock.patch('aria.orchestrator.workflow_runner.engine.Engine') as mock_engine_cls:
+ _create_workflow_runner(request, mock_workflow)
+ _, engine_kwargs = mock_engine_cls.call_args
+ assert isinstance(engine_kwargs.get('executors').values()[0], ProcessExecutor)
+
+
+def test_custom_executor(request):
+ mock_workflow = _setup_mock_workflow_in_service(request)
+
+ custom_executor = mock.MagicMock()
+ with mock.patch('aria.orchestrator.workflow_runner.engine.Engine') as mock_engine_cls:
+ _create_workflow_runner(request, mock_workflow, executor=custom_executor)
+ _, engine_kwargs = mock_engine_cls.call_args
+ assert engine_kwargs.get('executors').values()[0] == custom_executor
+
+
+def test_task_configuration_parameters(request):
+ mock_workflow = _setup_mock_workflow_in_service(request)
+
+ task_max_attempts = 5
+ task_retry_interval = 7
+ with mock.patch('aria.orchestrator.workflow_runner.engine.Engine.execute') as \
+ mock_engine_execute:
+ _create_workflow_runner(request, mock_workflow, task_max_attempts=task_max_attempts,
+ task_retry_interval=task_retry_interval).execute()
+ _, engine_kwargs = mock_engine_execute.call_args
+ assert engine_kwargs['ctx']._task_max_attempts == task_max_attempts
+ assert engine_kwargs['ctx']._task_retry_interval == task_retry_interval
+
+
+def test_execute(request, service):
+ mock_workflow = _setup_mock_workflow_in_service(request)
+
+ mock_engine = mock.MagicMock()
+ with mock.patch('aria.orchestrator.workflow_runner.engine.Engine.execute',
+ return_value=mock_engine) as mock_engine_execute:
+ workflow_runner = _create_workflow_runner(request, mock_workflow)
+ workflow_runner.execute()
+
+ _, engine_kwargs = mock_engine_execute.call_args
+ assert engine_kwargs['ctx'].service.id == service.id
+ assert engine_kwargs['ctx'].execution.workflow_name == 'test_workflow'
+
+ mock_engine_execute.assert_called_once_with(ctx=workflow_runner._workflow_context,
+ resuming=False,
+ retry_failed=False)
+
+
+def test_cancel_execution(request):
+ mock_workflow = _setup_mock_workflow_in_service(request)
+
+ mock_engine = mock.MagicMock()
+ with mock.patch('aria.orchestrator.workflow_runner.engine.Engine', return_value=mock_engine):
+ workflow_runner = _create_workflow_runner(request, mock_workflow)
+ workflow_runner.cancel()
+ mock_engine.cancel_execution.assert_called_once_with(ctx=workflow_runner._workflow_context)
+
+
+def test_execution_model_creation(request, service, model):
+ mock_workflow = _setup_mock_workflow_in_service(request)
+
+ with mock.patch('aria.orchestrator.workflow_runner.engine.Engine'):
+ workflow_runner = _create_workflow_runner(request, mock_workflow)
+
+ assert model.execution.get(workflow_runner.execution.id) == workflow_runner.execution
+ assert workflow_runner.execution.service.id == service.id
+ assert workflow_runner.execution.workflow_name == mock_workflow
+ assert workflow_runner.execution.created_at <= datetime.utcnow()
+ assert workflow_runner.execution.inputs == dict()
+
+
+def test_execution_inputs_override_workflow_inputs(request):
+ wf_inputs = {'input1': 'value1', 'input2': 'value2', 'input3': 5}
+ mock_workflow = _setup_mock_workflow_in_service(
+ request,
+ inputs=dict((name, models.Input.wrap(name, val)) for name, val
+ in wf_inputs.iteritems()))
+
+ with mock.patch('aria.orchestrator.workflow_runner.engine.Engine'):
+ workflow_runner = _create_workflow_runner(
+ request, mock_workflow, inputs={'input2': 'overriding-value2', 'input3': 7})
+
+ assert len(workflow_runner.execution.inputs) == 3
+ # did not override input1 - expecting the default value from the workflow inputs
+ assert workflow_runner.execution.inputs['input1'].value == 'value1'
+ # overrode input2
+ assert workflow_runner.execution.inputs['input2'].value == 'overriding-value2'
+ # overrode input of integer type
+ assert workflow_runner.execution.inputs['input3'].value == 7
+
+
+def test_execution_inputs_undeclared_inputs(request):
+ mock_workflow = _setup_mock_workflow_in_service(request)
+
+ with pytest.raises(modeling_exceptions.UndeclaredInputsException):
+ _create_workflow_runner(request, mock_workflow, inputs={'undeclared_input': 'value'})
+
+
+def test_execution_inputs_missing_required_inputs(request):
+ mock_workflow = _setup_mock_workflow_in_service(
+ request, inputs={'required_input': models.Input.wrap('required_input', value=None)})
+
+ with pytest.raises(modeling_exceptions.MissingRequiredInputsException):
+ _create_workflow_runner(request, mock_workflow, inputs={})
+
+
+def test_execution_inputs_wrong_type_inputs(request):
+ mock_workflow = _setup_mock_workflow_in_service(
+ request, inputs={'input': models.Input.wrap('input', 'value')})
+
+ with pytest.raises(modeling_exceptions.ParametersOfWrongTypeException):
+ _create_workflow_runner(request, mock_workflow, inputs={'input': 5})
+
+
+def test_execution_inputs_builtin_workflow_with_inputs(request):
+ # built-in workflows don't have inputs
+ with pytest.raises(modeling_exceptions.UndeclaredInputsException):
+ _create_workflow_runner(request, 'install', inputs={'undeclared_input': 'value'})
+
+
+def test_workflow_function_parameters(request, tmpdir):
+ # validating the workflow function is passed with the
+ # merged execution inputs, in dict form
+
+ # the workflow function parameters will be written to this file
+ output_path = str(tmpdir.join('output'))
+ wf_inputs = {'output_path': output_path, 'input1': 'value1', 'input2': 'value2', 'input3': 5}
+
+ mock_workflow = _setup_mock_workflow_in_service(
+ request, inputs=dict((name, models.Input.wrap(name, val)) for name, val
+ in wf_inputs.iteritems()))
+
+ _create_workflow_runner(request, mock_workflow,
+ inputs={'input2': 'overriding-value2', 'input3': 7})
+
+ with open(output_path) as f:
+ wf_call_kwargs = json.load(f)
+ assert len(wf_call_kwargs) == 3
+ assert wf_call_kwargs.get('input1') == 'value1'
+ assert wf_call_kwargs.get('input2') == 'overriding-value2'
+ assert wf_call_kwargs.get('input3') == 7
+
+
+@pytest.fixture
+def service(model):
+ # sets up a service in the storage
+ service_id = tests_mock.topology.create_simple_topology_two_nodes(model)
+ service = model.service.get(service_id)
+ return service
+
+
+def _setup_mock_workflow_in_service(request, inputs=None):
+ # sets up a mock workflow as part of the service, including uploading
+ # the workflow code to the service's dir on the resource storage
+ service = request.getfixturevalue('service')
+ resource = request.getfixturevalue('resource')
+
+ source = tests_mock.workflow.__file__
+ resource.service_template.upload(str(service.service_template.id), source)
+ mock_workflow_name = 'test_workflow'
+ arguments = {}
+ if inputs:
+ for input in inputs.itervalues():
+ arguments[input.name] = input.as_argument()
+ workflow = models.Operation(
+ name=mock_workflow_name,
+ service=service,
+ function='workflow.mock_workflow',
+ inputs=inputs or {},
+ arguments=arguments)
+ service.workflows[mock_workflow_name] = workflow
+ return mock_workflow_name
+
+
+def _create_workflow_runner(request, workflow_name, inputs=None, executor=None,
+ task_max_attempts=None, task_retry_interval=None):
+ # helper method for instantiating a workflow runner
+ service_id = request.getfixturevalue('service').id
+ model = request.getfixturevalue('model')
+ resource = request.getfixturevalue('resource')
+ plugin_manager = request.getfixturevalue('plugin_manager')
+
+ # task configuration parameters can't be set to None, therefore only
+ # passing those if they've been set by the test
+ task_configuration_kwargs = dict()
+ if task_max_attempts is not None:
+ task_configuration_kwargs['task_max_attempts'] = task_max_attempts
+ if task_retry_interval is not None:
+ task_configuration_kwargs['task_retry_interval'] = task_retry_interval
+
+ return WorkflowRunner(
+ workflow_name=workflow_name,
+ service_id=service_id,
+ inputs=inputs or {},
+ executor=executor,
+ model_storage=model,
+ resource_storage=resource,
+ plugin_manager=plugin_manager,
+ **task_configuration_kwargs)
+
+
+class TestResumableWorkflows(object):
+
+ def _create_initial_workflow_runner(
+ self, workflow_context, workflow, executor, inputs=None):
+
+ service = workflow_context.service
+ service.workflows['custom_workflow'] = tests_mock.models.create_operation(
+ 'custom_workflow',
+ operation_kwargs={
+ 'function': '{0}.{1}'.format(__name__, workflow.__name__),
+ 'inputs': dict((k, models.Input.wrap(k, v)) for k, v in (inputs or {}).items())
+ }
+ )
+ workflow_context.model.service.update(service)
+
+ wf_runner = WorkflowRunner(
+ service_id=workflow_context.service.id,
+ inputs=inputs or {},
+ model_storage=workflow_context.model,
+ resource_storage=workflow_context.resource,
+ plugin_manager=None,
+ workflow_name='custom_workflow',
+ executor=executor)
+ return wf_runner
+
+ @staticmethod
+ def _wait_for_active_and_cancel(workflow_runner):
+ if custom_events['is_active'].wait(60) is False:
+ raise TimeoutError("is_active wasn't set to True")
+ workflow_runner.cancel()
+ if custom_events['execution_cancelled'].wait(60) is False:
+ raise TimeoutError("Execution did not end")
+
+ def test_resume_workflow(self, workflow_context, thread_executor):
+ node = workflow_context.model.node.get_by_name(tests_mock.models.DEPENDENCY_NODE_NAME)
+ node.attributes['invocations'] = models.Attribute.wrap('invocations', 0)
+ self._create_interface(workflow_context, node, mock_pass_first_task_only)
+
+ wf_runner = self._create_initial_workflow_runner(
+ workflow_context, mock_parallel_tasks_workflow, thread_executor,
+ inputs={'number_of_tasks': 2})
+
+ wf_thread = Thread(target=wf_runner.execute)
+ wf_thread.daemon = True
+ wf_thread.start()
+
+ # Wait for the execution to start
+ self._wait_for_active_and_cancel(wf_runner)
+ node = workflow_context.model.node.refresh(node)
+
+ tasks = workflow_context.model.task.list(filters={'_stub_type': None})
+ assert any(task.status == task.SUCCESS for task in tasks)
+ assert any(task.status == task.RETRYING for task in tasks)
+ custom_events['is_resumed'].set()
+ assert any(task.status == task.RETRYING for task in tasks)
+
+ # Create a new workflow runner, with an existing execution id. This would cause
+ # the old execution to restart.
+ new_wf_runner = WorkflowRunner(
+ service_id=wf_runner.service.id,
+ inputs={},
+ model_storage=workflow_context.model,
+ resource_storage=workflow_context.resource,
+ plugin_manager=None,
+ execution_id=wf_runner.execution.id,
+ executor=thread_executor)
+
+ new_wf_runner.execute()
+
+ # Wait for it to finish and assert changes.
+ node = workflow_context.model.node.refresh(node)
+ assert all(task.status == task.SUCCESS for task in tasks)
+ assert node.attributes['invocations'].value == 3
+ assert wf_runner.execution.status == wf_runner.execution.SUCCEEDED
+
+ def test_resume_started_task(self, workflow_context, thread_executor):
+ node = workflow_context.model.node.get_by_name(tests_mock.models.DEPENDENCY_NODE_NAME)
+ node.attributes['invocations'] = models.Attribute.wrap('invocations', 0)
+ self._create_interface(workflow_context, node, mock_stuck_task)
+
+ wf_runner = self._create_initial_workflow_runner(
+ workflow_context, mock_parallel_tasks_workflow, thread_executor,
+ inputs={'number_of_tasks': 1})
+
+ wf_thread = Thread(target=wf_runner.execute)
+ wf_thread.daemon = True
+ wf_thread.start()
+
+ self._wait_for_active_and_cancel(wf_runner)
+ node = workflow_context.model.node.refresh(node)
+ task = workflow_context.model.task.list(filters={'_stub_type': None})[0]
+ assert node.attributes['invocations'].value == 1
+ assert task.status == task.STARTED
+ assert wf_runner.execution.status in (wf_runner.execution.CANCELLED,
+ wf_runner.execution.CANCELLING)
+ custom_events['is_resumed'].set()
+
+ new_thread_executor = thread.ThreadExecutor()
+ try:
+ new_wf_runner = WorkflowRunner(
+ service_id=wf_runner.service.id,
+ inputs={},
+ model_storage=workflow_context.model,
+ resource_storage=workflow_context.resource,
+ plugin_manager=None,
+ execution_id=wf_runner.execution.id,
+ executor=new_thread_executor)
+
+ new_wf_runner.execute()
+ finally:
+ new_thread_executor.close()
+
+ # Wait for it to finish and assert changes.
+ node = workflow_context.model.node.refresh(node)
+ assert node.attributes['invocations'].value == 2
+ assert task.status == task.SUCCESS
+ assert wf_runner.execution.status == wf_runner.execution.SUCCEEDED
+
+ def test_resume_failed_task(self, workflow_context, thread_executor):
+ node = workflow_context.model.node.get_by_name(tests_mock.models.DEPENDENCY_NODE_NAME)
+ node.attributes['invocations'] = models.Attribute.wrap('invocations', 0)
+ self._create_interface(workflow_context, node, mock_failed_before_resuming)
+
+ wf_runner = self._create_initial_workflow_runner(workflow_context,
+ mock_parallel_tasks_workflow,
+ thread_executor)
+ wf_thread = Thread(target=wf_runner.execute)
+ wf_thread.setDaemon(True)
+ wf_thread.start()
+
+ self._wait_for_active_and_cancel(wf_runner)
+ node = workflow_context.model.node.refresh(node)
+
+ task = workflow_context.model.task.list(filters={'_stub_type': None})[0]
+ assert node.attributes['invocations'].value == 2
+ assert task.status == task.STARTED
+ assert wf_runner.execution.status in (wf_runner.execution.CANCELLED,
+ wf_runner.execution.CANCELLING)
+
+ custom_events['is_resumed'].set()
+ assert node.attributes['invocations'].value == 2
+
+ # Create a new workflow runner, with an existing execution id. This would cause
+ # the old execution to restart.
+ new_thread_executor = thread.ThreadExecutor()
+ try:
+ new_wf_runner = WorkflowRunner(
+ service_id=wf_runner.service.id,
+ inputs={},
+ model_storage=workflow_context.model,
+ resource_storage=workflow_context.resource,
+ plugin_manager=None,
+ execution_id=wf_runner.execution.id,
+ executor=new_thread_executor)
+
+ new_wf_runner.execute()
+ finally:
+ new_thread_executor.close()
+
+ # Wait for it to finish and assert changes.
+ node = workflow_context.model.node.refresh(node)
+ assert node.attributes['invocations'].value == task.max_attempts - 1
+ assert task.status == task.SUCCESS
+ assert wf_runner.execution.status == wf_runner.execution.SUCCEEDED
+
+ def test_resume_failed_task_and_successful_task(self, workflow_context, thread_executor):
+ node = workflow_context.model.node.get_by_name(tests_mock.models.DEPENDENCY_NODE_NAME)
+ node.attributes['invocations'] = models.Attribute.wrap('invocations', 0)
+ self._create_interface(workflow_context, node, mock_pass_first_task_only)
+
+ wf_runner = self._create_initial_workflow_runner(
+ workflow_context,
+ mock_parallel_tasks_workflow,
+ thread_executor,
+ inputs={'retry_interval': 1, 'max_attempts': 2, 'number_of_tasks': 2}
+ )
+ wf_thread = Thread(target=wf_runner.execute)
+ wf_thread.setDaemon(True)
+ wf_thread.start()
+
+ if custom_events['execution_failed'].wait(60) is False:
+ raise TimeoutError("Execution did not end")
+
+ tasks = workflow_context.model.task.list(filters={'_stub_type': None})
+ node = workflow_context.model.node.refresh(node)
+ assert node.attributes['invocations'].value == 3
+ failed_task = [t for t in tasks if t.status == t.FAILED][0]
+
+ # First task passes
+ assert any(task.status == task.FAILED for task in tasks)
+ assert failed_task.attempts_count == 2
+ # Second task fails
+ assert any(task.status == task.SUCCESS for task in tasks)
+ assert wf_runner.execution.status in wf_runner.execution.FAILED
+
+ custom_events['is_resumed'].set()
+ new_thread_executor = thread.ThreadExecutor()
+ try:
+ new_wf_runner = WorkflowRunner(
+ service_id=wf_runner.service.id,
+ retry_failed_tasks=True,
+ inputs={},
+ model_storage=workflow_context.model,
+ resource_storage=workflow_context.resource,
+ plugin_manager=None,
+ execution_id=wf_runner.execution.id,
+ executor=new_thread_executor)
+
+ new_wf_runner.execute()
+ finally:
+ new_thread_executor.close()
+
+ # Wait for it to finish and assert changes.
+ node = workflow_context.model.node.refresh(node)
+ assert failed_task.attempts_count == 1
+ assert node.attributes['invocations'].value == 4
+ assert all(task.status == task.SUCCESS for task in tasks)
+ assert wf_runner.execution.status == wf_runner.execution.SUCCEEDED
+
+ def test_two_sequential_task_first_task_failed(self, workflow_context, thread_executor):
+ node = workflow_context.model.node.get_by_name(tests_mock.models.DEPENDENCY_NODE_NAME)
+ node.attributes['invocations'] = models.Attribute.wrap('invocations', 0)
+ self._create_interface(workflow_context, node, mock_fail_first_task_only)
+
+ wf_runner = self._create_initial_workflow_runner(
+ workflow_context,
+ mock_sequential_tasks_workflow,
+ thread_executor,
+ inputs={'retry_interval': 1, 'max_attempts': 1, 'number_of_tasks': 2}
+ )
+ wf_thread = Thread(target=wf_runner.execute)
+ wf_thread.setDaemon(True)
+ wf_thread.start()
+
+ if custom_events['execution_failed'].wait(60) is False:
+ raise TimeoutError("Execution did not end")
+
+ tasks = workflow_context.model.task.list(filters={'_stub_type': None})
+ node = workflow_context.model.node.refresh(node)
+ assert node.attributes['invocations'].value == 1
+ assert any(t.status == t.FAILED for t in tasks)
+ assert any(t.status == t.PENDING for t in tasks)
+
+ custom_events['is_resumed'].set()
+ new_thread_executor = thread.ThreadExecutor()
+ try:
+ new_wf_runner = WorkflowRunner(
+ service_id=wf_runner.service.id,
+ inputs={},
+ model_storage=workflow_context.model,
+ resource_storage=workflow_context.resource,
+ plugin_manager=None,
+ execution_id=wf_runner.execution.id,
+ executor=new_thread_executor)
+
+ new_wf_runner.execute()
+ finally:
+ new_thread_executor.close()
+
+ # Wait for it to finish and assert changes.
+ node = workflow_context.model.node.refresh(node)
+ assert node.attributes['invocations'].value == 2
+ assert any(t.status == t.SUCCESS for t in tasks)
+ assert any(t.status == t.FAILED for t in tasks)
+ assert wf_runner.execution.status == wf_runner.execution.SUCCEEDED
+
+
+
+ @staticmethod
+ @pytest.fixture
+ def thread_executor():
+ result = thread.ThreadExecutor()
+ try:
+ yield result
+ finally:
+ result.close()
+
+ @staticmethod
+ @pytest.fixture
+ def workflow_context(tmpdir):
+ workflow_context = tests_mock.context.simple(str(tmpdir))
+ yield workflow_context
+ storage.release_sqlite_storage(workflow_context.model)
+
+ @staticmethod
+ def _create_interface(ctx, node, func, arguments=None):
+ interface_name = 'aria.interfaces.lifecycle'
+ operation_kwargs = dict(function='{name}.{func.__name__}'.format(
+ name=__name__, func=func))
+ if arguments:
+ # the operation has to declare the arguments before those may be passed
+ operation_kwargs['arguments'] = arguments
+ operation_name = 'create'
+ interface = tests_mock.models.create_interface(node.service, interface_name, operation_name,
+ operation_kwargs=operation_kwargs)
+ node.interfaces[interface.name] = interface
+ ctx.model.node.update(node)
+
+ return node, interface_name, operation_name
+
+ @staticmethod
+ def _engine(workflow_func, workflow_context, executor):
+ graph = workflow_func(ctx=workflow_context)
+ execution = workflow_context.execution
+ graph_compiler.GraphCompiler(workflow_context, executor.__class__).compile(graph)
+ workflow_context.execution = execution
+
+ return engine.Engine(executors={executor.__class__: executor})
+
+ @pytest.fixture(autouse=True)
+ def register_to_events(self):
+ def execution_cancelled(*args, **kwargs):
+ custom_events['execution_cancelled'].set()
+
+ def execution_failed(*args, **kwargs):
+ custom_events['execution_failed'].set()
+
+ events.on_cancelled_workflow_signal.connect(execution_cancelled)
+ events.on_failure_workflow_signal.connect(execution_failed)
+ yield
+ events.on_cancelled_workflow_signal.disconnect(execution_cancelled)
+ events.on_failure_workflow_signal.disconnect(execution_failed)
+ for event in custom_events.values():
+ event.clear()
+
+
+@workflow
+def mock_sequential_tasks_workflow(ctx, graph,
+ retry_interval=1, max_attempts=10, number_of_tasks=1):
+ node = ctx.model.node.get_by_name(tests_mock.models.DEPENDENCY_NODE_NAME)
+ graph.sequence(*_create_tasks(node, retry_interval, max_attempts, number_of_tasks))
+
+
+@workflow
+def mock_parallel_tasks_workflow(ctx, graph,
+ retry_interval=1, max_attempts=10, number_of_tasks=1):
+ node = ctx.model.node.get_by_name(tests_mock.models.DEPENDENCY_NODE_NAME)
+ graph.add_tasks(*_create_tasks(node, retry_interval, max_attempts, number_of_tasks))
+
+
+def _create_tasks(node, retry_interval, max_attempts, number_of_tasks):
+ return [
+ api.task.OperationTask(node,
+ 'aria.interfaces.lifecycle',
+ 'create',
+ retry_interval=retry_interval,
+ max_attempts=max_attempts)
+ for _ in xrange(number_of_tasks)
+ ]
+
+
+
+@operation
+def mock_failed_before_resuming(ctx):
+ """
+ The task should run atmost ctx.task.max_attempts - 1 times, and only then pass.
+ overall, the number of invocations should be ctx.task.max_attempts - 1
+ """
+ ctx.node.attributes['invocations'] += 1
+
+ if ctx.node.attributes['invocations'] == 2:
+ custom_events['is_active'].set()
+ # unfreeze the thread only when all of the invocations are done
+ while ctx.node.attributes['invocations'] < ctx.task.max_attempts - 1:
+ time.sleep(5)
+
+ elif ctx.node.attributes['invocations'] == ctx.task.max_attempts - 1:
+ # pass only just before the end.
+ return
+ else:
+ # fail o.w.
+ raise FailingTask("stop this task")
+
+
+@operation
+def mock_stuck_task(ctx):
+ ctx.node.attributes['invocations'] += 1
+ while not custom_events['is_resumed'].isSet():
+ if not custom_events['is_active'].isSet():
+ custom_events['is_active'].set()
+ time.sleep(5)
+
+
+@operation
+def mock_pass_first_task_only(ctx):
+ ctx.node.attributes['invocations'] += 1
+
+ if ctx.node.attributes['invocations'] != 1:
+ custom_events['is_active'].set()
+ if not custom_events['is_resumed'].isSet():
+ # if resume was called, increase by one. o/w fail the execution - second task should
+ # fail as long it was not a part of resuming the workflow
+ raise FailingTask("wasn't resumed yet")
+
+
+@operation
+def mock_fail_first_task_only(ctx):
+ ctx.node.attributes['invocations'] += 1
+
+ if not custom_events['is_resumed'].isSet() and ctx.node.attributes['invocations'] == 1:
+ raise FailingTask("First task should fail")
Code Review / multicloud / azure.git / blobdiff