X-Git-Url: https://gerrit.onap.org/r/gitweb?a=blobdiff_plain;f=azure%2Faria%2Faria-extension-cloudify%2Fsrc%2Faria%2Ftests%2Fmodeling%2Ftest_mixins.py;fp=azure%2Faria%2Faria-extension-cloudify%2Fsrc%2Faria%2Ftests%2Fmodeling%2Ftest_mixins.py;h=2d94d7c250120bebba6b1ac9d2ebe63fa6ccbf92;hb=7409dfb144cf2a06210400134d822a1393462b1f;hp=0000000000000000000000000000000000000000;hpb=9e65649dfff8f00dc0a0ef6b10d020ae0e2255ba;p=multicloud%2Fazure.git diff --git a/azure/aria/aria-extension-cloudify/src/aria/tests/modeling/test_mixins.py b/azure/aria/aria-extension-cloudify/src/aria/tests/modeling/test_mixins.py new file mode 100644 index 0000000..2d94d7c --- /dev/null +++ b/azure/aria/aria-extension-cloudify/src/aria/tests/modeling/test_mixins.py @@ -0,0 +1,215 @@ +# 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 sqlalchemy + +from aria.storage import ( + ModelStorage, + sql_mapi +) +from aria import modeling +from aria.modeling.exceptions import ValueFormatException + +from ..storage import ( + release_sqlite_storage, + init_inmemory_model_storage +) +from . import MockModel +from ..mock import ( + models, + context as mock_context +) + + +@pytest.fixture +def storage(): + base_storage = ModelStorage(sql_mapi.SQLAlchemyModelAPI, + initiator=init_inmemory_model_storage) + base_storage.register(MockModel) + yield base_storage + release_sqlite_storage(base_storage) + + +@pytest.fixture(scope='module', autouse=True) +def module_cleanup(): + modeling.models.aria_declarative_base.metadata.remove(MockModel.__table__) # pylint: disable=no-member + + +@pytest.fixture +def context(tmpdir): + ctx = mock_context.simple(str(tmpdir)) + yield ctx + release_sqlite_storage(ctx.model) + + +def test_inner_dict_update(storage): + inner_dict = {'inner_value': 1} + + mock_model = MockModel(model_dict={'inner_dict': inner_dict, 'value': 0}) + storage.mock_model.put(mock_model) + + storage_mm = storage.mock_model.get(mock_model.id) + assert storage_mm == mock_model + + storage_mm.model_dict['inner_dict']['inner_value'] = 2 + storage_mm.model_dict['value'] = -1 + storage.mock_model.update(storage_mm) + storage_mm = storage.mock_model.get(storage_mm.id) + + assert storage_mm.model_dict['inner_dict']['inner_value'] == 2 + assert storage_mm.model_dict['value'] == -1 + + +def test_inner_list_update(storage): + mock_model = MockModel(model_list=[0, [1]]) + storage.mock_model.put(mock_model) + + storage_mm = storage.mock_model.get(mock_model.id) + assert storage_mm == mock_model + + storage_mm.model_list[1][0] = 'new_inner_value' + storage_mm.model_list[0] = 'new_value' + storage.mock_model.update(storage_mm) + storage_mm = storage.mock_model.get(storage_mm.id) + + assert storage_mm.model_list[1][0] == 'new_inner_value' + assert storage_mm.model_list[0] == 'new_value' + + +def test_model_to_dict(context): + service = context.service + service = service.to_dict() + + expected_keys = [ + 'description', + 'created_at', + 'updated_at' + ] + + for expected_key in expected_keys: + assert expected_key in service + + +def test_relationship_model_ordering(context): + service = context.model.service.get_by_name(models.SERVICE_NAME) + source_node = context.model.node.get_by_name(models.DEPENDENT_NODE_NAME) + target_node = context.model.node.get_by_name(models.DEPENDENCY_NODE_NAME) + + new_node_template = modeling.models.NodeTemplate( + name='new_node_template', + type=source_node.type, + service_template=service.service_template + ) + + new_node = modeling.models.Node( + name='new_node', + type=source_node.type, + service=service, + version=None, + node_template=new_node_template, + state=modeling.models.Node.INITIAL, + ) + + source_node.outbound_relationships.append(modeling.models.Relationship( + source_node=source_node, + target_node=new_node, + )) + + new_node.outbound_relationships.append(modeling.models.Relationship( # pylint: disable=no-member + source_node=new_node, + target_node=target_node, + )) + + context.model.node_template.put(new_node_template) + context.model.node.put(new_node) + context.model.node.refresh(source_node) + context.model.node.refresh(target_node) + + def flip_and_assert(node, direction): + """ + Reversed the order of relationships and assert effects took place. + :param node: the node instance to operate on + :param direction: the type of relationships to flip (inbound/outbound) + :return: + """ + assert direction in ('inbound', 'outbound') + + def get_relationships(): + return getattr(node, direction + '_relationships') + + relationships = get_relationships() + assert len(relationships) == 2 + + reversed_relationship = list(reversed(relationships)) + assert relationships != reversed_relationship + + relationships[:] = reversed_relationship + context.model.node.update(node) + assert get_relationships() == reversed_relationship + + flip_and_assert(source_node, 'outbound') + flip_and_assert(target_node, 'inbound') + + +class StrictClass(modeling.models.aria_declarative_base, modeling.mixins.ModelMixin): + __tablename__ = 'strict_class' + + strict_dict = sqlalchemy.Column(modeling.types.StrictDict(basestring, basestring)) + strict_list = sqlalchemy.Column(modeling.types.StrictList(basestring)) + + +def test_strict_dict(): + + strict_class = StrictClass() + + def assert_strict(sc): + with pytest.raises(ValueFormatException): + sc.strict_dict = {'key': 1} + + with pytest.raises(ValueFormatException): + sc.strict_dict = {1: 'value'} + + with pytest.raises(ValueFormatException): + sc.strict_dict = {1: 1} + + assert_strict(strict_class) + strict_class.strict_dict = {'key': 'value'} + assert strict_class.strict_dict == {'key': 'value'} + + assert_strict(strict_class) + with pytest.raises(ValueFormatException): + strict_class.strict_dict['key'] = 1 + with pytest.raises(ValueFormatException): + strict_class.strict_dict[1] = 'value' + with pytest.raises(ValueFormatException): + strict_class.strict_dict[1] = 1 + + +def test_strict_list(): + strict_class = StrictClass() + + def assert_strict(sc): + with pytest.raises(ValueFormatException): + sc.strict_list = [1] + + assert_strict(strict_class) + strict_class.strict_list = ['item'] + assert strict_class.strict_list == ['item'] + + assert_strict(strict_class) + with pytest.raises(ValueFormatException): + strict_class.strict_list[0] = 1