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/12/13 10:33:48 UTC
[1/2] incubator-ariatosca git commit: ARIA-31 Add registry mechanism
for extensions [Forced Update!]
Repository: incubator-ariatosca
Updated Branches:
refs/heads/ARIA-26-plugin-mechanism 4e30e09b7 -> 5c3bee47d (forced update)
ARIA-31 Add registry mechanism for extensions
Project: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/commit/04c9bd07
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/tree/04c9bd07
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/diff/04c9bd07
Branch: refs/heads/ARIA-26-plugin-mechanism
Commit: 04c9bd07916f957ddd88b933067266177a242a42
Parents: c6c92ae
Author: Dan Kilman <da...@gigaspaces.com>
Authored: Mon Dec 5 15:28:29 2016 +0200
Committer: Dan Kilman <da...@gigaspaces.com>
Committed: Sun Dec 11 19:26:28 2016 +0200
----------------------------------------------------------------------
aria/__init__.py | 25 +--
aria/cli/commands.py | 9 +-
aria/extension.py | 125 +++++++++++++++
aria/orchestrator/events.py | 36 +++++
aria/orchestrator/events/__init__.py | 57 -------
.../events/builtin_event_handler.py | 123 ---------------
.../events/workflow_engine_event_handler.py | 74 ---------
aria/orchestrator/workflows/__init__.py | 3 +
aria/orchestrator/workflows/core/engine.py | 2 +
.../workflows/core/events_handler.py | 113 ++++++++++++++
aria/orchestrator/workflows/events_logging.py | 65 ++++++++
aria/parser/__init__.py | 5 +-
aria/parser/loading/__init__.py | 3 +-
aria/parser/loading/uri.py | 5 +-
aria/parser/presentation/__init__.py | 3 +-
aria/parser/presentation/source.py | 7 +-
aria/parser/specification.py | 6 +-
aria/storage/structures.py | 4 +-
aria/utils/plugin.py | 39 -----
aria/utils/threading.py | 7 +-
extensions/aria_extension_tosca/__init__.py | 52 ++++---
.../simple_v1_0/data_types.py | 5 +-
requirements.txt | 1 +
tests/orchestrator/conftest.py | 23 +++
tests/orchestrator/events/__init__.py | 14 --
.../events/test_builtin_event_handlers.py | 14 --
.../test_workflow_enginge_event_handlers.py | 0
.../workflows/executor/test_executor.py | 16 +-
tests/test_extension.py | 156 +++++++++++++++++++
29 files changed, 602 insertions(+), 390 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/04c9bd07/aria/__init__.py
----------------------------------------------------------------------
diff --git a/aria/__init__.py b/aria/__init__.py
index b000397..0f7bec6 100644
--- a/aria/__init__.py
+++ b/aria/__init__.py
@@ -17,13 +17,18 @@
Aria top level package
"""
-import sys
import pkgutil
+try:
+ import pkg_resources
+except ImportError:
+ pkg_resources = None
+
from .VERSION import version as __version__
from .orchestrator.decorators import workflow, operation
from . import (
+ extension,
utils,
parser,
storage,
@@ -41,19 +46,17 @@ _resource_storage = {}
def install_aria_extensions():
"""
- Iterates all Python packages with names beginning with :code:`aria_extension_` and calls
- their :code:`install_aria_extension` function if they have it.
+ Iterates all Python packages with names beginning with :code:`aria_extension_` and all
+ :code:`aria_extension` entry points and loads them.
+ It then invokes all registered extension functions.
"""
-
for loader, module_name, _ in pkgutil.iter_modules():
if module_name.startswith('aria_extension_'):
- module = loader.find_module(module_name).load_module(module_name)
-
- if hasattr(module, 'install_aria_extension'):
- module.install_aria_extension()
-
- # Loading the module has contaminated sys.modules, so we'll clean it up
- del sys.modules[module_name]
+ loader.find_module(module_name).load_module(module_name)
+ if pkg_resources:
+ for entry_point in pkg_resources.iter_entry_points(group='aria_extension'):
+ entry_point.load()
+ extension.init()
def application_model_storage(api, api_kwargs=None):
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/04c9bd07/aria/cli/commands.py
----------------------------------------------------------------------
diff --git a/aria/cli/commands.py b/aria/cli/commands.py
index 3426bb0..141da07 100644
--- a/aria/cli/commands.py
+++ b/aria/cli/commands.py
@@ -28,13 +28,14 @@ from importlib import import_module
from yaml import safe_load, YAMLError
+from .. import extension
from .. import (application_model_storage, application_resource_storage)
from ..logger import LoggerMixin
from ..storage import (FileSystemModelDriver, FileSystemResourceDriver)
from ..orchestrator.context.workflow import WorkflowContext
from ..orchestrator.workflows.core.engine import Engine
from ..orchestrator.workflows.executor.thread import ThreadExecutor
-from ..parser import (DSL_SPECIFICATION_PACKAGES, iter_specifications)
+from ..parser import iter_specifications
from ..parser.consumption import (
ConsumptionContext,
ConsumerChain,
@@ -45,7 +46,7 @@ from ..parser.consumption import (
Inputs,
Instance
)
-from ..parser.loading import (LiteralLocation, UriLocation, URI_LOADER_PREFIXES)
+from ..parser.loading import LiteralLocation, UriLocation
from ..utils.application import StorageManager
from ..utils.caching import cachedmethod
from ..utils.console import (puts, Colored, indent)
@@ -315,7 +316,7 @@ class ParseCommand(BaseCommand):
if args_namespace.prefix:
for prefix in args_namespace.prefix:
- URI_LOADER_PREFIXES.append(prefix)
+ extension.parser.uri_loader_prefix().append(prefix)
cachedmethod.ENABLED = args_namespace.cached_methods
@@ -376,7 +377,7 @@ class SpecCommand(BaseCommand):
super(SpecCommand, self).__call__(args_namespace, unknown_args)
# Make sure that all @dsl_specification decorators are processed
- for pkg in DSL_SPECIFICATION_PACKAGES:
+ for pkg in extension.parser.specification_package():
import_modules(pkg)
# TODO: scan YAML documents as well
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/04c9bd07/aria/extension.py
----------------------------------------------------------------------
diff --git a/aria/extension.py b/aria/extension.py
new file mode 100644
index 0000000..ddb7c25
--- /dev/null
+++ b/aria/extension.py
@@ -0,0 +1,125 @@
+# 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.
+
+# pylint: disable=no-self-use
+
+from .utils import collections
+
+
+class _Registrar(object):
+
+ def __init__(self, registry):
+ if not isinstance(registry, (dict, list)):
+ raise RuntimeError('Unsupported registry type')
+ self._registry = registry
+
+ def register(self, function):
+ result = function()
+ if isinstance(self._registry, dict):
+ for key in result:
+ if key in self._registry:
+ raise RuntimeError('Re-definition of {0} in {1}'.format(key, function.__name__))
+ self._registry.update(result)
+ elif isinstance(self._registry, list):
+ if not isinstance(result, (list, tuple, set)):
+ result = [result]
+ self._registry += list(result)
+ else:
+ raise RuntimeError('Illegal state')
+
+ def __call__(self):
+ return self._registry
+
+
+def _registrar(function):
+ function._registrar_function = True
+ return function
+
+
+class _ExtensionRegistration(object):
+ """Base class for extension class decorators"""
+
+ def __init__(self):
+ self._registrars = {}
+ self._registered_classes = []
+ for attr, value in vars(self.__class__).items():
+ try:
+ is_registrar_function = value._registrar_function
+ except AttributeError:
+ is_registrar_function = False
+ if is_registrar_function:
+ registrar = _Registrar(registry=getattr(self, attr)())
+ setattr(self, attr, registrar)
+ self._registrars[attr] = registrar
+
+ def __call__(self, cls):
+ self._registered_classes.append(cls)
+ return cls
+
+ def init(self):
+ """
+ Initialize all registrars by calling all registered functions
+ """
+ registered_instances = [cls() for cls in self._registered_classes]
+ for name, registrar in self._registrars.items():
+ for instance in registered_instances:
+ registrating_function = getattr(instance, name, None)
+ if registrating_function:
+ registrar.register(registrating_function)
+
+
+class _ParserExtensionRegistration(_ExtensionRegistration):
+ """Parser extensions class decorator"""
+
+ @_registrar
+ def presenter_class(self):
+ """
+ Presentation class registration.
+ Implementing functions can return a single class or a list/tuple of classes
+ """
+ return []
+
+ @_registrar
+ def specification_package(self):
+ """
+ Specification package registration.
+ Implementing functions can return a package name or a list/tuple of names
+ """
+ return []
+
+ @_registrar
+ def specification_url(self):
+ """
+ Specification URL registration.
+ Implementing functions should return a dictionary from names to URLs
+ """
+ return {}
+
+ @_registrar
+ def uri_loader_prefix(self):
+ """
+ URI loader prefix registration.
+ Implementing functions can return a single prefix or a list/tuple of prefixes
+ """
+ return collections.StrictList(value_class=basestring)
+
+parser = _ParserExtensionRegistration()
+
+
+def init():
+ """
+ Initialize all registrars by calling all registered functions
+ """
+ parser.init()
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/04c9bd07/aria/orchestrator/events.py
----------------------------------------------------------------------
diff --git a/aria/orchestrator/events.py b/aria/orchestrator/events.py
new file mode 100644
index 0000000..a1c4922
--- /dev/null
+++ b/aria/orchestrator/events.py
@@ -0,0 +1,36 @@
+# 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.
+
+"""
+ARIA's events Sub-Package
+Path: aria.events
+
+Events package provides events mechanism for different executions in aria.
+"""
+
+from blinker import signal
+
+# workflow engine task signals:
+sent_task_signal = signal('sent_task_signal')
+start_task_signal = signal('start_task_signal')
+on_success_task_signal = signal('success_task_signal')
+on_failure_task_signal = signal('failure_task_signal')
+
+# workflow engine workflow signals:
+start_workflow_signal = signal('start_workflow_signal')
+on_cancelling_workflow_signal = signal('on_cancelling_workflow_signal')
+on_cancelled_workflow_signal = signal('on_cancelled_workflow_signal')
+on_success_workflow_signal = signal('on_success_workflow_signal')
+on_failure_workflow_signal = signal('on_failure_workflow_signal')
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/04c9bd07/aria/orchestrator/events/__init__.py
----------------------------------------------------------------------
diff --git a/aria/orchestrator/events/__init__.py b/aria/orchestrator/events/__init__.py
deleted file mode 100644
index fbc0f32..0000000
--- a/aria/orchestrator/events/__init__.py
+++ /dev/null
@@ -1,57 +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.
-
-"""
-ARIA's events Sub-Package
-Path: aria.events
-
-Events package provides events mechanism for different executions in aria.
-
-
-1. storage_event_handler: implementation of storage handlers for workflow and operation events.
-2. logger_event_handler: implementation of logger handlers for workflow and operation events.
-
-API:
- * start_task_signal
- * on_success_task_signal
- * on_failure_task_signal
- * start_workflow_signal
- * on_success_workflow_signal
- * on_failure_workflow_signal
-"""
-
-import os
-
-from blinker import signal
-
-from aria.utils.plugin import plugin_installer
-
-# workflow engine task signals:
-sent_task_signal = signal('sent_task_signal')
-start_task_signal = signal('start_task_signal')
-on_success_task_signal = signal('success_task_signal')
-on_failure_task_signal = signal('failure_task_signal')
-
-# workflow engine workflow signals:
-start_workflow_signal = signal('start_workflow_signal')
-on_cancelling_workflow_signal = signal('on_cancelling_workflow_signal')
-on_cancelled_workflow_signal = signal('on_cancelled_workflow_signal')
-on_success_workflow_signal = signal('on_success_workflow_signal')
-on_failure_workflow_signal = signal('on_failure_workflow_signal')
-
-plugin_installer(
- path=os.path.dirname(os.path.realpath(__file__)),
- plugin_suffix='_event_handler',
- package=__package__)
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/04c9bd07/aria/orchestrator/events/builtin_event_handler.py
----------------------------------------------------------------------
diff --git a/aria/orchestrator/events/builtin_event_handler.py b/aria/orchestrator/events/builtin_event_handler.py
deleted file mode 100644
index c5cccfe..0000000
--- a/aria/orchestrator/events/builtin_event_handler.py
+++ /dev/null
@@ -1,123 +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.
-
-"""
-Aria's events Sub-Package
-Path: aria.events.storage_event_handler
-
-Implementation of storage handlers for workflow and operation events.
-"""
-
-
-from datetime import (
- datetime,
- timedelta,
-)
-
-from . import (
- start_workflow_signal,
- on_success_workflow_signal,
- on_failure_workflow_signal,
- on_cancelled_workflow_signal,
- on_cancelling_workflow_signal,
- sent_task_signal,
- start_task_signal,
- on_success_task_signal,
- on_failure_task_signal,
-)
-
-
-@sent_task_signal.connect
-def _task_sent(task, *args, **kwargs):
- with task._update():
- task.status = task.SENT
-
-
-@start_task_signal.connect
-def _task_started(task, *args, **kwargs):
- with task._update():
- task.started_at = datetime.utcnow()
- task.status = task.STARTED
-
-
-@on_failure_task_signal.connect
-def _task_failed(task, *args, **kwargs):
- with task._update():
- should_retry = (
- (task.retry_count < task.max_attempts - 1 or
- task.max_attempts == task.INFINITE_RETRIES) and
- # ignore_failure check here means the task will not be retries and it will be marked as
- # failed. The engine will also look at ignore_failure so it won't fail the workflow.
- not task.ignore_failure)
- if should_retry:
- task.status = task.RETRYING
- task.retry_count += 1
- task.due_at = datetime.utcnow() + timedelta(seconds=task.retry_interval)
- else:
- task.ended_at = datetime.utcnow()
- task.status = task.FAILED
-
-
-@on_success_task_signal.connect
-def _task_succeeded(task, *args, **kwargs):
- with task._update():
- task.ended_at = datetime.utcnow()
- task.status = task.SUCCESS
-
-
-@start_workflow_signal.connect
-def _workflow_started(workflow_context, *args, **kwargs):
- execution = workflow_context.execution
- execution.status = execution.STARTED
- execution.started_at = datetime.utcnow()
- workflow_context.execution = execution
-
-
-@on_failure_workflow_signal.connect
-def _workflow_failed(workflow_context, exception, *args, **kwargs):
- execution = workflow_context.execution
- execution.error = str(exception)
- execution.status = execution.FAILED
- execution.ended_at = datetime.utcnow()
- workflow_context.execution = execution
-
-
-@on_success_workflow_signal.connect
-def _workflow_succeeded(workflow_context, *args, **kwargs):
- execution = workflow_context.execution
- execution.status = execution.TERMINATED
- execution.ended_at = datetime.utcnow()
- workflow_context.execution = execution
-
-
-@on_cancelled_workflow_signal.connect
-def _workflow_cancelled(workflow_context, *args, **kwargs):
- execution = workflow_context.execution
- # _workflow_cancelling function may have called this function
- # already
- if execution.status == execution.CANCELLED:
- return
- execution.status = execution.CANCELLED
- execution.ended_at = datetime.utcnow()
- workflow_context.execution = execution
-
-
-@on_cancelling_workflow_signal.connect
-def _workflow_cancelling(workflow_context, *args, **kwargs):
- execution = workflow_context.execution
- if execution.status == execution.PENDING:
- return _workflow_cancelled(workflow_context=workflow_context)
- execution.status = execution.CANCELLING
- workflow_context.execution = execution
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/04c9bd07/aria/orchestrator/events/workflow_engine_event_handler.py
----------------------------------------------------------------------
diff --git a/aria/orchestrator/events/workflow_engine_event_handler.py b/aria/orchestrator/events/workflow_engine_event_handler.py
deleted file mode 100644
index 7df11d1..0000000
--- a/aria/orchestrator/events/workflow_engine_event_handler.py
+++ /dev/null
@@ -1,74 +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.
-
-
-"""
-Aria's events Sub-Package
-Path: aria.events.storage_event_handler
-
-Implementation of logger handlers for workflow and operation events.
-"""
-
-from . import (
- start_task_signal,
- on_success_task_signal,
- on_failure_task_signal,
- start_workflow_signal,
- on_success_workflow_signal,
- on_failure_workflow_signal,
- on_cancelled_workflow_signal,
- on_cancelling_workflow_signal,
-)
-
-
-@start_task_signal.connect
-def _start_task_handler(task, **kwargs):
- task.logger.debug('Event: Starting task: {task.name}'.format(task=task))
-
-
-@on_success_task_signal.connect
-def _success_task_handler(task, **kwargs):
- task.logger.debug('Event: Task success: {task.name}'.format(task=task))
-
-
-@on_failure_task_signal.connect
-def _failure_operation_handler(task, **kwargs):
- task.logger.error('Event: Task failure: {task.name}'.format(task=task),
- exc_info=kwargs.get('exception', True))
-
-
-@start_workflow_signal.connect
-def _start_workflow_handler(context, **kwargs):
- context.logger.debug('Event: Starting workflow: {context.name}'.format(context=context))
-
-
-@on_failure_workflow_signal.connect
-def _failure_workflow_handler(context, **kwargs):
- context.logger.debug('Event: Workflow failure: {context.name}'.format(context=context))
-
-
-@on_success_workflow_signal.connect
-def _success_workflow_handler(context, **kwargs):
- context.logger.debug('Event: Workflow success: {context.name}'.format(context=context))
-
-
-@on_cancelled_workflow_signal.connect
-def _cancel_workflow_handler(context, **kwargs):
- context.logger.debug('Event: Workflow cancelled: {context.name}'.format(context=context))
-
-
-@on_cancelling_workflow_signal.connect
-def _cancelling_workflow_handler(context, **kwargs):
- context.logger.debug('Event: Workflow cancelling: {context.name}'.format(context=context))
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/04c9bd07/aria/orchestrator/workflows/__init__.py
----------------------------------------------------------------------
diff --git a/aria/orchestrator/workflows/__init__.py b/aria/orchestrator/workflows/__init__.py
index ae1e83e..e0c979a 100644
--- a/aria/orchestrator/workflows/__init__.py
+++ b/aria/orchestrator/workflows/__init__.py
@@ -12,3 +12,6 @@
# 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 required so that logging signals are registered
+from . import events_logging
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/04c9bd07/aria/orchestrator/workflows/core/engine.py
----------------------------------------------------------------------
diff --git a/aria/orchestrator/workflows/core/engine.py b/aria/orchestrator/workflows/core/engine.py
index 2d26aeb..7886b7a 100644
--- a/aria/orchestrator/workflows/core/engine.py
+++ b/aria/orchestrator/workflows/core/engine.py
@@ -29,6 +29,8 @@ from aria.orchestrator import events
from .. import exceptions
from . import task as engine_task
from . import translation
+# Import required so all signals are registered
+from . import events_handler # pylint: disable=unused-import
class Engine(logger.LoggerMixin):
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/04c9bd07/aria/orchestrator/workflows/core/events_handler.py
----------------------------------------------------------------------
diff --git a/aria/orchestrator/workflows/core/events_handler.py b/aria/orchestrator/workflows/core/events_handler.py
new file mode 100644
index 0000000..d05cbcb
--- /dev/null
+++ b/aria/orchestrator/workflows/core/events_handler.py
@@ -0,0 +1,113 @@
+# 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.
+
+"""
+Aria's events Sub-Package
+Path: aria.events.storage_event_handler
+
+Implementation of storage handlers for workflow and operation events.
+"""
+
+
+from datetime import (
+ datetime,
+ timedelta,
+)
+
+from ... import events
+
+
+@events.sent_task_signal.connect
+def _task_sent(task, *args, **kwargs):
+ with task._update():
+ task.status = task.SENT
+
+
+@events.start_task_signal.connect
+def _task_started(task, *args, **kwargs):
+ with task._update():
+ task.started_at = datetime.utcnow()
+ task.status = task.STARTED
+
+
+@events.on_failure_task_signal.connect
+def _task_failed(task, *args, **kwargs):
+ with task._update():
+ should_retry = (
+ (task.retry_count < task.max_attempts - 1 or
+ task.max_attempts == task.INFINITE_RETRIES) and
+ # ignore_failure check here means the task will not be retries and it will be marked as
+ # failed. The engine will also look at ignore_failure so it won't fail the workflow.
+ not task.ignore_failure)
+ if should_retry:
+ task.status = task.RETRYING
+ task.retry_count += 1
+ task.due_at = datetime.utcnow() + timedelta(seconds=task.retry_interval)
+ else:
+ task.ended_at = datetime.utcnow()
+ task.status = task.FAILED
+
+
+@events.on_success_task_signal.connect
+def _task_succeeded(task, *args, **kwargs):
+ with task._update():
+ task.ended_at = datetime.utcnow()
+ task.status = task.SUCCESS
+
+
+@events.start_workflow_signal.connect
+def _workflow_started(workflow_context, *args, **kwargs):
+ execution = workflow_context.execution
+ execution.status = execution.STARTED
+ execution.started_at = datetime.utcnow()
+ workflow_context.execution = execution
+
+
+@events.on_failure_workflow_signal.connect
+def _workflow_failed(workflow_context, exception, *args, **kwargs):
+ execution = workflow_context.execution
+ execution.error = str(exception)
+ execution.status = execution.FAILED
+ execution.ended_at = datetime.utcnow()
+ workflow_context.execution = execution
+
+
+@events.on_success_workflow_signal.connect
+def _workflow_succeeded(workflow_context, *args, **kwargs):
+ execution = workflow_context.execution
+ execution.status = execution.TERMINATED
+ execution.ended_at = datetime.utcnow()
+ workflow_context.execution = execution
+
+
+@events.on_cancelled_workflow_signal.connect
+def _workflow_cancelled(workflow_context, *args, **kwargs):
+ execution = workflow_context.execution
+ # _workflow_cancelling function may have called this function
+ # already
+ if execution.status == execution.CANCELLED:
+ return
+ execution.status = execution.CANCELLED
+ execution.ended_at = datetime.utcnow()
+ workflow_context.execution = execution
+
+
+@events.on_cancelling_workflow_signal.connect
+def _workflow_cancelling(workflow_context, *args, **kwargs):
+ execution = workflow_context.execution
+ if execution.status == execution.PENDING:
+ return _workflow_cancelled(workflow_context=workflow_context)
+ execution.status = execution.CANCELLING
+ workflow_context.execution = execution
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/04c9bd07/aria/orchestrator/workflows/events_logging.py
----------------------------------------------------------------------
diff --git a/aria/orchestrator/workflows/events_logging.py b/aria/orchestrator/workflows/events_logging.py
new file mode 100644
index 0000000..409ce0a
--- /dev/null
+++ b/aria/orchestrator/workflows/events_logging.py
@@ -0,0 +1,65 @@
+# 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.
+
+
+"""
+Aria's events Sub-Package
+Path: aria.events.storage_event_handler
+
+Implementation of logger handlers for workflow and operation events.
+"""
+
+from .. import events
+
+
+@events.start_task_signal.connect
+def _start_task_handler(task, **kwargs):
+ task.logger.debug('Event: Starting task: {task.name}'.format(task=task))
+
+
+@events.on_success_task_signal.connect
+def _success_task_handler(task, **kwargs):
+ task.logger.debug('Event: Task success: {task.name}'.format(task=task))
+
+
+@events.on_failure_task_signal.connect
+def _failure_operation_handler(task, **kwargs):
+ task.logger.error('Event: Task failure: {task.name}'.format(task=task),
+ exc_info=kwargs.get('exception', True))
+
+
+@events.start_workflow_signal.connect
+def _start_workflow_handler(context, **kwargs):
+ context.logger.debug('Event: Starting workflow: {context.name}'.format(context=context))
+
+
+@events.on_failure_workflow_signal.connect
+def _failure_workflow_handler(context, **kwargs):
+ context.logger.debug('Event: Workflow failure: {context.name}'.format(context=context))
+
+
+@events.on_success_workflow_signal.connect
+def _success_workflow_handler(context, **kwargs):
+ context.logger.debug('Event: Workflow success: {context.name}'.format(context=context))
+
+
+@events.on_cancelled_workflow_signal.connect
+def _cancel_workflow_handler(context, **kwargs):
+ context.logger.debug('Event: Workflow cancelled: {context.name}'.format(context=context))
+
+
+@events.on_cancelling_workflow_signal.connect
+def _cancelling_workflow_handler(context, **kwargs):
+ context.logger.debug('Event: Workflow cancelling: {context.name}'.format(context=context))
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/04c9bd07/aria/parser/__init__.py
----------------------------------------------------------------------
diff --git a/aria/parser/__init__.py b/aria/parser/__init__.py
index 2a83cd4..9ab8785 100644
--- a/aria/parser/__init__.py
+++ b/aria/parser/__init__.py
@@ -13,8 +13,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-from .specification import (DSL_SPECIFICATION_PACKAGES, DSL_SPECIFICATION_URLS, dsl_specification,
- iter_specifications)
+from .specification import dsl_specification, iter_specifications
MODULES = (
@@ -27,7 +26,5 @@ MODULES = (
__all__ = (
'MODULES',
- 'DSL_SPECIFICATION_PACKAGES',
- 'DSL_SPECIFICATION_URLS',
'dsl_specification',
'iter_specifications')
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/04c9bd07/aria/parser/loading/__init__.py
----------------------------------------------------------------------
diff --git a/aria/parser/loading/__init__.py b/aria/parser/loading/__init__.py
index f331e39..006f164 100644
--- a/aria/parser/loading/__init__.py
+++ b/aria/parser/loading/__init__.py
@@ -20,7 +20,7 @@ from .loader import Loader
from .source import LoaderSource, DefaultLoaderSource
from .location import Location, UriLocation, LiteralLocation
from .literal import LiteralLoader
-from .uri import URI_LOADER_PREFIXES, UriTextLoader
+from .uri import UriTextLoader
from .request import SESSION, SESSION_CACHE_PATH, RequestLoader, RequestTextLoader
from .file import FileTextLoader
@@ -37,7 +37,6 @@ __all__ = (
'UriLocation',
'LiteralLocation',
'LiteralLoader',
- 'URI_LOADER_PREFIXES',
'UriTextLoader',
'SESSION',
'SESSION_CACHE_PATH',
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/04c9bd07/aria/parser/loading/uri.py
----------------------------------------------------------------------
diff --git a/aria/parser/loading/uri.py b/aria/parser/loading/uri.py
index f94a003..f0cde3a 100644
--- a/aria/parser/loading/uri.py
+++ b/aria/parser/loading/uri.py
@@ -16,6 +16,7 @@
import os
from urlparse import urljoin
+from ...extension import parser
from ...utils.collections import StrictList
from ...utils.uris import as_file
from .loader import Loader
@@ -23,8 +24,6 @@ from .file import FileTextLoader
from .request import RequestTextLoader
from .exceptions import DocumentNotFoundException
-URI_LOADER_PREFIXES = StrictList(value_class=basestring)
-
class UriTextLoader(Loader):
"""
@@ -58,7 +57,7 @@ class UriTextLoader(Loader):
add_prefix(origin_location.prefix)
add_prefixes(context.prefixes)
- add_prefixes(URI_LOADER_PREFIXES)
+ add_prefixes(parser.uri_loader_prefix())
def open(self):
try:
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/04c9bd07/aria/parser/presentation/__init__.py
----------------------------------------------------------------------
diff --git a/aria/parser/presentation/__init__.py b/aria/parser/presentation/__init__.py
index ba7a163..a681695 100644
--- a/aria/parser/presentation/__init__.py
+++ b/aria/parser/presentation/__init__.py
@@ -18,7 +18,7 @@ from .exceptions import PresenterException, PresenterNotFoundError
from .context import PresentationContext
from .presenter import Presenter
from .presentation import Value, PresentationBase, Presentation, AsIsPresentation
-from .source import PRESENTER_CLASSES, PresenterSource, DefaultPresenterSource
+from .source import PresenterSource, DefaultPresenterSource
from .null import NULL, none_to_null, null_to_none
from .fields import (Field, has_fields, short_form_field, allow_unknown_fields, primitive_field,
primitive_list_field, primitive_dict_field, primitive_dict_unknown_fields,
@@ -42,7 +42,6 @@ __all__ = (
'Presentation',
'AsIsPresentation',
'PresenterSource',
- 'PRESENTER_CLASSES',
'DefaultPresenterSource',
'NULL',
'none_to_null',
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/04c9bd07/aria/parser/presentation/source.py
----------------------------------------------------------------------
diff --git a/aria/parser/presentation/source.py b/aria/parser/presentation/source.py
index 8ff4cab..7198b07 100644
--- a/aria/parser/presentation/source.py
+++ b/aria/parser/presentation/source.py
@@ -13,9 +13,10 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-from .exceptions import PresenterNotFoundError
-PRESENTER_CLASSES = []
+from ...extension import parser
+
+from .exceptions import PresenterNotFoundError
class PresenterSource(object):
@@ -36,7 +37,7 @@ class DefaultPresenterSource(PresenterSource):
def __init__(self, classes=None):
if classes is None:
- classes = PRESENTER_CLASSES
+ classes = parser.presenter_class()
self.classes = classes
def get_presenter(self, raw):
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/04c9bd07/aria/parser/specification.py
----------------------------------------------------------------------
diff --git a/aria/parser/specification.py b/aria/parser/specification.py
index 1c7e1f2..1df11ce 100644
--- a/aria/parser/specification.py
+++ b/aria/parser/specification.py
@@ -15,12 +15,10 @@
import re
+from ..extension import parser
from ..utils.collections import OrderedDict
from ..utils.formatting import full_type_name
-
-DSL_SPECIFICATION_PACKAGES = []
-DSL_SPECIFICATION_URLS = {}
_DSL_SPECIFICATIONS = {}
@@ -84,7 +82,7 @@ def _section_key(value):
def _fix_details(details, spec):
code = details.get('code')
doc = details.get('doc')
- url = DSL_SPECIFICATION_URLS.get(spec)
+ url = parser.specification_url().get(spec)
if (url is not None) and (doc is not None):
# Look for a URL in ReST docstring that begins with our url
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/04c9bd07/aria/storage/structures.py
----------------------------------------------------------------------
diff --git a/aria/storage/structures.py b/aria/storage/structures.py
index 8dbd2a9..8afa40c 100644
--- a/aria/storage/structures.py
+++ b/aria/storage/structures.py
@@ -106,11 +106,11 @@ class _MutableType(TypeDecorator):
def python_type(self):
raise NotImplementedError
+ impl = VARCHAR
+
def process_literal_param(self, value, dialect):
pass
- impl = VARCHAR
-
def process_bind_param(self, value, dialect):
if value is not None:
value = json.dumps(value)
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/04c9bd07/aria/utils/plugin.py
----------------------------------------------------------------------
diff --git a/aria/utils/plugin.py b/aria/utils/plugin.py
deleted file mode 100644
index bb2b974..0000000
--- a/aria/utils/plugin.py
+++ /dev/null
@@ -1,39 +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.
-
-"""
-Contains utility methods that enable dynamic python code loading
-# TODO: merge with tools.module
-"""
-
-import os
-from importlib import import_module
-
-
-def plugin_installer(path, plugin_suffix, package=None, callback=None):
- """
- Load each module under ``path`` that ends with ``plugin_suffix``. If ``callback`` is supplied,
- call it with each loaded module.
- """
- assert callback is None or callable(callback)
- plugin_suffix = '{0}.py'.format(plugin_suffix)
-
- for file_name in os.listdir(path):
- if not file_name.endswith(plugin_suffix):
- continue
- module_name = '{0}.{1}'.format(package, file_name[:-3]) if package else file_name[:-3]
- module = import_module(module_name)
- if callback:
- callback(module)
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/04c9bd07/aria/utils/threading.py
----------------------------------------------------------------------
diff --git a/aria/utils/threading.py b/aria/utils/threading.py
index 575d011..b99250d 100644
--- a/aria/utils/threading.py
+++ b/aria/utils/threading.py
@@ -90,7 +90,7 @@ class FixedThreadPoolExecutor(object):
_CYANIDE = object() # Special task marker used to kill worker threads.
def __init__(self,
- size=multiprocessing.cpu_count() * 2 + 1,
+ size=None,
timeout=None,
print_exceptions=False):
"""
@@ -100,6 +100,11 @@ class FixedThreadPoolExecutor(object):
:param print_exceptions: Set to true in order to
print exceptions from tasks. (Defaults to false)
"""
+ if not size:
+ try:
+ size = multiprocessing.cpu_count() * 2 + 1
+ except NotImplementedError:
+ size = 3
self.size = size
self.timeout = timeout
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/04c9bd07/extensions/aria_extension_tosca/__init__.py
----------------------------------------------------------------------
diff --git a/extensions/aria_extension_tosca/__init__.py b/extensions/aria_extension_tosca/__init__.py
index 54e1c84..d93dce2 100644
--- a/extensions/aria_extension_tosca/__init__.py
+++ b/extensions/aria_extension_tosca/__init__.py
@@ -15,34 +15,38 @@
import os.path
-from aria.parser import (DSL_SPECIFICATION_PACKAGES, DSL_SPECIFICATION_URLS)
-from aria.parser.presentation import PRESENTER_CLASSES
-from aria.parser.loading import URI_LOADER_PREFIXES
+from aria import extension
from .simple_v1_0 import ToscaSimplePresenter1_0
from .simple_nfv_v1_0 import ToscaSimpleNfvPresenter1_0
-def install_aria_extension():
- '''
- Installs the TOSCA extension to ARIA.
- '''
-
- global PRESENTER_CLASSES # pylint: disable=global-statement
- PRESENTER_CLASSES += (ToscaSimplePresenter1_0, ToscaSimpleNfvPresenter1_0)
-
- # DSL specification
- DSL_SPECIFICATION_PACKAGES.append('aria_extension_tosca')
- DSL_SPECIFICATION_URLS['yaml-1.1'] = \
- 'http://yaml.org'
- DSL_SPECIFICATION_URLS['tosca-simple-1.0'] = \
- 'http://docs.oasis-open.org/tosca/TOSCA-Simple-Profile-YAML/v1.0/cos01' \
- '/TOSCA-Simple-Profile-YAML-v1.0-cos01.html'
- DSL_SPECIFICATION_URLS['tosca-simple-nfv-1.0'] = \
- 'http://docs.oasis-open.org/tosca/tosca-nfv/v1.0/tosca-nfv-v1.0.html'
-
- # Imports
- the_dir = os.path.dirname(__file__)
- URI_LOADER_PREFIXES.append(os.path.join(the_dir, 'profiles'))
+
+@extension.parser
+class ParserExtensions(object):
+
+ @staticmethod
+ def presenter_class():
+ return ToscaSimplePresenter1_0, ToscaSimpleNfvPresenter1_0
+
+ @staticmethod
+ def specification_package():
+ return 'aria_extension_tosca'
+
+ @staticmethod
+ def specification_url():
+ return {
+ 'yaml-1.1': 'http://yaml.org',
+ 'tosca-simple-1.0': 'http://docs.oasis-open.org/tosca/TOSCA-Simple-Profile-YAML/v1.0/'
+ 'cos01/TOSCA-Simple-Profile-YAML-v1.0-cos01.html',
+ 'tosca-simple-nfv-1.0': 'http://docs.oasis-open.org/tosca/tosca-nfv/v1.0/'
+ 'tosca-nfv-v1.0.html'
+ }
+
+ @staticmethod
+ def uri_loader_prefix():
+ the_dir = os.path.dirname(__file__)
+ return os.path.join(the_dir, 'profiles')
+
MODULES = (
'simple_v1_0',
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/04c9bd07/extensions/aria_extension_tosca/simple_v1_0/data_types.py
----------------------------------------------------------------------
diff --git a/extensions/aria_extension_tosca/simple_v1_0/data_types.py b/extensions/aria_extension_tosca/simple_v1_0/data_types.py
index 1fdbe6e..a06834c 100644
--- a/extensions/aria_extension_tosca/simple_v1_0/data_types.py
+++ b/extensions/aria_extension_tosca/simple_v1_0/data_types.py
@@ -14,8 +14,11 @@
# limitations under the License.
import re
-from functools import total_ordering
from datetime import datetime, tzinfo, timedelta
+try:
+ from functools import total_ordering
+except ImportError:
+ from total_ordering import total_ordering
from aria.parser import dsl_specification
from aria.utils.collections import StrictDict, OrderedDict
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/04c9bd07/requirements.txt
----------------------------------------------------------------------
diff --git a/requirements.txt b/requirements.txt
index 7e87c67..31b0b79 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -17,6 +17,7 @@ retrying==1.3.3
blinker==1.4
importlib==1.0.4 ; python_version < '2.7'
ordereddict==1.1 ; python_version < '2.7'
+total-ordering==0.1.0 ; python_version < '2.7'
jsonpickle
ruamel.yaml==0.11.15
Jinja2==2.8
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/04c9bd07/tests/orchestrator/conftest.py
----------------------------------------------------------------------
diff --git a/tests/orchestrator/conftest.py b/tests/orchestrator/conftest.py
new file mode 100644
index 0000000..4b24f18
--- /dev/null
+++ b/tests/orchestrator/conftest.py
@@ -0,0 +1,23 @@
+# 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
+
+import aria
+
+
+@pytest.fixture(scope='session', autouse=True)
+def install_aria_extensions():
+ aria.install_aria_extensions()
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/04c9bd07/tests/orchestrator/events/__init__.py
----------------------------------------------------------------------
diff --git a/tests/orchestrator/events/__init__.py b/tests/orchestrator/events/__init__.py
deleted file mode 100644
index ae1e83e..0000000
--- a/tests/orchestrator/events/__init__.py
+++ /dev/null
@@ -1,14 +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.
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/04c9bd07/tests/orchestrator/events/test_builtin_event_handlers.py
----------------------------------------------------------------------
diff --git a/tests/orchestrator/events/test_builtin_event_handlers.py b/tests/orchestrator/events/test_builtin_event_handlers.py
deleted file mode 100644
index ae1e83e..0000000
--- a/tests/orchestrator/events/test_builtin_event_handlers.py
+++ /dev/null
@@ -1,14 +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.
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/04c9bd07/tests/orchestrator/events/test_workflow_enginge_event_handlers.py
----------------------------------------------------------------------
diff --git a/tests/orchestrator/events/test_workflow_enginge_event_handlers.py b/tests/orchestrator/events/test_workflow_enginge_event_handlers.py
deleted file mode 100644
index e69de29..0000000
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/04c9bd07/tests/orchestrator/workflows/executor/test_executor.py
----------------------------------------------------------------------
diff --git a/tests/orchestrator/workflows/executor/test_executor.py b/tests/orchestrator/workflows/executor/test_executor.py
index a425799..654542c 100644
--- a/tests/orchestrator/workflows/executor/test_executor.py
+++ b/tests/orchestrator/workflows/executor/test_executor.py
@@ -20,6 +20,14 @@ from contextlib import contextmanager
import pytest
import retrying
+try:
+ import celery as _celery
+ app = _celery.Celery()
+ app.conf.update(CELERY_RESULT_BACKEND='amqp://')
+except ImportError:
+ _celery = None
+ app = None
+
from aria.storage import models
from aria.orchestrator import events
from aria.orchestrator.workflows.executor import (
@@ -29,14 +37,6 @@ from aria.orchestrator.workflows.executor import (
# 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):
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/04c9bd07/tests/test_extension.py
----------------------------------------------------------------------
diff --git a/tests/test_extension.py b/tests/test_extension.py
new file mode 100644
index 0000000..f0378fd
--- /dev/null
+++ b/tests/test_extension.py
@@ -0,0 +1,156 @@
+# 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
+
+# #pylint: disable=no-member,no-method-argument,unused-variable
+
+
+class TestRegistrar(object):
+
+ def test_list_based_registrar_with_single_element_registration(self):
+ class ExtensionRegistration(extension._ExtensionRegistration):
+ @extension._registrar
+ def list_based_registrar(*_):
+ return []
+ extension_registration = ExtensionRegistration()
+
+ @extension_registration
+ class Extension(object):
+ def list_based_registrar(self):
+ return True
+
+ assert extension_registration.list_based_registrar() == []
+ extension_registration.init()
+ assert extension_registration.list_based_registrar() == [True]
+
+ def test_list_based_registrar_with_sequence_element_registration(self):
+ class ExtensionRegistration(extension._ExtensionRegistration):
+ @extension._registrar
+ def list_based_registrar1(*_):
+ return []
+
+ @extension._registrar
+ def list_based_registrar2(*_):
+ return []
+
+ @extension._registrar
+ def list_based_registrar3(*_):
+ return []
+ extension_registration = ExtensionRegistration()
+
+ @extension_registration
+ class Extension(object):
+ def list_based_registrar1(*_):
+ return [True, True]
+
+ def list_based_registrar2(*_):
+ return True, True
+
+ def list_based_registrar3(*_):
+ return set([True])
+
+ extension_registration.init()
+ assert extension_registration.list_based_registrar1() == [True, True]
+ assert extension_registration.list_based_registrar2() == [True, True]
+ assert extension_registration.list_based_registrar3() == [True]
+
+ def test_dict_based_registrar(self):
+ class ExtensionRegistration(extension._ExtensionRegistration):
+ @extension._registrar
+ def dict_based_registrar(*_):
+ return {}
+ extension_registration = ExtensionRegistration()
+
+ @extension_registration
+ class Extension1(object):
+ def dict_based_registrar(self):
+ return {
+ 'a': 'a',
+ 'b': 'b'
+ }
+
+ @extension_registration
+ class Extension2(object):
+ def dict_based_registrar(self):
+ return {
+ 'c': 'c',
+ 'd': 'd'
+ }
+
+ assert extension_registration.dict_based_registrar() == {}
+ extension_registration.init()
+ assert extension_registration.dict_based_registrar() == {
+ 'a': 'a',
+ 'b': 'b',
+ 'c': 'c',
+ 'd': 'd'
+ }
+
+ def test_invalid_duplicate_key_dict_based_registrar(self):
+ class ExtensionRegistration(extension._ExtensionRegistration):
+ @extension._registrar
+ def dict_based_registrar(*_):
+ return {}
+ extension_registration = ExtensionRegistration()
+
+ @extension_registration
+ class Extension1(object):
+ def dict_based_registrar(self):
+ return {
+ 'a': 'val1',
+ }
+
+ @extension_registration
+ class Extension2(object):
+ def dict_based_registrar(self):
+ return {
+ 'a': 'val2',
+ }
+
+ with pytest.raises(RuntimeError):
+ extension_registration.init()
+
+ def test_unsupported_registrar(self):
+ with pytest.raises(RuntimeError):
+ class ExtensionRegistration(extension._ExtensionRegistration):
+ @extension._registrar
+ def unsupported_registrar(*_):
+ return set()
+ extension_registration = ExtensionRegistration()
+
+ @extension_registration
+ class Extension(object):
+ def unsupported_registrar(self):
+ return True
+
+ extension_registration.init()
+
+ def test_unimplemented_registration(self):
+ class ExtensionRegistration(extension._ExtensionRegistration):
+ @extension._registrar
+ def list_based_registrar(*_):
+ return []
+ extension_registration = ExtensionRegistration()
+
+ @extension_registration
+ class Extension(object):
+ pass
+
+ assert extension_registration.list_based_registrar() == []
+ extension_registration.init()
+ assert extension_registration.list_based_registrar() == []
[2/2] incubator-ariatosca git commit: ARIA-26 TBD
Posted by da...@apache.org.
ARIA-26 TBD
Project: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/commit/5c3bee47
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/tree/5c3bee47
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/diff/5c3bee47
Branch: refs/heads/ARIA-26-plugin-mechanism
Commit: 5c3bee47d6acbb46235c07b0ff5b5be2460ccb1e
Parents: 04c9bd0
Author: Dan Kilman <da...@gigaspaces.com>
Authored: Sun Nov 27 16:31:29 2016 +0200
Committer: Dan Kilman <da...@gigaspaces.com>
Committed: Tue Dec 13 12:32:04 2016 +0200
----------------------------------------------------------------------
aria/cli/commands.py | 4 +-
aria/orchestrator/workflows/api/task.py | 42 +++++---
aria/orchestrator/workflows/core/task.py | 10 +-
.../orchestrator/workflows/executor/__init__.py | 2 +-
.../orchestrator/workflows/executor/blocking.py | 36 -------
.../workflows/executor/multiprocess.py | 86 ++++++++++------
aria/orchestrator/workflows/executor/thread.py | 3 +-
aria/storage/models.py | 5 +
aria/utils/plugin.py | 72 +++++++++++++
requirements.txt | 3 +-
.../workflows/executor/test_executor.py | 103 +++++++++----------
11 files changed, 225 insertions(+), 141 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/5c3bee47/aria/cli/commands.py
----------------------------------------------------------------------
diff --git a/aria/cli/commands.py b/aria/cli/commands.py
index 141da07..825be68 100644
--- a/aria/cli/commands.py
+++ b/aria/cli/commands.py
@@ -34,7 +34,7 @@ from ..logger import LoggerMixin
from ..storage import (FileSystemModelDriver, FileSystemResourceDriver)
from ..orchestrator.context.workflow import WorkflowContext
from ..orchestrator.workflows.core.engine import Engine
-from ..orchestrator.workflows.executor.thread import ThreadExecutor
+from ..orchestrator.workflows.executor.multiprocess import MultiprocessExecutor
from ..parser import iter_specifications
from ..parser.consumption import (
ConsumptionContext,
@@ -252,7 +252,7 @@ class ExecuteCommand(BaseCommand):
)
workflow_function = self._load_workflow_handler(workflow['operation'])
tasks_graph = workflow_function(workflow_context, **workflow_context.parameters)
- executor = ThreadExecutor()
+ executor = MultiprocessExecutor()
workflow_engine = Engine(executor=executor,
workflow_context=workflow_context,
tasks_graph=tasks_graph)
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/5c3bee47/aria/orchestrator/workflows/api/task.py
----------------------------------------------------------------------
diff --git a/aria/orchestrator/workflows/api/task.py b/aria/orchestrator/workflows/api/task.py
index 1c12407..70f9773 100644
--- a/aria/orchestrator/workflows/api/task.py
+++ b/aria/orchestrator/workflows/api/task.py
@@ -67,11 +67,11 @@ class OperationTask(BaseTask):
max_attempts=None,
retry_interval=None,
ignore_failure=None,
- inputs=None):
+ inputs=None,
+ plugin=None):
"""
Creates an operation task using the name, details, node instance and any additional kwargs.
:param name: the operation of the name.
- :param operation_details: the details for the operation.
:param actor: the operation host on which this operation is registered.
:param inputs: operation inputs.
"""
@@ -82,6 +82,7 @@ class OperationTask(BaseTask):
self.name = '{name}.{actor.id}'.format(name=name, actor=actor)
self.operation_mapping = operation_mapping
self.inputs = inputs or {}
+ self.plugin = plugin or {}
self.max_attempts = (self.workflow_context._task_max_attempts
if max_attempts is None else max_attempts)
self.retry_interval = (self.workflow_context._task_retry_interval
@@ -98,15 +99,13 @@ class OperationTask(BaseTask):
:param name: the name of the operation.
"""
assert isinstance(instance, models.NodeInstance)
- operation_details = instance.node.operations[name]
- operation_inputs = operation_details.get('inputs', {})
- operation_inputs.update(inputs or {})
- return cls(name=name,
- actor=instance,
- operation_mapping=operation_details.get('operation', ''),
- inputs=operation_inputs,
- *args,
- **kwargs)
+ return cls._instance(instance=instance,
+ name=name,
+ operation_details=instance.node.operations[name],
+ inputs=inputs,
+ plugins=instance.node.plugins or [],
+ *args,
+ **kwargs)
@classmethod
def relationship_instance(cls, instance, name, operation_end, inputs=None, *args, **kwargs):
@@ -125,12 +124,31 @@ class OperationTask(BaseTask):
cls.TARGET_OPERATION, cls.SOURCE_OPERATION
))
operation_details = getattr(instance.relationship, operation_end)[name]
+ if operation_end == cls.SOURCE_OPERATION:
+ plugins = instance.relationship.source_node.plugins
+ else:
+ plugins = instance.relationship.target_node.plugins
+ return cls._instance(instance=instance,
+ name=name,
+ operation_details=operation_details,
+ inputs=inputs,
+ plugins=plugins or [],
+ *args,
+ **kwargs)
+
+ @classmethod
+ def _instance(cls, instance, name, operation_details, inputs, plugins, *args, **kwargs):
+ operation_mapping = operation_details.get('operation')
operation_inputs = operation_details.get('inputs', {})
operation_inputs.update(inputs or {})
+ plugin_name = operation_details.get('plugin')
+ matching_plugins = [p for p in plugins if p['name'] == plugin_name]
+ plugin = matching_plugins[0] if matching_plugins else {}
return cls(actor=instance,
name=name,
- operation_mapping=operation_details.get('operation'),
+ operation_mapping=operation_mapping,
inputs=operation_inputs,
+ plugin=plugin,
*args,
**kwargs)
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/5c3bee47/aria/orchestrator/workflows/core/task.py
----------------------------------------------------------------------
diff --git a/aria/orchestrator/workflows/core/task.py b/aria/orchestrator/workflows/core/task.py
index 0be17fe..3d869ba 100644
--- a/aria/orchestrator/workflows/core/task.py
+++ b/aria/orchestrator/workflows/core/task.py
@@ -106,8 +106,9 @@ class OperationTask(BaseTask):
def __init__(self, api_task, *args, **kwargs):
super(OperationTask, self).__init__(id=api_task.id, **kwargs)
self._workflow_context = api_task._workflow_context
- base_task_model = api_task._workflow_context.model.task.model_cls
+ model = api_task._workflow_context.model
+ base_task_model = model.task.model_cls
if isinstance(api_task.actor, models.NodeInstance):
context_class = operation_context.NodeOperationContext
task_model_cls = base_task_model.as_node_instance
@@ -117,7 +118,11 @@ class OperationTask(BaseTask):
else:
raise RuntimeError('No operation context could be created for {actor.model_cls}'
.format(actor=api_task.actor))
-
+ plugin = api_task.plugin
+ plugins = model.plugin.list(filters={'package_name': plugin.get('package_name', ''),
+ 'package_version': plugin.get('package_version', '')},
+ include=['id'])
+ plugin_id = plugins[0].id if plugins else None
operation_task = task_model_cls(
name=api_task.name,
operation_mapping=api_task.operation_mapping,
@@ -127,6 +132,7 @@ class OperationTask(BaseTask):
max_attempts=api_task.max_attempts,
retry_interval=api_task.retry_interval,
ignore_failure=api_task.ignore_failure,
+ plugin_id=plugin_id
)
self._workflow_context.model.task.put(operation_task)
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/5c3bee47/aria/orchestrator/workflows/executor/__init__.py
----------------------------------------------------------------------
diff --git a/aria/orchestrator/workflows/executor/__init__.py b/aria/orchestrator/workflows/executor/__init__.py
index 16b6c9b..7b205c1 100644
--- a/aria/orchestrator/workflows/executor/__init__.py
+++ b/aria/orchestrator/workflows/executor/__init__.py
@@ -18,5 +18,5 @@ Executors for task execution
"""
-from . import blocking, multiprocess, thread
+from . import multiprocess, thread
from .base import BaseExecutor
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/5c3bee47/aria/orchestrator/workflows/executor/blocking.py
----------------------------------------------------------------------
diff --git a/aria/orchestrator/workflows/executor/blocking.py b/aria/orchestrator/workflows/executor/blocking.py
deleted file mode 100644
index 9d3a9ba..0000000
--- a/aria/orchestrator/workflows/executor/blocking.py
+++ /dev/null
@@ -1,36 +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.
-
-"""
-Blocking executor
-"""
-
-from aria.utils import imports
-from .base import BaseExecutor
-
-
-class CurrentThreadBlockingExecutor(BaseExecutor):
- """
- Executor which runs tasks in the current thread (blocking)
- """
-
- def execute(self, task):
- self._task_started(task)
- try:
- task_func = imports.load_attribute(task.operation_mapping)
- task_func(ctx=task.context, **task.inputs)
- self._task_succeeded(task)
- except BaseException as e:
- self._task_failed(task, exception=e)
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/5c3bee47/aria/orchestrator/workflows/executor/multiprocess.py
----------------------------------------------------------------------
diff --git a/aria/orchestrator/workflows/executor/multiprocess.py b/aria/orchestrator/workflows/executor/multiprocess.py
index d770e07..7fbf7e9 100644
--- a/aria/orchestrator/workflows/executor/multiprocess.py
+++ b/aria/orchestrator/workflows/executor/multiprocess.py
@@ -17,7 +17,10 @@
Multiprocess based executor
"""
+import collections
import multiprocessing
+import os
+import sys
import threading
import jsonpickle
@@ -26,73 +29,90 @@ from aria.utils import imports
from .base import BaseExecutor
+_TaskStarted = collections.namedtuple('_TaskStarted', 'task_id')
+_TaskSucceeded = collections.namedtuple('_TaskSucceeded', 'task_id')
+_TaskFailed = collections.namedtuple('_TaskFailed', 'task_id exception')
+
+
class MultiprocessExecutor(BaseExecutor):
"""
Executor which runs tasks in a multiprocess environment
"""
- def __init__(self, pool_size=1, *args, **kwargs):
+ def __init__(self, pool_size=1, plugin_manager=None, *args, **kwargs):
super(MultiprocessExecutor, self).__init__(*args, **kwargs)
self._stopped = False
- self._manager = multiprocessing.Manager()
+ self._plugin_manager = plugin_manager
+ self._multiprocessing_manager = multiprocessing.Manager()
self._queue = self._manager.Queue()
self._tasks = {}
self._listener_thread = threading.Thread(target=self._listener)
self._listener_thread.daemon = True
self._listener_thread.start()
- self._pool = multiprocessing.Pool(processes=pool_size)
-
- def execute(self, task):
- self._tasks[task.id] = task
- self._pool.apply_async(_multiprocess_handler, args=(
- self._queue,
- task.context,
- task.id,
- task.operation_mapping,
- task.inputs))
+ self._pool = multiprocessing.Pool(processes=pool_size, maxtasksperchild=1)
def close(self):
self._pool.close()
self._stopped = True
self._pool.join()
+ self._multiprocessing_manager.shutdown()
+ self._multiprocessing_manager.join()
self._listener_thread.join()
+ def execute(self, task):
+ self._tasks[task.id] = task
+ kwargs = {
+ 'queue': self._queue,
+ 'ctx': task.context,
+ 'task_id': task.id,
+ 'operation_mapping': task.operation_mapping,
+ 'operation_inputs': task.inputs,
+ 'plugin_prefix': None
+ }
+ if task.plugin_id and self._plugin_manager:
+ kwargs['plugin_prefix'] = self._plugin_manager.get_plugin_prefix(task.plugin)
+ self._pool.apply_async(_handler, kwds=kwargs)
+
+ def _remove_task(self, task_id):
+ return self._tasks.pop(task_id)
+
def _listener(self):
while not self._stopped:
try:
message = self._queue.get(timeout=1)
- if message.type == 'task_started':
+ if isinstance(message, _TaskStarted):
self._task_started(self._tasks[message.task_id])
- elif message.type == 'task_succeeded':
+ elif isinstance(message, _TaskSucceeded):
self._task_succeeded(self._remove_task(message.task_id))
- elif message.type == 'task_failed':
+ elif isinstance(message, _TaskFailed):
self._task_failed(self._remove_task(message.task_id),
exception=jsonpickle.loads(message.exception))
else:
- # TODO: something
- raise RuntimeError()
+ raise RuntimeError('Invalid state')
# Daemon threads
except BaseException:
pass
- def _remove_task(self, task_id):
- return self._tasks.pop(task_id)
-
-
-class _MultiprocessMessage(object):
-
- def __init__(self, type, task_id, exception=None):
- self.type = type
- self.task_id = task_id
- self.exception = exception
-
-def _multiprocess_handler(queue, ctx, task_id, operation_mapping, operation_inputs):
- queue.put(_MultiprocessMessage(type='task_started', task_id=task_id))
+def _handler(queue, ctx, task_id, operation_mapping, operation_inputs, plugin_prefix):
+ queue.put(_TaskStarted(task_id))
+ if plugin_prefix:
+ bin_dir = 'Scripts' if os.name == 'nt' else 'bin'
+ os.environ['PATH'] = '{0}{1}{2}'.format(
+ os.path.join(plugin_prefix, bin_dir),
+ os.pathsep,
+ os.environ.get('PATH', ''))
+ if os.name == 'nt':
+ pythonpath_dirs = [os.path.join(plugin_prefix, 'Lib', 'site-packages')]
+ else:
+ 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']]
+ sys.path.extend(pythonpath_dirs)
try:
task_func = imports.load_attribute(operation_mapping)
task_func(ctx=ctx, **operation_inputs)
- queue.put(_MultiprocessMessage(type='task_succeeded', task_id=task_id))
+ queue.put(_TaskSucceeded(task_id))
except BaseException as e:
- queue.put(_MultiprocessMessage(type='task_failed', task_id=task_id,
- exception=jsonpickle.dumps(e)))
+ queue.put(_TaskFailed(task_id, exception=jsonpickle.dumps(e)))
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/5c3bee47/aria/orchestrator/workflows/executor/thread.py
----------------------------------------------------------------------
diff --git a/aria/orchestrator/workflows/executor/thread.py b/aria/orchestrator/workflows/executor/thread.py
index 76ceefd..16a58fb 100644
--- a/aria/orchestrator/workflows/executor/thread.py
+++ b/aria/orchestrator/workflows/executor/thread.py
@@ -26,7 +26,8 @@ from .base import BaseExecutor
class ThreadExecutor(BaseExecutor):
"""
- Executor which runs tasks in a separate thread
+ Executor which runs tasks in a separate thread. It's easier writing tests
+ using this executor rather than the full blown multiprocessing executor.
"""
def __init__(self, pool_size=1, *args, **kwargs):
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/5c3bee47/aria/storage/models.py
----------------------------------------------------------------------
diff --git a/aria/storage/models.py b/aria/storage/models.py
index 6302e66..d467ce9 100644
--- a/aria/storage/models.py
+++ b/aria/storage/models.py
@@ -550,6 +550,11 @@ class Task(SQLModelBase):
name = Column(String)
operation_mapping = Column(String)
inputs = Column(Dict)
+ plugin_id = foreign_key(Plugin.id, nullable=True)
+
+ @declared_attr
+ def plugin(cls):
+ return one_to_many_relationship(cls, Plugin, cls.plugin_id)
@declared_attr
def execution(cls):
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/5c3bee47/aria/utils/plugin.py
----------------------------------------------------------------------
diff --git a/aria/utils/plugin.py b/aria/utils/plugin.py
new file mode 100644
index 0000000..944ba62
--- /dev/null
+++ b/aria/utils/plugin.py
@@ -0,0 +1,72 @@
+# 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
+import tempfile
+import subprocess
+import sys
+from datetime import datetime
+
+import wagon
+
+
+def create(source, destination_dir):
+ return wagon.create(source=source, archive_destination_dir=destination_dir)
+
+
+class PluginManager(object):
+
+ def __init(self, model, plugins_dir):
+ self._model = model
+ self._plugins_dir = plugins_dir
+
+ def install(self, source):
+ metadata = wagon.show(source)
+ cls = self._model.plugin.model_cls
+ plugin = cls(
+ archive_name=metadata['archive_name'],
+ supported_platform=metadata['supported_platform'],
+ supported_py_versions=metadata['supported_python_versions'],
+ 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'],
+ package_name=metadata['package_name'],
+ package_version=metadata['package_version'],
+ package_source=metadata['package_source'],
+ wheels=metadata['wheels'],
+ uploaded_at=datetime.now()
+ )
+ if len(self._model.plugin.list(filters={'package_name': plugin.package_name,
+ 'package_version': plugin.package_version})):
+ raise RuntimeError
+
+ self._install_wagon(source=source, prefix=self.get_plugin_prefix(plugin))
+ self._model.plugin.store(plugin)
+ return plugin
+
+ def get_plugin_prefix(self, plugin):
+ return os.path.join(
+ self._plugins_dir,
+ '{0}-{1}'.format(plugin.package_name, plugin.package_version))
+
+ def _install_wagon(source, prefix):
+ with tempfile.NamedTemporaryFile() as constraint:
+ constraint.write(subprocess.check_output([sys.executable, '-m', 'pip', 'freeze']))
+ constraint.flush()
+ wagon.install(
+ source=source,
+ install_args='--prefix="{prefix}" --constraint="{constraint}"'.format(
+ prefix=prefix,
+ constraint=constraint.name))
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/5c3bee47/requirements.txt
----------------------------------------------------------------------
diff --git a/requirements.txt b/requirements.txt
index 31b0b79..0005a5e 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -24,4 +24,5 @@ Jinja2==2.8
shortuuid==0.4.3
CacheControl[filecache]==0.11.6
clint==0.5.1
-SQLAlchemy==1.1.4
\ No newline at end of file
+SQLAlchemy==1.1.4
+wagon==0.5.0
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/5c3bee47/tests/orchestrator/workflows/executor/test_executor.py
----------------------------------------------------------------------
diff --git a/tests/orchestrator/workflows/executor/test_executor.py b/tests/orchestrator/workflows/executor/test_executor.py
index 654542c..a76adca 100644
--- a/tests/orchestrator/workflows/executor/test_executor.py
+++ b/tests/orchestrator/workflows/executor/test_executor.py
@@ -33,52 +33,28 @@ from aria.orchestrator import events
from aria.orchestrator.workflows.executor import (
thread,
multiprocess,
- blocking,
# celery
)
-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 test_execute(executor):
+ 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]:
+ 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 mock_successful_task(**_):
@@ -128,14 +104,35 @@ class MockTask(object):
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
+@pytest.fixture(params=[
+ (thread.ThreadExecutor, {'pool_size': 1}),
+ (thread.ThreadExecutor, {'pool_size': 2}),
+ (multiprocess.MultiprocessExecutor, {'pool_size': 1}),
+ (multiprocess.MultiprocessExecutor, {'pool_size': 2}),
+ # (celery.CeleryExecutor, {'app': app})
+])
+def executor(request):
+ executor_cls, executor_kwargs = request.param
+ result = executor_cls(**executor_kwargs)
+ yield result
+ result.close()
+
+
+@pytest.fixture(autouse=True)
+def register_signals():
+ 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
+ events.start_task_signal.connect(start_handler)
+ events.on_success_task_signal.connect(success_handler)
+ events.on_failure_task_signal.connect(failure_handler)
+ yield
+ events.start_task_signal.disconnect(start_handler)
+ events.on_success_task_signal.disconnect(success_handler)
+ events.on_failure_task_signal.disconnect(failure_handler)