You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@ariatosca.apache.org by mx...@apache.org on 2017/05/07 16:47:26 UTC
[1/3] incubator-ariatosca git commit: ARIA-160 Operation toolbelt
unit tests fail spordically [Forced Update!]
Repository: incubator-ariatosca
Updated Branches:
refs/heads/ARIA-214-Dry-execution-changes-the-state-of-non-implemented-operations 4ef3c4e0a -> d5ce0c676 (forced update)
ARIA-160 Operation toolbelt unit tests fail spordically
Inroduced a filesystem based data holder
Project: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/commit/45c158ef
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/tree/45c158ef
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/diff/45c158ef
Branch: refs/heads/ARIA-214-Dry-execution-changes-the-state-of-non-implemented-operations
Commit: 45c158eff0d7ea97be0937cba9009522733948ae
Parents: 1cb3086
Author: max-orlov <ma...@gigaspaces.com>
Authored: Wed May 3 12:42:27 2017 +0300
Committer: max-orlov <ma...@gigaspaces.com>
Committed: Sun May 7 14:59:29 2017 +0300
----------------------------------------------------------------------
aria/orchestrator/workflows/core/task.py | 5 +-
tests/helpers.py | 38 ++++++++++
tests/orchestrator/context/test_operation.py | 87 +++++++++++++----------
tests/orchestrator/context/test_toolbelt.py | 18 +++--
4 files changed, 100 insertions(+), 48 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/45c158ef/aria/orchestrator/workflows/core/task.py
----------------------------------------------------------------------
diff --git a/aria/orchestrator/workflows/core/task.py b/aria/orchestrator/workflows/core/task.py
index 0e081c2..78159c4 100644
--- a/aria/orchestrator/workflows/core/task.py
+++ b/aria/orchestrator/workflows/core/task.py
@@ -175,10 +175,9 @@ class OperationTask(BaseTask):
self._update_fields = {}
try:
yield
- task = self.model_task
for key, value in self._update_fields.items():
- setattr(task, key, value)
- self.model_task = task
+ setattr(self.model_task, key, value)
+ self.model_task = self.model_task
finally:
self._update_fields = None
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/45c158ef/tests/helpers.py
----------------------------------------------------------------------
diff --git a/tests/helpers.py b/tests/helpers.py
index 472d696..423e63f 100644
--- a/tests/helpers.py
+++ b/tests/helpers.py
@@ -14,6 +14,8 @@
# limitations under the License.
import os
+import tempfile
+import json
from . import ROOT_DIR
from .resources import DIR as RESOURCES_DIR
@@ -29,3 +31,39 @@ def get_resource_uri(*args):
def get_service_template_uri(*args):
return os.path.join(RESOURCES_DIR, 'service-templates', *args)
+
+
+class FilesystemDataHolder(object):
+ _tmpfile = tempfile.NamedTemporaryFile('w')
+
+ def _load(self):
+ return json.load(open(self._tmpfile.name))
+
+ def _dump(self, value):
+ return json.dump(value, open(self._tmpfile.name, 'w'))
+
+ def __init__(self):
+ self.clear()
+
+ def __setitem__(self, key, value):
+ dict_ = self._load()
+ dict_[key] = value
+ self._dump(dict_)
+
+ def __getitem__(self, item):
+ return self._load()[item]
+
+ def __iter__(self):
+ return iter(self._load())
+
+ def clear(self):
+ self._dump({})
+
+ def get(self, item, default=None):
+ return self._load().get(item, default)
+
+ def setdefault(self, key, value):
+ dict_ = self._load()
+ return_value = dict_.setdefault(key, value)
+ self._dump(dict_)
+ return return_value
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/45c158ef/tests/orchestrator/context/test_operation.py
----------------------------------------------------------------------
diff --git a/tests/orchestrator/context/test_operation.py b/tests/orchestrator/context/test_operation.py
index 971e0db..3180d89 100644
--- a/tests/orchestrator/context/test_operation.py
+++ b/tests/orchestrator/context/test_operation.py
@@ -28,14 +28,17 @@ from aria.orchestrator import context
from aria.orchestrator.workflows import api
import tests
-from tests import mock, storage
+from tests import (
+ mock,
+ storage,
+ helpers
+)
from . import (
op_path,
execute,
)
-global_test_holder = {}
-
+global_test_holder = helpers.FilesystemDataHolder()
@pytest.fixture
def ctx(tmpdir):
@@ -75,7 +78,7 @@ def test_node_operation_task_execution(ctx, thread_executor):
node.service,
interface_name,
operation_name,
- operation_kwargs=dict(implementation=op_path(basic_operation, module_path=__name__),
+ operation_kwargs=dict(implementation=op_path(basic_node_operation, module_path=__name__),
inputs=inputs)
)
node.interfaces[interface.name] = interface
@@ -94,18 +97,11 @@ def test_node_operation_task_execution(ctx, thread_executor):
execute(workflow_func=basic_workflow, workflow_context=ctx, executor=thread_executor)
- operation_context = global_test_holder[api.task.OperationTask.NAME_FORMAT.format(
- type='node',
- name=node.name,
- interface=interface_name,
- operation=operation_name
- )]
-
- assert isinstance(operation_context, context.operation.NodeOperationContext)
+ assert global_test_holder['ctx_name'] == context.operation.NodeOperationContext.__name__
# Task bases assertions
- assert operation_context.task.actor == node
- assert operation_context.task.name == api.task.OperationTask.NAME_FORMAT.format(
+ assert global_test_holder['actor_name'] == node.name
+ assert global_test_holder['task_name'] == api.task.OperationTask.NAME_FORMAT.format(
type='node',
name=node.name,
interface=interface_name,
@@ -113,12 +109,12 @@ def test_node_operation_task_execution(ctx, thread_executor):
)
operations = interface.operations
assert len(operations) == 1
- assert operation_context.task.implementation == operations.values()[0].implementation # pylint: disable=no-member
- assert operation_context.task.inputs['putput'].value is True
+ assert global_test_holder['implementation'] == operations.values()[0].implementation # pylint: disable=no-member
+ assert global_test_holder['inputs']['putput'] is True
# Context based attributes (sugaring)
- assert operation_context.node_template == node.node_template
- assert operation_context.node == node
+ assert global_test_holder['template_name'] == node.node_template.name
+ assert global_test_holder['node_name'] == node.name
def test_relationship_operation_task_execution(ctx, thread_executor):
@@ -131,7 +127,8 @@ def test_relationship_operation_task_execution(ctx, thread_executor):
relationship.source_node.service,
interface_name,
operation_name,
- operation_kwargs=dict(implementation=op_path(basic_operation, module_path=__name__),
+ operation_kwargs=dict(implementation=op_path(basic_relationship_operation,
+ module_path=__name__),
inputs=inputs),
)
@@ -151,21 +148,14 @@ def test_relationship_operation_task_execution(ctx, thread_executor):
execute(workflow_func=basic_workflow, workflow_context=ctx, executor=thread_executor)
- operation_context = global_test_holder[api.task.OperationTask.NAME_FORMAT.format(
- type='relationship',
- name=relationship.name,
- interface=interface_name,
- operation=operation_name
- )]
-
- assert isinstance(operation_context, context.operation.RelationshipOperationContext)
+ assert global_test_holder['ctx_name'] == context.operation.RelationshipOperationContext.__name__
# Task bases assertions
- assert operation_context.task.actor == relationship
- assert interface_name in operation_context.task.name
+ assert global_test_holder['actor_name'] == relationship.name
+ assert interface_name in global_test_holder['task_name']
operations = interface.operations
- assert operation_context.task.implementation == operations.values()[0].implementation # pylint: disable=no-member
- assert operation_context.task.inputs['putput'].value is True
+ assert global_test_holder['implementation'] == operations.values()[0].implementation # pylint: disable=no-member
+ assert global_test_holder['inputs']['putput'] is True
# Context based attributes (sugaring)
dependency_node_template = ctx.model.node_template.get_by_name(
@@ -175,11 +165,11 @@ def test_relationship_operation_task_execution(ctx, thread_executor):
mock.models.DEPENDENT_NODE_TEMPLATE_NAME)
dependent_node = ctx.model.node.get_by_name(mock.models.DEPENDENT_NODE_NAME)
- assert operation_context.target_node_template == dependency_node_template
- assert operation_context.target_node == dependency_node
- assert operation_context.relationship == relationship
- assert operation_context.source_node_template == dependent_node_template
- assert operation_context.source_node == dependent_node
+ assert global_test_holder['target_node_template_name'] == dependency_node_template.name
+ assert global_test_holder['target_node_name'] == dependency_node.name
+ assert global_test_holder['relationship_name'] == relationship.name
+ assert global_test_holder['source_node_template_name'] == dependent_node_template.name
+ assert global_test_holder['source_node_name'] == dependent_node.name
def test_invalid_task_operation_id(ctx, thread_executor):
@@ -386,8 +376,29 @@ def logged_operation(ctx, **_):
@operation
-def basic_operation(ctx, **_):
- global_test_holder[ctx.name] = ctx
+def basic_node_operation(ctx, **_):
+ operation_common(ctx)
+ global_test_holder['template_name'] = ctx.node_template.name
+ global_test_holder['node_name'] = ctx.node.name
+
+
+@operation
+def basic_relationship_operation(ctx, **_):
+ operation_common(ctx)
+ global_test_holder['target_node_template_name'] = ctx.target_node_template.name
+ global_test_holder['target_node_name'] = ctx.target_node.name
+ global_test_holder['relationship_name'] = ctx.relationship.name
+ global_test_holder['source_node_template_name'] = ctx.source_node_template.name
+ global_test_holder['source_node_name'] = ctx.source_node.name
+
+
+def operation_common(ctx):
+ global_test_holder['ctx_name'] = ctx.__class__.__name__
+
+ global_test_holder['actor_name'] = ctx.task.actor.name
+ global_test_holder['task_name'] = ctx.task.name
+ global_test_holder['implementation'] = ctx.task.implementation
+ global_test_holder['inputs'] = dict(i.unwrap() for i in ctx.task.inputs.values())
@operation
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/45c158ef/tests/orchestrator/context/test_toolbelt.py
----------------------------------------------------------------------
diff --git a/tests/orchestrator/context/test_toolbelt.py b/tests/orchestrator/context/test_toolbelt.py
index ecc3ac2..822ac5a 100644
--- a/tests/orchestrator/context/test_toolbelt.py
+++ b/tests/orchestrator/context/test_toolbelt.py
@@ -19,15 +19,18 @@ from aria import workflow, operation
from aria.orchestrator import context
from aria.orchestrator.workflows import api
from aria.orchestrator.workflows.executor import thread
-from aria.orchestrator.context.toolbelt import RelationshipToolBelt
-from tests import mock, storage
+from tests import (
+ mock,
+ storage,
+ helpers
+)
from . import (
op_path,
execute,
)
-global_test_holder = {}
+global_test_holder = helpers.FilesystemDataHolder()
@pytest.fixture
@@ -85,6 +88,8 @@ def test_host_ip(workflow_context, executor):
inputs=inputs)
)
dependency_node.interfaces[interface.name] = interface
+ dependency_node.runtime_properties['ip'] = '1.1.1.1'
+
workflow_context.model.node.update(dependency_node)
@workflow
@@ -131,12 +136,11 @@ def test_relationship_tool_belt(workflow_context, executor):
execute(workflow_func=basic_workflow, workflow_context=workflow_context, executor=executor)
- assert isinstance(global_test_holder.get(api.task.OperationTask.NAME_FORMAT.format(
+ assert global_test_holder.get(api.task.OperationTask.NAME_FORMAT.format(
type='relationship',
name=relationship.name,
interface=interface_name,
- operation=operation_name
- )), RelationshipToolBelt)
+ operation=operation_name)) == relationship.source_node.name
def test_wrong_model_toolbelt():
@@ -151,7 +155,7 @@ def host_ip(toolbelt, **_):
@operation(toolbelt=True)
def relationship_operation(ctx, toolbelt, **_):
- global_test_holder[ctx.name] = toolbelt
+ global_test_holder[ctx.name] = toolbelt._op_context.source_node.name
@pytest.fixture(autouse=True)
[2/3] incubator-ariatosca git commit: ARIA-215 Refactor
plugin-related code into PluginManager
Posted by mx...@apache.org.
ARIA-215 Refactor plugin-related code into PluginManager
Refactored plugin-related code from ProcessExecutor into PluginManager.
Additionally, renamed plugin_prefix to plugin_dir in PluginManager.
Project: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/commit/3e1ed14c
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/tree/3e1ed14c
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/diff/3e1ed14c
Branch: refs/heads/ARIA-214-Dry-execution-changes-the-state-of-non-implemented-operations
Commit: 3e1ed14c00ea2c83fafdca8ec0e37817bea1d5e8
Parents: 45c158e
Author: Ran Ziv <ra...@gigaspaces.com>
Authored: Sun May 7 16:36:39 2017 +0300
Committer: Ran Ziv <ra...@gigaspaces.com>
Committed: Sun May 7 17:27:36 2017 +0300
----------------------------------------------------------------------
aria/orchestrator/plugin.py | 36 +++++++++++-
aria/orchestrator/workflows/executor/process.py | 60 ++++++--------------
aria/utils/process.py | 47 +++++++++++++++
tests/utils/test_plugin.py | 6 +-
4 files changed, 100 insertions(+), 49 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/3e1ed14c/aria/orchestrator/plugin.py
----------------------------------------------------------------------
diff --git a/aria/orchestrator/plugin.py b/aria/orchestrator/plugin.py
index f99666c..8fbcf5a 100644
--- a/aria/orchestrator/plugin.py
+++ b/aria/orchestrator/plugin.py
@@ -23,6 +23,9 @@ from datetime import datetime
import wagon
from . import exceptions
+from ..utils import process as process_utils
+
+_IS_WIN = os.name == 'nt'
class PluginManager(object):
@@ -62,11 +65,40 @@ class PluginManager(object):
raise exceptions.PluginAlreadyExistsError(
'Plugin {0}, version {1} already exists'.format(plugin.package_name,
plugin.package_version))
- self._install_wagon(source=source, prefix=self.get_plugin_prefix(plugin))
+ self._install_wagon(source=source, prefix=self.get_plugin_dir(plugin))
self._model.plugin.put(plugin)
return plugin
- def get_plugin_prefix(self, plugin):
+ def load_plugin(self, plugin, env=None):
+ """
+ Load the plugin into an environment.
+ Loading the plugin means the plugin's code and binaries paths will be appended to the
+ environment's PATH and PYTHONPATH, thereby allowing usage of the plugin.
+ :param plugin: The plugin to load
+ :param env: The environment to load the plugin into; If `None`, os.environ will be used.
+ """
+ env = env or os.environ
+ plugin_dir = self.get_plugin_dir(plugin)
+
+ # Update PATH environment variable to include plugin's bin dir
+ bin_dir = 'Scripts' if _IS_WIN else 'bin'
+ process_utils.append_to_path(os.path.join(plugin_dir, bin_dir), env=env)
+
+ # Update PYTHONPATH environment variable to include plugin's site-packages
+ # directories
+ if _IS_WIN:
+ pythonpath_dirs = [os.path.join(plugin_dir, 'Lib', 'site-packages')]
+ else:
+ # In some linux environments, there will be both a lib and a lib64 directory
+ # with the latter, containing compiled packages.
+ pythonpath_dirs = [os.path.join(
+ plugin_dir, 'lib{0}'.format(b),
+ 'python{0}.{1}'.format(sys.version_info[0], sys.version_info[1]),
+ 'site-packages') for b in ('', '64')]
+
+ process_utils.append_to_pythonpath(*pythonpath_dirs, env=env)
+
+ def get_plugin_dir(self, plugin):
return os.path.join(
self._plugins_dir,
'{0}-{1}'.format(plugin.package_name, plugin.package_version))
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/3e1ed14c/aria/orchestrator/workflows/executor/process.py
----------------------------------------------------------------------
diff --git a/aria/orchestrator/workflows/executor/process.py b/aria/orchestrator/workflows/executor/process.py
index 8481406..f3daf04 100644
--- a/aria/orchestrator/workflows/executor/process.py
+++ b/aria/orchestrator/workflows/executor/process.py
@@ -47,13 +47,12 @@ from aria.storage import instrumentation
from aria.extension import process_executor
from aria.utils import (
imports,
- exceptions
+ exceptions,
+ process as process_utils
)
from aria.modeling import types as modeling_types
-_IS_WIN = os.name == 'nt'
-
_INT_FMT = 'I'
_INT_SIZE = struct.calcsize(_INT_FMT)
UPDATE_TRACKED_CHANGES_FAILED_STR = \
@@ -127,13 +126,7 @@ class ProcessExecutor(base.BaseExecutor):
with open(arguments_json_path, 'wb') as f:
f.write(pickle.dumps(self._create_arguments_dict(task)))
- env = os.environ.copy()
- # See _update_env for plugin_prefix usage
- if task.plugin_fk and self._plugin_manager:
- plugin_prefix = self._plugin_manager.get_plugin_prefix(task.plugin)
- else:
- plugin_prefix = None
- self._update_env(env=env, plugin_prefix=plugin_prefix)
+ env = self._construct_subprocess_env(task=task)
# Asynchronously start the operation in a subprocess
subprocess.Popen(
'{0} {1} {2}'.format(sys.executable, __file__, arguments_json_path),
@@ -156,40 +149,19 @@ class ProcessExecutor(base.BaseExecutor):
'context': task.context.serialization_dict,
}
- def _update_env(self, env, plugin_prefix):
- pythonpath_dirs = []
- # If this is a plugin operation, plugin prefix will point to where
- # This plugin is installed.
- # We update the environment variables that the subprocess will be started with based on it
- if plugin_prefix:
-
- # Update PATH environment variable to include plugin's bin dir
- bin_dir = 'Scripts' if _IS_WIN else 'bin'
- env['PATH'] = '{0}{1}{2}'.format(
- os.path.join(plugin_prefix, bin_dir),
- os.pathsep,
- env.get('PATH', ''))
-
- # Update PYTHONPATH environment variable to include plugin's site-packages
- # directories
- if _IS_WIN:
- pythonpath_dirs = [os.path.join(plugin_prefix, 'Lib', 'site-packages')]
- else:
- # In some linux environments, there will be both a lib and a lib64 directory
- # with the latter, containing compiled packages.
- pythonpath_dirs = [os.path.join(
- plugin_prefix, 'lib{0}'.format(b),
- 'python{0}.{1}'.format(sys.version_info[0], sys.version_info[1]),
- 'site-packages') for b in ('', '64')]
-
- # Add used supplied directories to injected PYTHONPATH
- pythonpath_dirs.extend(self._python_path)
-
- if pythonpath_dirs:
- env['PYTHONPATH'] = '{0}{1}{2}'.format(
- os.pathsep.join(pythonpath_dirs),
- os.pathsep,
- env.get('PYTHONPATH', ''))
+ def _construct_subprocess_env(self, task):
+ env = os.environ.copy()
+
+ if task.plugin_fk and self._plugin_manager:
+ # If this is a plugin operation,
+ # load the plugin on the subprocess env we're constructing
+ self._plugin_manager.load_plugin(task.plugin, env=env)
+
+ # Add user supplied directories to injected PYTHONPATH
+ if self._python_path:
+ process_utils.append_to_pythonpath(*self._python_path, env=env)
+
+ return env
def _listener(self):
# Notify __init__ method this thread has actually started
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/3e1ed14c/aria/utils/process.py
----------------------------------------------------------------------
diff --git a/aria/utils/process.py b/aria/utils/process.py
new file mode 100644
index 0000000..9aeae67
--- /dev/null
+++ b/aria/utils/process.py
@@ -0,0 +1,47 @@
+# 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 os
+
+
+def append_to_path(*args, **kwargs):
+ """
+ Appends one or more paths to the system path of an environment.
+ The environment will be that of the current process unless another is passed using the
+ 'env' keyword argument.
+ :param args: paths to append
+ :param kwargs: 'env' may be used to pass a custom environment to use
+ """
+ _append_to_path('PATH', *args, **kwargs)
+
+
+def append_to_pythonpath(*args, **kwargs):
+ """
+ Appends one or more paths to the python path of an environment.
+ The environment will be that of the current process unless another is passed using the
+ 'env' keyword argument.
+ :param args: paths to append
+ :param kwargs: 'env' may be used to pass a custom environment to use
+ """
+ _append_to_path('PYTHONPATH', *args, **kwargs)
+
+
+def _append_to_path(path, *args, **kwargs):
+ env = kwargs.get('env') or os.environ
+ env[path] = '{0}{1}{2}'.format(
+ os.pathsep.join(args),
+ os.pathsep,
+ env.get(path, '')
+ )
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/3e1ed14c/tests/utils/test_plugin.py
----------------------------------------------------------------------
diff --git a/tests/utils/test_plugin.py b/tests/utils/test_plugin.py
index 3350247..c91d0c9 100644
--- a/tests/utils/test_plugin.py
+++ b/tests/utils/test_plugin.py
@@ -38,9 +38,9 @@ class TestPluginManager(object):
assert plugin.package_name == PACKAGE_NAME
assert plugin.package_version == PACKAGE_VERSION
assert plugin == model.plugin.get(plugin.id)
- plugin_prefix = os.path.join(plugins_dir, '{0}-{1}'.format(PACKAGE_NAME, PACKAGE_VERSION))
- assert os.path.isdir(plugin_prefix)
- assert plugin_prefix == plugin_manager.get_plugin_prefix(plugin)
+ plugin_dir = os.path.join(plugins_dir, '{0}-{1}'.format(PACKAGE_NAME, PACKAGE_VERSION))
+ assert os.path.isdir(plugin_dir)
+ assert plugin_dir == plugin_manager.get_plugin_dir(plugin)
def test_install_already_exits(self, plugin_manager, mock_plugin):
plugin_manager.install(mock_plugin)
[3/3] incubator-ariatosca git commit: ARIA-214 Dry execution changes
the state of non implemented operations
Posted by mx...@apache.org.
ARIA-214 Dry execution changes the state of non implemented operations
Project: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/commit/d5ce0c67
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/tree/d5ce0c67
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/diff/d5ce0c67
Branch: refs/heads/ARIA-214-Dry-execution-changes-the-state-of-non-implemented-operations
Commit: d5ce0c67671ad8f6922622bf8022c007e982a444
Parents: 3e1ed14
Author: max-orlov <ma...@gigaspaces.com>
Authored: Sun May 7 16:12:56 2017 +0300
Committer: max-orlov <ma...@gigaspaces.com>
Committed: Sun May 7 19:47:19 2017 +0300
----------------------------------------------------------------------
aria/orchestrator/workflows/core/task.py | 3 ---
aria/orchestrator/workflows/core/translation.py | 6 +----
aria/orchestrator/workflows/executor/base.py | 18 ++++++++------
aria/orchestrator/workflows/executor/celery.py | 2 +-
aria/orchestrator/workflows/executor/dry.py | 26 ++++++++++----------
aria/orchestrator/workflows/executor/process.py | 2 +-
aria/orchestrator/workflows/executor/thread.py | 2 +-
tests/orchestrator/workflows/core/test_task.py | 7 ++----
.../workflows/executor/test_process_executor.py | 18 +-------------
9 files changed, 30 insertions(+), 54 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/d5ce0c67/aria/orchestrator/workflows/core/task.py
----------------------------------------------------------------------
diff --git a/aria/orchestrator/workflows/core/task.py b/aria/orchestrator/workflows/core/task.py
index 78159c4..b3dfb3c 100644
--- a/aria/orchestrator/workflows/core/task.py
+++ b/aria/orchestrator/workflows/core/task.py
@@ -163,9 +163,6 @@ class OperationTask(BaseTask):
self._task_id = task_model.id
self._update_fields = None
- def execute(self):
- super(OperationTask, self).execute()
-
@contextmanager
def _update(self):
"""
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/d5ce0c67/aria/orchestrator/workflows/core/translation.py
----------------------------------------------------------------------
diff --git a/aria/orchestrator/workflows/core/translation.py b/aria/orchestrator/workflows/core/translation.py
index 0bbce90..fec108b 100644
--- a/aria/orchestrator/workflows/core/translation.py
+++ b/aria/orchestrator/workflows/core/translation.py
@@ -48,11 +48,7 @@ def build_execution_graph(
execution_graph, dependencies, default=[start_task])
if isinstance(api_task, api.task.OperationTask):
- if api_task.implementation:
- operation_task = core_task.OperationTask(api_task, executor=default_executor)
- else:
- operation_task = core_task.OperationTask(api_task,
- executor=base.EmptyOperationExecutor())
+ operation_task = core_task.OperationTask(api_task, executor=default_executor)
_add_task_and_dependencies(execution_graph, operation_task, operation_dependencies)
elif isinstance(api_task, api.task.WorkflowTask):
# Build the graph recursively while adding start and end markers
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/d5ce0c67/aria/orchestrator/workflows/executor/base.py
----------------------------------------------------------------------
diff --git a/aria/orchestrator/workflows/executor/base.py b/aria/orchestrator/workflows/executor/base.py
index a225837..f2a3c20 100644
--- a/aria/orchestrator/workflows/executor/base.py
+++ b/aria/orchestrator/workflows/executor/base.py
@@ -25,13 +25,21 @@ class BaseExecutor(logger.LoggerMixin):
"""
Base class for executors for running tasks
"""
+ def _execute(self, task):
+ raise NotImplementedError
def execute(self, task):
"""
Execute a task
:param task: task to execute
"""
- raise NotImplementedError
+ if task.implementation:
+ self._execute(task)
+ else:
+ # If a task i missing an implementation, we still need to update the state of the task,
+ # and log the operation. Here we send a signal which basically does both.
+ self._task_started(task)
+ self._task_succeeded(task)
def close(self):
"""
@@ -52,12 +60,6 @@ class BaseExecutor(logger.LoggerMixin):
events.on_success_task_signal.send(task)
-class StubTaskExecutor(BaseExecutor):
+class StubTaskExecutor(BaseExecutor): # pylint: disable=abstract-method
def execute(self, task):
task.status = task.SUCCESS
-
-
-class EmptyOperationExecutor(BaseExecutor):
- def execute(self, task):
- events.start_task_signal.send(task)
- events.on_success_task_signal.send(task)
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/d5ce0c67/aria/orchestrator/workflows/executor/celery.py
----------------------------------------------------------------------
diff --git a/aria/orchestrator/workflows/executor/celery.py b/aria/orchestrator/workflows/executor/celery.py
index 7bd9b7c..bbddc25 100644
--- a/aria/orchestrator/workflows/executor/celery.py
+++ b/aria/orchestrator/workflows/executor/celery.py
@@ -42,7 +42,7 @@ class CeleryExecutor(BaseExecutor):
self._receiver_thread.start()
self._started_queue.get(timeout=30)
- def execute(self, task):
+ def _execute(self, task):
self._tasks[task.id] = task
inputs = dict(inp.unwrap() for inp in task.inputs.values())
inputs['ctx'] = task.context
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/d5ce0c67/aria/orchestrator/workflows/executor/dry.py
----------------------------------------------------------------------
diff --git a/aria/orchestrator/workflows/executor/dry.py b/aria/orchestrator/workflows/executor/dry.py
index eb70a41..f6fb7a6 100644
--- a/aria/orchestrator/workflows/executor/dry.py
+++ b/aria/orchestrator/workflows/executor/dry.py
@@ -21,11 +21,10 @@ from datetime import datetime
from .base import BaseExecutor
-class DryExecutor(BaseExecutor):
+class DryExecutor(BaseExecutor): # pylint: disable=abstract-method
"""
Executor which dry runs tasks - prints task information without causing any side effects
"""
-
def execute(self, task):
# updating the task manually instead of calling self._task_started(task),
# to avoid any side effects raising that event might cause
@@ -33,19 +32,20 @@ class DryExecutor(BaseExecutor):
task.started_at = datetime.utcnow()
task.status = task.STARTED
- if hasattr(task.actor, 'source_node'):
- name = '{source_node.name}->{target_node.name}'.format(
- source_node=task.actor.source_node, target_node=task.actor.target_node)
- else:
- name = task.actor.name
+ if task.implementation:
+ if hasattr(task.actor, 'source_node'):
+ name = '{source_node.name}->{target_node.name}'.format(
+ source_node=task.actor.source_node, target_node=task.actor.target_node)
+ else:
+ name = task.actor.name
- task.context.logger.info(
- '<dry> {name} {task.interface_name}.{task.operation_name} started...'
- .format(name=name, task=task))
+ task.context.logger.info(
+ '<dry> {name} {task.interface_name}.{task.operation_name} started...'
+ .format(name=name, task=task))
- task.context.logger.info(
- '<dry> {name} {task.interface_name}.{task.operation_name} successful'
- .format(name=name, task=task))
+ task.context.logger.info(
+ '<dry> {name} {task.interface_name}.{task.operation_name} successful'
+ .format(name=name, task=task))
# updating the task manually instead of calling self._task_succeeded(task),
# to avoid any side effects raising that event might cause
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/d5ce0c67/aria/orchestrator/workflows/executor/process.py
----------------------------------------------------------------------
diff --git a/aria/orchestrator/workflows/executor/process.py b/aria/orchestrator/workflows/executor/process.py
index f3daf04..e464f7d 100644
--- a/aria/orchestrator/workflows/executor/process.py
+++ b/aria/orchestrator/workflows/executor/process.py
@@ -116,7 +116,7 @@ class ProcessExecutor(base.BaseExecutor):
self._server_socket.close()
self._listener_thread.join(timeout=60)
- def execute(self, task):
+ def _execute(self, task):
self._check_closed()
self._tasks[task.id] = task
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/d5ce0c67/aria/orchestrator/workflows/executor/thread.py
----------------------------------------------------------------------
diff --git a/aria/orchestrator/workflows/executor/thread.py b/aria/orchestrator/workflows/executor/thread.py
index 836b2bf..f53362a 100644
--- a/aria/orchestrator/workflows/executor/thread.py
+++ b/aria/orchestrator/workflows/executor/thread.py
@@ -46,7 +46,7 @@ class ThreadExecutor(BaseExecutor):
thread.start()
self._pool.append(thread)
- def execute(self, task):
+ def _execute(self, task):
self._queue.put(task)
def close(self):
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/d5ce0c67/tests/orchestrator/workflows/core/test_task.py
----------------------------------------------------------------------
diff --git a/tests/orchestrator/workflows/core/test_task.py b/tests/orchestrator/workflows/core/test_task.py
index 748ee20..50ca7f5 100644
--- a/tests/orchestrator/workflows/core/test_task.py
+++ b/tests/orchestrator/workflows/core/test_task.py
@@ -24,7 +24,6 @@ from aria.orchestrator.workflows import (
api,
core,
exceptions,
- executor
)
from tests import mock, storage
@@ -71,8 +70,7 @@ class TestOperationTask(object):
node,
interface_name=NODE_INTERFACE_NAME,
operation_name=NODE_OPERATION_NAME)
- core_task = core.task.OperationTask(api_task=api_task,
- executor=executor.base.EmptyOperationExecutor())
+ core_task = core.task.OperationTask(api_task=api_task, executor=None)
return api_task, core_task
def _create_relationship_operation_task(self, ctx, relationship):
@@ -81,8 +79,7 @@ class TestOperationTask(object):
relationship,
interface_name=RELATIONSHIP_INTERFACE_NAME,
operation_name=RELATIONSHIP_OPERATION_NAME)
- core_task = core.task.OperationTask(api_task=api_task,
- executor=executor.base.EmptyOperationExecutor())
+ core_task = core.task.OperationTask(api_task=api_task, executor=None)
return api_task, core_task
def test_node_operation_task_creation(self, ctx):
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/d5ce0c67/tests/orchestrator/workflows/executor/test_process_executor.py
----------------------------------------------------------------------
diff --git a/tests/orchestrator/workflows/executor/test_process_executor.py b/tests/orchestrator/workflows/executor/test_process_executor.py
index b353518..5f240b2 100644
--- a/tests/orchestrator/workflows/executor/test_process_executor.py
+++ b/tests/orchestrator/workflows/executor/test_process_executor.py
@@ -13,7 +13,6 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-import logging
import os
import Queue
@@ -66,7 +65,7 @@ class TestProcessExecutor(object):
def test_closed(self, executor):
executor.close()
with pytest.raises(RuntimeError) as exc_info:
- executor.execute(task=None)
+ executor.execute(task=MockTask(implementation='some.implementation'))
assert 'closed' in exc_info.value.message
@@ -82,18 +81,3 @@ def mock_plugin(plugin_manager, tmpdir):
source = os.path.join(tests.resources.DIR, 'plugins', 'mock-plugin1')
plugin_path = create_plugin(source=source, destination_dir=str(tmpdir))
return plugin_manager.install(source=plugin_path)
-
-
-class MockContext(object):
-
- def __init__(self, *args, **kwargs):
- self.logger = logging.getLogger('mock_logger')
- self.task = type('SubprocessMockTask', (object, ), {'plugin': None})
- self.serialization_dict = {'context_cls': self.__class__, 'context': {}}
-
- def __getattr__(self, item):
- return None
-
- @classmethod
- def deserialize_from_dict(cls, **kwargs):
- return cls()