1 files changed, 215 insertions, 0 deletions
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