ariatosca-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From mxm...@apache.org
Subject [07/14] incubator-ariatosca git commit: ARIA-74 Add process executor extension registration hooks
Date Mon, 30 Jan 2017 14:15:05 GMT
ARIA-74 Add process executor extension registration hooks


Project: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/commit/0382b229
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/tree/0382b229
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/diff/0382b229

Branch: refs/heads/ARIA-66-Convert-custom-parser-fields-into-sqla-based-fields
Commit: 0382b229c8eddcbf95ea7792b2bdcd9d9dda303b
Parents: eaf9974
Author: Dan Kilman <dank@gigaspaces.com>
Authored: Mon Jan 23 16:55:41 2017 +0200
Committer: Dan Kilman <dank@gigaspaces.com>
Committed: Mon Jan 23 19:21:32 2017 +0200

----------------------------------------------------------------------
 aria/extension.py                               | 15 ++++
 aria/orchestrator/plugin.py                     |  9 ++-
 aria/orchestrator/workflows/executor/process.py |  5 ++
 .../executor/test_process_executor_extension.py | 80 ++++++++++++++++++++
 4 files changed, 106 insertions(+), 3 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/0382b229/aria/extension.py
----------------------------------------------------------------------
diff --git a/aria/extension.py b/aria/extension.py
index ddb7c25..4dba74f 100644
--- a/aria/extension.py
+++ b/aria/extension.py
@@ -118,8 +118,23 @@ class _ParserExtensionRegistration(_ExtensionRegistration):
 parser = _ParserExtensionRegistration()
 
 
+class _ProcessExecutorExtensionRegistration(_ExtensionRegistration):
+    """Process executor extension class decorator"""
+
+    @_registrar
+    def decorate(self):
+        """
+        The operation function executed by the process executor will be decorated with the
function
+        returned from decorate().
+        """
+        return []
+
+process_executor = _ProcessExecutorExtensionRegistration()
+
+
 def init():
     """
     Initialize all registrars by calling all registered functions
     """
     parser.init()
+    process_executor.init()

http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/0382b229/aria/orchestrator/plugin.py
----------------------------------------------------------------------
diff --git a/aria/orchestrator/plugin.py b/aria/orchestrator/plugin.py
index 3005756..381504e 100644
--- a/aria/orchestrator/plugin.py
+++ b/aria/orchestrator/plugin.py
@@ -39,14 +39,17 @@ class PluginManager(object):
         """
         metadata = wagon.show(source)
         cls = self._model.plugin.model_cls
+
+        os_props = metadata['build_server_os_properties']
+
         plugin = cls(
             archive_name=metadata['archive_name'],
             supported_platform=metadata['supported_platform'],
             supported_py_versions=metadata['supported_python_versions'],
             # Remove suffix colon after upgrading wagon to > 0.5.0
-            distribution=metadata['build_server_os_properties']['distribution:'],
-            distribution_release=metadata['build_server_os_properties']['distribution_version'],
-            distribution_version=metadata['build_server_os_properties']['distribution_release'],
+            distribution=os_props.get('distribution:') or os_props.get('distribution'),
+            distribution_release=os_props['distribution_version'],
+            distribution_version=os_props['distribution_release'],
             package_name=metadata['package_name'],
             package_version=metadata['package_version'],
             package_source=metadata['package_source'],

http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/0382b229/aria/orchestrator/workflows/executor/process.py
----------------------------------------------------------------------
diff --git a/aria/orchestrator/workflows/executor/process.py b/aria/orchestrator/workflows/executor/process.py
index 2cc9178..9fa0302 100644
--- a/aria/orchestrator/workflows/executor/process.py
+++ b/aria/orchestrator/workflows/executor/process.py
@@ -39,6 +39,8 @@ import Queue
 
 import jsonpickle
 
+import aria
+from aria.extension import process_executor
 from aria.utils import imports
 from aria.utils import exceptions
 from aria.orchestrator.workflows.executor import base
@@ -291,6 +293,9 @@ def _main():
         try:
             ctx = serialization.operation_context_from_dict(context_dict)
             task_func = imports.load_attribute(operation_mapping)
+            aria.install_aria_extensions()
+            for decorate in process_executor.decorate():
+                task_func = decorate(task_func)
             task_func(ctx=ctx, **operation_inputs)
             messenger.succeeded(tracked_changes=instrument.tracked_changes)
         except BaseException as e:

http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/0382b229/tests/orchestrator/workflows/executor/test_process_executor_extension.py
----------------------------------------------------------------------
diff --git a/tests/orchestrator/workflows/executor/test_process_executor_extension.py b/tests/orchestrator/workflows/executor/test_process_executor_extension.py
new file mode 100644
index 0000000..4a8ef57
--- /dev/null
+++ b/tests/orchestrator/workflows/executor/test_process_executor_extension.py
@@ -0,0 +1,80 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import pytest
+
+from aria import extension
+from aria.orchestrator.workflows import api
+from aria.orchestrator.workflows.core import engine
+from aria.orchestrator.workflows.executor import process
+from aria.orchestrator import workflow, operation
+
+import tests
+from tests import mock
+from tests import storage
+
+
+def test_decorate_extension(context, executor):
+    inputs = {'input1': 1, 'input2': 2}
+
+    def get_node_instance(ctx):
+        return ctx.model.node_instance.get_by_name(mock.models.DEPENDENCY_NODE_INSTANCE_NAME)
+
+    @workflow
+    def mock_workflow(ctx, graph):
+        node_instance = get_node_instance(ctx)
+        op = 'test.op'
+        op_dict = {'operation': '{0}.{1}'.format(__name__, _mock_operation.__name__)}
+        node_instance.node.operations['test.op'] = op_dict
+        task = api.task.OperationTask.node_instance(instance=node_instance, name=op, inputs=inputs)
+        graph.add_tasks(task)
+        return graph
+    graph = mock_workflow(ctx=context)  # pylint: disable=no-value-for-parameter
+    eng = engine.Engine(executor=executor, workflow_context=context, tasks_graph=graph)
+    eng.execute()
+    out = get_node_instance(context).runtime_properties['out']
+    assert out['wrapper_inputs'] == inputs
+    assert out['function_inputs'] == inputs
+
+
+@extension.process_executor
+class MockProcessExecutorExtension(object):
+
+    def decorate(self):
+        def decorator(function):
+            def wrapper(ctx, **operation_inputs):
+                ctx.node_instance.runtime_properties['out'] = {'wrapper_inputs': operation_inputs}
+                function(ctx=ctx, **operation_inputs)
+            return wrapper
+        return decorator
+
+
+@operation
+def _mock_operation(ctx, **operation_inputs):
+    ctx.node_instance.runtime_properties['out']['function_inputs'] = operation_inputs
+
+
+@pytest.fixture
+def executor():
+    result = process.ProcessExecutor(python_path=[tests.ROOT_DIR])
+    yield result
+    result.close()
+
+
+@pytest.fixture
+def context(tmpdir):
+    result = mock.context.simple(storage.get_sqlite_api_kwargs(str(tmpdir)))
+    yield result
+    storage.release_sqlite_storage(result.model)


Mime
View raw message