--- /dev/null
+#
+# Copyright (c) 2017 GigaSpaces Technologies Ltd. All rights reserved.
+#
+# Licensed 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 copy
+import datetime
+import contextlib
+
+import pytest
+
+from aria import (workflow, operation)
+from aria.modeling import models
+from aria.orchestrator import events
+from aria.orchestrator.workflows import api
+from aria.orchestrator.workflows.exceptions import ExecutorException
+from aria.orchestrator.workflows.executor import process
+from aria.orchestrator.workflows.core import (engine, graph_compiler)
+from aria.orchestrator.exceptions import (TaskAbortException, TaskRetryException)
+from aria.utils import type as type_
+
+import tests
+from tests import (mock, storage, conftest)
+from tests.orchestrator.workflows.helpers import events_collector
+
+from adapters import context_adapter
+
+
+@pytest.fixture(autouse=True)
+def cleanup_logger(request):
+ conftest.logging_handler_cleanup(request)
+
+
+class TestCloudifyContextAdapter(object):
+
+ def test_node_instance_operation(self, executor, workflow_context):
+ node_template = self._get_node_template(workflow_context)
+ node_type = 'aria.plugin.nodes.App'
+ node_instance_property = models.Property.wrap('hello', 'world')
+ node_template.type = models.Type(variant='variant', name=node_type)
+ node = self._get_node(workflow_context)
+ node_instance_attribute = models.Attribute.wrap('hello2', 'world2')
+ node.attributes[node_instance_attribute.name] = node_instance_attribute
+ node.properties[node_instance_property.name] = node_instance_property
+ workflow_context.model.node.update(node)
+ workflow_context.model.node_template.update(node_template)
+
+ out = self._run(executor, workflow_context, _test_node_instance_operation)
+
+ node_template = self._get_node_template(workflow_context)
+ node = self._get_node(workflow_context)
+ assert out['type'] == context_adapter.NODE_INSTANCE
+ assert out['node']['id'] == node_template.id
+ assert out['node']['name'] == node_template.name
+ assert out['node']['properties'] == \
+ {node_instance_property.name: node_instance_property.value}
+ assert out['node']['type'] == node_type
+ assert out['node']['type_hierarchy'] == ['cloudify.plugin.nodes.App']
+ assert out['instance']['id'] == node.id
+ assert out['instance']['runtime_properties'] == \
+ {node_instance_attribute.name: node_instance_attribute.value}
+ assert not out['source']
+ assert not out['target']
+
+ def test_node_instance_relationships(self, executor, workflow_context):
+ relationship_node_template = self._get_dependency_node_template(workflow_context)
+ relationship_node_instance = self._get_dependency_node(workflow_context)
+ relationship = relationship_node_instance.inbound_relationships[0]
+ relationship_type = models.Type(variant='variant', name='test.relationships.Relationship')
+ relationship.type = relationship_type
+ workflow_context.model.relationship.update(relationship)
+
+ out = self._run(executor, workflow_context, _test_node_instance_relationships)
+
+ assert len(out['instance']['relationships']) == 1
+ relationship = out['instance']['relationships'][0]
+ assert relationship['type'] == relationship_type.name
+ assert relationship['type_hierarchy'] == [relationship_type.name]
+ assert relationship['target']['node']['id'] == relationship_node_template.id
+ assert relationship['target']['instance']['id'] == relationship_node_instance.id
+
+ def test_source_operation(self, executor, workflow_context):
+ self._test_relationship_operation(executor, workflow_context, operation_end='source')
+
+ def test_target_operation(self, executor, workflow_context):
+ self._test_relationship_operation(executor, workflow_context, operation_end='target')
+
+ def _test_relationship_operation(self, executor, workflow_context, operation_end):
+ out = self._run(
+ executor, workflow_context, _test_relationship_operation, operation_end=operation_end)
+
+ source_node = self._get_node_template(workflow_context)
+ source_node_instance = self._get_node(workflow_context)
+ target_node = self._get_dependency_node_template(workflow_context)
+ target_node_instance = self._get_dependency_node(workflow_context)
+ assert out['type'] == context_adapter.RELATIONSHIP_INSTANCE
+ assert out['source']['node']['id'] == source_node.id
+ assert out['source']['instance']['id'] == source_node_instance.id
+ assert out['target']['node']['id'] == target_node.id
+ assert out['target']['instance']['id'] == target_node_instance.id
+ assert not out['node']
+ assert not out['instance']
+
+ def test_host_ip(self, executor, workflow_context):
+ node = self._get_node_template(workflow_context)
+ node.type_hierarchy = ['aria.nodes.Compute']
+ node_instance = self._get_node(workflow_context)
+ node_instance.host_fk = node_instance.id
+ node_instance_ip = '120.120.120.120'
+ node_instance.attributes['ip'] = models.Attribute.wrap('ip', node_instance_ip)
+ workflow_context.model.node_template.update(node)
+ workflow_context.model.node.update(node_instance)
+
+ out = self._run(executor, workflow_context, _test_host_ip)
+
+ assert out['instance']['host_ip'] == node_instance_ip
+
+ def test_get_and_download_resource_and_render(self, tmpdir, executor, workflow_context):
+ resource_path = 'resource'
+ variable = 'VALUE'
+ content = '{{ctx.service.name}}-{{variable}}'
+ rendered = '{0}-{1}'.format(workflow_context.service.name, variable)
+ source = tmpdir.join(resource_path)
+ source.write(content)
+ workflow_context.resource.service.upload(
+ entry_id=str(workflow_context.service.id),
+ source=str(source),
+ path=resource_path)
+
+ out = self._run(executor, workflow_context, _test_get_and_download_resource_and_render,
+ inputs={'resource': resource_path,
+ 'variable': variable})
+
+ assert out['get_resource'] == content
+ assert out['get_resource_and_render'] == rendered
+ with open(out['download_resource'], 'rb') as f:
+ assert f.read() == content
+ with open(out['download_resource_and_render'], 'rb') as f:
+ assert f.read() == rendered
+
+ os.remove(out['download_resource'])
+ os.remove(out['download_resource_and_render'])
+
+ def test_retry(self, executor, workflow_context):
+ message = 'retry-message'
+ retry_interval = 0.01
+
+ exception = self._run_and_get_task_exceptions(
+ executor, workflow_context, _test_retry,
+ inputs={'message': message, 'retry_interval': retry_interval},
+ max_attempts=2
+ )[-1]
+
+ assert isinstance(exception, TaskRetryException)
+ assert exception.message == message
+ assert exception.retry_interval == retry_interval
+
+ out = self._get_node(workflow_context).attributes['out'].value
+ assert out['operation']['retry_number'] == 1
+ assert out['operation']['max_retries'] == 1
+
+ def test_logger_and_send_event(self, executor, workflow_context):
+ # TODO: add assertions of output once process executor output can be captured
+ message = 'logger-message'
+ event = 'event-message'
+ self._run(executor, workflow_context, _test_logger_and_send_event,
+ inputs={'message': message, 'event': event})
+
+ def test_plugin(self, executor, workflow_context, tmpdir):
+ plugin = self._put_plugin(workflow_context)
+ out = self._run(executor, workflow_context, _test_plugin, plugin=plugin)
+
+ expected_workdir = tmpdir.join(
+ 'workdir', 'plugins', str(workflow_context.service.id), plugin.name)
+ assert out['plugin']['name'] == plugin.name
+ assert out['plugin']['package_name'] == plugin.package_name
+ assert out['plugin']['package_version'] == plugin.package_version
+ assert out['plugin']['workdir'] == str(expected_workdir)
+
+ def test_importable_ctx_and_inputs(self, executor, workflow_context):
+ test_inputs = {'input1': 1, 'input2': 2}
+ plugin = self._put_plugin(workflow_context, mock_cfy_plugin=True)
+
+ out = self._run(executor, workflow_context, _test_importable_ctx_and_inputs,
+ inputs=test_inputs,
+ skip_common_assert=True,
+ plugin=plugin)
+ assert out['inputs'] == test_inputs
+
+ def test_non_recoverable_error(self, executor, workflow_context):
+ message = 'NON_RECOVERABLE_MESSAGE'
+ plugin = self._put_plugin(workflow_context, mock_cfy_plugin=True)
+
+ exception = self._run_and_get_task_exceptions(
+ executor, workflow_context, _test_non_recoverable_error,
+ inputs={'message': message},
+ skip_common_assert=True,
+ plugin=plugin
+ )[0]
+ assert isinstance(exception, TaskAbortException)
+ assert exception.message == message
+
+ def test_recoverable_error(self, executor, workflow_context):
+ message = 'RECOVERABLE_MESSAGE'
+ plugin = self._put_plugin(workflow_context, mock_cfy_plugin=True)
+
+ retry_interval = 0.01
+ exception = self._run_and_get_task_exceptions(
+ executor, workflow_context, _test_recoverable_error,
+ inputs={'message': message, 'retry_interval': retry_interval},
+ skip_common_assert=True,
+ plugin=plugin
+ )[0]
+ assert isinstance(exception, TaskRetryException)
+ assert message in exception.message
+ assert exception.retry_interval == retry_interval
+
+ def _test_common(self, out, workflow_context):
+ assert out['execution_id'] == workflow_context.execution.id
+ assert out['workflow_id'] == workflow_context.execution.workflow_name
+ assert out['rest_token'] is None
+ assert out['task_id'][0] == out['task_id'][1]
+ assert out['task_name'][0] == out['task_name'][1]
+ assert out['task_target'] is None
+ assert out['task_queue'] is None
+ assert out['provider_context'] == {}
+ assert out['blueprint']['id'] == workflow_context.service_template.id
+ assert out['deployment']['id'] == workflow_context.service.id
+ assert out['operation']['name'][0] == out['operation']['name'][1]
+ assert out['operation']['retry_number'][0] == out['operation']['retry_number'][1]
+ assert out['operation']['max_retries'][0] == out['operation']['max_retries'][1] - 1
+ assert out['bootstrap_context']['resources_prefix'] == ''
+ assert out['bootstrap_context']['broker_config'] == {}
+ assert out['bootstrap_context']['cloudify_agent']['any'] is None
+ assert out['agent']['init_script'] is None
+
+ def _run(self,
+ executor,
+ workflow_context,
+ func,
+ inputs=None,
+ max_attempts=None,
+ skip_common_assert=False,
+ operation_end=None,
+ plugin=None):
+ interface_name = 'test'
+ operation_name = 'op'
+ op_dict = {'function': '{0}.{1}'.format(__name__, func.__name__),
+ 'plugin': plugin,
+ 'arguments': inputs or {}}
+ node = self._get_node(workflow_context)
+
+ if operation_end:
+ actor = relationship = node.outbound_relationships[0]
+ relationship.interfaces[interface_name] = mock.models.create_interface(
+ relationship.source_node.service,
+ interface_name,
+ operation_name,
+ operation_kwargs=op_dict
+ )
+ workflow_context.model.relationship.update(relationship)
+
+ else:
+ actor = node
+ node.interfaces[interface_name] = mock.models.create_interface(
+ node.service,
+ interface_name,
+ operation_name,
+ operation_kwargs=op_dict
+ )
+ workflow_context.model.node.update(node)
+
+ if inputs:
+ operation_inputs = \
+ actor.interfaces[interface_name].operations[operation_name].inputs
+ for input_name, input in inputs.iteritems():
+ operation_inputs[input_name] = models.Input(name=input_name,
+ type_name=type_.full_type_name(input))
+
+ @workflow
+ def mock_workflow(graph, **kwargs):
+ task = api.task.OperationTask(
+ actor,
+ interface_name,
+ operation_name,
+ arguments=inputs or {},
+ max_attempts=max_attempts
+ )
+ graph.add_tasks(task)
+
+ tasks_graph = mock_workflow(ctx=workflow_context)
+ graph_compiler.GraphCompiler(workflow_context, executor.__class__).compile(tasks_graph)
+ eng = engine.Engine(executors={executor.__class__: executor})
+ eng.execute(workflow_context)
+ out = self._get_node(workflow_context).attributes['out'].value
+ if not skip_common_assert:
+ self._test_common(out, workflow_context)
+ return out
+
+ def _get_dependency_node_template(self, workflow_context):
+ return workflow_context.model.node_template.get_by_name(
+ mock.models.DEPENDENCY_NODE_TEMPLATE_NAME)
+
+ def _get_dependency_node(self, workflow_context):
+ return workflow_context.model.node.get_by_name(mock.models.DEPENDENCY_NODE_NAME)
+
+ def _get_node_template(self, workflow_context):
+ return workflow_context.model.node_template.get_by_name(
+ mock.models.DEPENDENT_NODE_TEMPLATE_NAME)
+
+ def _get_node(self, workflow_context):
+ return workflow_context.model.node.get_by_name(mock.models.DEPENDENT_NODE_NAME)
+
+ def _run_and_get_task_exceptions(self, *args, **kwargs):
+ signal = events.on_failure_task_signal
+ with events_collector(signal) as collected:
+ with pytest.raises(ExecutorException):
+ self._run(*args, **kwargs)
+ return [event['kwargs']['exception'] for event in collected[signal]]
+
+ @pytest.fixture
+ def executor(self):
+ result = process.ProcessExecutor(python_path=[tests.ROOT_DIR])
+ yield result
+ result.close()
+
+ @pytest.fixture
+ def workflow_context(self, tmpdir):
+ result = mock.context.simple(
+ str(tmpdir),
+ context_kwargs=dict(workdir=str(tmpdir.join('workdir')))
+ )
+ yield result
+ storage.release_sqlite_storage(result.model)
+
+ def _put_plugin(self, workflow_context, mock_cfy_plugin=False):
+ name = 'PLUGIN'
+ archive_name = 'ARCHIVE'
+ package_name = 'PACKAGE'
+ package_version = '0.1.1'
+
+ plugin = models.Plugin(
+ name=name,
+ archive_name=archive_name,
+ package_name=package_name,
+ package_version=package_version,
+ uploaded_at=datetime.datetime.now(),
+ wheels=['cloudify_plugins_common'] if mock_cfy_plugin else []
+ )
+
+ workflow_context.model.plugin.put(plugin)
+
+ return plugin
+
+
+@operation
+def _test_node_instance_operation(ctx):
+ with _adapter(ctx) as (adapter, out):
+ node = adapter.node
+ instance = adapter.instance
+ out.update({
+ 'node': {
+ 'id': node.id,
+ 'name': node.name,
+ 'properties': copy.deepcopy(node.properties),
+ 'type': node.type,
+ 'type_hierarchy': node.type_hierarchy
+ },
+ 'instance': {
+ 'id': instance.id,
+ 'runtime_properties': copy.deepcopy(instance.runtime_properties)
+ }
+ })
+ try:
+ assert adapter.source
+ out['source'] = True
+ except TaskAbortException:
+ out['source'] = False
+ try:
+ assert adapter.target
+ out['target'] = True
+ except TaskAbortException:
+ out['target'] = False
+
+
+@operation
+def _test_node_instance_relationships(ctx):
+ with _adapter(ctx) as (adapter, out):
+ relationships = [{'type': r.type,
+ 'type_hierarchy': [t.name for t in r.type_hierarchy],
+ 'target': {'node': {'id': r.target.node.id},
+ 'instance': {'id': r.target.instance.id}}}
+ for r in adapter.instance.relationships]
+ out['instance'] = {'relationships': relationships}
+
+
+@operation
+def _test_relationship_operation(ctx):
+ with _adapter(ctx) as (adapter, out):
+ out.update({
+ 'source': {'node': {'id': adapter.source.node.id},
+ 'instance': {'id': adapter.source.instance.id}},
+ 'target': {'node': {'id': adapter.target.node.id},
+ 'instance': {'id': adapter.target.instance.id}}
+ })
+ try:
+ assert adapter.node
+ out['node'] = True
+ except TaskAbortException:
+ out['node'] = False
+ try:
+ assert adapter.instance
+ out['instance'] = True
+ except TaskAbortException:
+ out['instance'] = False
+
+
+@operation
+def _test_host_ip(ctx):
+ with _adapter(ctx) as (adapter, out):
+ out['instance'] = {'host_ip': adapter.instance.host_ip}
+
+
+@operation
+def _test_get_and_download_resource_and_render(ctx, resource, variable):
+ with _adapter(ctx) as (adapter, out):
+ out.update({
+ 'get_resource': adapter.get_resource(resource),
+ 'get_resource_and_render': adapter.get_resource_and_render(
+ resource, template_variables={'variable': variable}
+ ),
+ 'download_resource': adapter.download_resource(resource),
+ 'download_resource_and_render': adapter.download_resource_and_render(
+ resource, template_variables={'variable': variable}
+ )
+ })
+
+
+@operation
+def _test_retry(ctx, message, retry_interval):
+ with _adapter(ctx) as (adapter, out):
+ op = adapter.operation
+ out['operation'] = {'retry_number': op.retry_number, 'max_retries': op.max_retries}
+ op.retry(message, retry_after=retry_interval)
+
+
+@operation
+def _test_logger_and_send_event(ctx, message, event):
+ with _adapter(ctx) as (adapter, _):
+ adapter.logger.info(message)
+ adapter.send_event(event)
+
+
+@operation
+def _test_plugin(ctx):
+ with _adapter(ctx) as (adapter, out):
+ plugin = adapter.plugin
+ out['plugin'] = {
+ 'name': plugin.name,
+ 'package_name': plugin.package_name,
+ 'package_version': plugin.package_version,
+ 'workdir': plugin.workdir
+ }
+
+
+@operation
+def _test_importable_ctx_and_inputs(**_):
+ from cloudify import ctx
+ from cloudify.state import ctx_parameters
+ ctx.instance.runtime_properties['out'] = {'inputs': dict(ctx_parameters)}
+
+
+@operation
+def _test_non_recoverable_error(message, **_):
+ from cloudify.exceptions import NonRecoverableError
+ raise NonRecoverableError(message)
+
+
+@operation
+def _test_recoverable_error(message, retry_interval, **_):
+ from cloudify.exceptions import RecoverableError
+ raise RecoverableError(message, retry_interval)
+
+
+def _test_common(out, ctx, adapter):
+ op = adapter.operation
+ bootstrap_context = adapter.bootstrap_context
+ out.update({
+ 'type': adapter.type,
+ 'execution_id': adapter.execution_id,
+ 'workflow_id': adapter.workflow_id,
+ 'rest_token': adapter.rest_token,
+ 'task_id': (adapter.task_id, ctx.task.id),
+ 'task_name': (adapter.task_name, ctx.task.function),
+ 'task_target': adapter.task_target,
+ 'task_queue': adapter.task_queue,
+ 'provider_context': adapter.provider_context,
+ 'blueprint': {'id': adapter.blueprint.id},
+ 'deployment': {'id': adapter.deployment.id},
+ 'operation': {
+ 'name': [op.name, ctx.name.split('@')[0].replace(':', '.')],
+ 'retry_number': [op.retry_number, ctx.task.attempts_count - 1],
+ 'max_retries': [op.max_retries, ctx.task.max_attempts]
+ },
+ 'bootstrap_context': {
+ 'broker_config': bootstrap_context.broker_config('arg1', 'arg2', arg3='arg3'),
+ # All attribute access of cloudify_agent returns none
+ 'cloudify_agent': {'any': bootstrap_context.cloudify_agent.any},
+ 'resources_prefix': bootstrap_context.resources_prefix
+ },
+ 'agent': {
+ 'init_script': adapter.agent.init_script('arg1', 'arg2', arg3='arg3')
+ }
+ })
+
+
+@contextlib.contextmanager
+def _adapter(ctx):
+ out = {}
+ adapter = context_adapter.CloudifyContextAdapter(ctx)
+ _test_common(out, ctx, adapter)
+ try:
+ yield adapter, out
+ finally:
+ try:
+ instance = adapter.instance
+ except TaskAbortException:
+ instance = adapter.source.instance
+ instance.runtime_properties['out'] = out
Code Review / multicloud / azure.git / blobdiff