--- /dev/null
+# 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
+
+from sqlalchemy import (
+ Column,
+ Integer,
+ Text
+)
+
+from aria import (
+ application_model_storage,
+ modeling
+)
+from aria.storage import (
+ ModelStorage,
+ exceptions,
+ sql_mapi,
+)
+
+from tests import (
+ mock,
+ storage as tests_storage,
+ modeling as tests_modeling
+)
+
+
+@pytest.fixture
+def storage():
+ base_storage = ModelStorage(sql_mapi.SQLAlchemyModelAPI,
+ initiator=tests_storage.init_inmemory_model_storage)
+ base_storage.register(tests_modeling.MockModel)
+ yield base_storage
+ tests_storage.release_sqlite_storage(base_storage)
+
+
+@pytest.fixture(scope='module', autouse=True)
+def module_cleanup():
+ modeling.models.aria_declarative_base.metadata.remove(tests_modeling.MockModel.__table__) #pylint: disable=no-member
+
+
+def test_storage_base(storage):
+ with pytest.raises(AttributeError):
+ storage.non_existent_attribute()
+
+
+def test_model_storage(storage):
+ mock_model = tests_modeling.MockModel(value=0, name='model_name')
+ storage.mock_model.put(mock_model)
+
+ assert storage.mock_model.get_by_name('model_name') == mock_model
+
+ assert [mm_from_storage for mm_from_storage in storage.mock_model.iter()] == [mock_model]
+ assert [mm_from_storage for mm_from_storage in storage.mock_model] == [mock_model]
+
+ storage.mock_model.delete(mock_model)
+ with pytest.raises(exceptions.StorageError):
+ storage.mock_model.get(mock_model.id)
+
+
+def test_application_storage_factory():
+ storage = application_model_storage(sql_mapi.SQLAlchemyModelAPI,
+ initiator=tests_storage.init_inmemory_model_storage)
+
+ assert storage.service_template
+ assert storage.node_template
+ assert storage.group_template
+ assert storage.policy_template
+ assert storage.substitution_template
+ assert storage.substitution_template_mapping
+ assert storage.requirement_template
+ assert storage.relationship_template
+ assert storage.capability_template
+ assert storage.interface_template
+ assert storage.operation_template
+ assert storage.artifact_template
+
+ assert storage.service
+ assert storage.node
+ assert storage.group
+ assert storage.policy
+ assert storage.substitution
+ assert storage.substitution_mapping
+ assert storage.relationship
+ assert storage.capability
+ assert storage.interface
+ assert storage.operation
+ assert storage.artifact
+
+ assert storage.execution
+ assert storage.service_update
+ assert storage.service_update_step
+ assert storage.service_modification
+ assert storage.plugin
+ assert storage.task
+
+ assert storage.input
+ assert storage.output
+ assert storage.property
+ assert storage.attribute
+
+ assert storage.type
+ assert storage.metadata
+
+ tests_storage.release_sqlite_storage(storage)
+
+
+def test_cascade_deletion(context):
+ service = context.model.service.list()[0]
+
+ assert len(context.model.service_template.list()) == 1
+ assert len(service.nodes) == len(context.model.node.list()) == 2
+
+ context.model.service.delete(service)
+
+ assert len(context.model.service_template.list()) == 1
+ assert len(context.model.service.list()) == 0
+ assert len(context.model.node.list()) == 0
+
+
+@pytest.fixture
+def context(tmpdir):
+ result = mock.context.simple(str(tmpdir))
+ yield result
+ tests_storage.release_sqlite_storage(result.model)
+
+
+def test_mapi_include(context):
+ service1 = context.model.service.list()[0]
+ service1.name = 'service1'
+ service1.service_template.name = 'service_template1'
+ context.model.service.update(service1)
+
+ service_template2 = mock.models.create_service_template('service_template2')
+ service2 = mock.models.create_service(service_template2, 'service2')
+ context.model.service.put(service2)
+
+ assert service1 != service2
+ assert service1.service_template != service2.service_template
+
+ def assert_include(service):
+ st_name = context.model.service.get(service.id, include=('service_template_name',))
+ st_name_list = context.model.service.list(filters={'id': service.id},
+ include=('service_template_name', ))
+ assert len(st_name) == len(st_name_list) == 1
+ assert st_name[0] == st_name_list[0][0] == service.service_template.name
+
+ assert_include(service1)
+ assert_include(service2)
+
+
+class MockModel(modeling.models.aria_declarative_base, modeling.mixins.ModelMixin): #pylint: disable=abstract-method
+ __tablename__ = 'op_mock_model'
+
+ name = Column(Text)
+ value = Column(Integer)
+
+
+class TestFilterOperands(object):
+
+ @pytest.fixture()
+ def storage(self):
+ model_storage = application_model_storage(
+ sql_mapi.SQLAlchemyModelAPI, initiator=tests_storage.init_inmemory_model_storage)
+ model_storage.register(MockModel)
+ for value in (1, 2, 3, 4):
+ model_storage.op_mock_model.put(MockModel(value=value))
+ yield model_storage
+ tests_storage.release_sqlite_storage(model_storage)
+
+ def test_gt(self, storage):
+ assert len(storage.op_mock_model.list(filters=dict(value=dict(gt=3)))) == 1
+ assert len(storage.op_mock_model.list(filters=dict(value=dict(gt=4)))) == 0
+
+ def test_ge(self, storage):
+ assert len(storage.op_mock_model.list(filters=dict(value=dict(ge=3)))) == 2
+ assert len(storage.op_mock_model.list(filters=dict(value=dict(ge=5)))) == 0
+
+ def test_lt(self, storage):
+ assert len(storage.op_mock_model.list(filters=dict(value=dict(lt=2)))) == 1
+ assert len(storage.op_mock_model.list(filters=dict(value=dict(lt=1)))) == 0
+
+ def test_le(self, storage):
+ assert len(storage.op_mock_model.list(filters=dict(value=dict(le=2)))) == 2
+ assert len(storage.op_mock_model.list(filters=dict(value=dict(le=0)))) == 0
+
+ def test_eq(self, storage):
+ assert len(storage.op_mock_model.list(filters=dict(value=dict(eq=2)))) == 1
+ assert len(storage.op_mock_model.list(filters=dict(value=dict(eq=0)))) == 0
+
+ def test_neq(self, storage):
+ assert len(storage.op_mock_model.list(filters=dict(value=dict(ne=2)))) == 3
+
+ def test_gt_and_lt(self, storage):
+ assert len(storage.op_mock_model.list(filters=dict(value=dict(gt=1, lt=3)))) == 1
+ assert len(storage.op_mock_model.list(filters=dict(value=dict(gt=2, lt=2)))) == 0
+
+ def test_eq_and_ne(self, storage):
+ assert len(storage.op_mock_model.list(filters=dict(value=dict(eq=1, ne=3)))) == 1
+ assert len(storage.op_mock_model.list(filters=dict(value=dict(eq=1, ne=1)))) == 0
Code Review / multicloud / azure.git / blobdiff