deb-murano/murano/engine/mock_context_manager.py
Stan Lagun 0a835b87f0 Refactoring of the ObjectStore passing in DSL
ObjectStore is a container for all of the objects that are either
loaded from object model or created at runtime. Thus ObjectStore
instance is required in order to either use new() function or use
class() contract with object model snippet.

In current implementation each object had a object_store property
that was pointing to the ObjectStore instance used to create it (the
one that has the object). In order to create new object an ObjectStore
of the $this object was used for that purpose. This approach caused
several issues:
* There is no $this object in static methods and thus no way to obtain
   current ObjectStore
* Objects are bind to their stores so temporary ObjectStore instances
  cannot be used to load object graphs

 This refactoring moves ObjectStore instances to the thread-local attribute.
 Thus there is always a "current" ObjectStore.

 Also because ObjectStore has a reference to the MuranoDslExecutor instance
 it becomes possible to get current executor and everything else that it contains
 (current execution session, attribute store, package loader) from the local
 thread storage without passing it in the hidden context variables.

Change-Id: I19a08b391486c1ae1b98ca559c8d3fafaca6e1ef
2016-07-27 02:32:18 +00:00

151 lines
5.6 KiB
Python

# 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 helpers
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_type_context(self, murano_type):
original_context = super(
MockContextManager, self).create_type_context(murano_type)
mock_context = self._create_new_ctx_for_class(murano_type.name)
if mock_context:
result_context = helpers.link_contexts(
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 = helpers.link_contexts(
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',
yaqltypes.AnyOf(dsl.MuranoTypeParameter(), dsl.MuranoObjectParameter()))
@specs.parameter('target_method', yaqltypes.String())
@specs.parameter('mock_object', dsl.MuranoObjectParameter())
@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_manager
current_class = helpers.get_type(context)
mock_func = current_class.find_single_method(mock_name)
original_class = target.type
original_function = original_class.find_single_method(target_method)
result_fd = original_function.instance_stub.clone()
def payload_adapter(__context, __sender, *args, **kwargs):
return mock_func.invoke(
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',
yaqltypes.AnyOf(dsl.MuranoTypeParameter(), dsl.MuranoObjectParameter()))
@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_manager
original_class = target.type
original_function = original_class.find_single_method(target_method)
result_fd = original_function.instance_stub.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)