deb-murano/murano/engine/mock_context_manager.py

# Copyright (c) 2015 Mirantis Inc.
#
# 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 six
from yaql.language import specs
from yaql.language import yaqltypes

from murano.common import engine
from murano.dsl import constants
from murano.dsl import dsl
from murano.dsl import dsl_types
from murano.dsl import helpers
from murano.dsl import linked_context
from murano.dsl import yaql_integration


class MockContextManager(engine.ContextManager):

    def __init__(self):
        # { object_id: [mock_function_definitions] }
        self.obj_mock_ctx = {}

        # { class_name: [mock_function_definitions] }
        self.class_mock_ctx = {}

    def _create_new_ctx(self, mock_funcs):
        mock_context = yaql_integration.create_empty_context()
        for mock_func in mock_funcs:
            mock_context.register_function(mock_func)
        return mock_context

    def _create_new_ctx_for_class(self, name):
        new_context = None
        if name in self.class_mock_ctx:
            new_context = self._create_new_ctx(self.class_mock_ctx[name])
        return new_context

    def _create_new_ctx_for_obj(self, obj_id):
        new_context = None
        if obj_id in self.obj_mock_ctx:
            new_context = self._create_new_ctx(self.obj_mock_ctx[obj_id])
        return new_context

    def create_class_context(self, murano_class):
        original_context = super(MockContextManager,
                                 self).create_class_context(murano_class)

        mock_context = self._create_new_ctx_for_class(murano_class.name)
        if mock_context:
            result_context = linked_context.link(
                original_context, mock_context).create_child_context()
        else:
            result_context = original_context
        return result_context

    def create_object_context(self, murano_obj):
        original_context = super(MockContextManager,
                                 self).create_object_context(murano_obj)

        mock_context = self._create_new_ctx_for_obj(murano_obj.type.name)
        if mock_context:
            result_context = linked_context.link(
                original_context, mock_context).create_child_context()
        else:
            result_context = original_context
        return result_context

    def create_root_context(self, runtime_version):
        root_context = super(MockContextManager, self).create_root_context(
            runtime_version)
        constext = root_context.create_child_context()
        constext.register_function(inject_method_with_str, name='inject')
        constext.register_function(inject_method_with_yaql_expr,
                                   name='inject')
        constext.register_function(with_original)
        return constext


@specs.parameter('kwargs', yaqltypes.Lambda(with_context=True))
def with_original(context, **kwargs):
    new_context = context.create_child_context()

    original_context = context[constants.CTX_ORIGINAL_CONTEXT]
    for k, v in six.iteritems(kwargs):
        new_context['$' + k] = v(original_context)
    return new_context


@specs.parameter('target',
                 dsl.OneOf(dsl.MuranoTypeName(), dsl_types.MuranoObject))
@specs.parameter('target_method', yaqltypes.String())
@specs.parameter('mock_object', dsl_types.MuranoObject)
@specs.parameter('mock_name', yaqltypes.String())
def inject_method_with_str(context, target, target_method,
                           mock_object, mock_name):
    ctx_manager = helpers.get_executor(context).context_manager

    current_class = helpers.get_type(context)
    mock_func = current_class.find_single_method(mock_name)

    if isinstance(target, dsl_types.MuranoClassReference):
        original_class = target.murano_class
    else:
        original_class = target.type

    original_function = original_class.find_single_method(target_method)
    result_fd = original_function.yaql_function_definition.clone()

    def payload_adapter(__context, __sender, *args, **kwargs):
        executor = helpers.get_executor(__context)
        return mock_func.invoke(
            executor, mock_object, args, kwargs, __context, True)

    result_fd.payload = payload_adapter
    existing_mocks = ctx_manager.class_mock_ctx.setdefault(
        original_class.name, [])
    existing_mocks.append(result_fd)


@specs.parameter('target',
                 dsl.OneOf(dsl.MuranoTypeName(), dsl_types.MuranoObject))
@specs.parameter('target_method', yaqltypes.String())
@specs.parameter('expr', yaqltypes.Lambda(with_context=True))
def inject_method_with_yaql_expr(context, target, target_method, expr):
    ctx_manager = helpers.get_executor(context).context_manager
    if isinstance(target, dsl_types.MuranoClassReference):
        original_class = target.murano_class
    else:
        original_class = target.type

    original_function = original_class.find_single_method(target_method)
    result_fd = original_function.yaql_function_definition.clone()

    def payload_adapter(__super, __context, __sender, *args, **kwargs):
        new_context = context.create_child_context()
        new_context[constants.CTX_ORIGINAL_CONTEXT] = __context
        mock_obj = context[constants.CTX_THIS]
        new_context.register_function(lambda: __super(*args, **kwargs),
                                      name='originalMethod')
        return expr(new_context, mock_obj, *args, **kwargs)

    result_fd.payload = payload_adapter
    result_fd.insert_parameter('__super', yaqltypes.Super())
    existing_mocks = ctx_manager.class_mock_ctx.setdefault(
        original_class.name, [])
    existing_mocks.append(result_fd)