diff options
Diffstat (limited to 'azure/aria/aria-extension-cloudify/src/aria/tests/storage/test_model_storage.py')
| -rw-r--r-- | azure/aria/aria-extension-cloudify/src/aria/tests/storage/test_model_storage.py | 213 |
1 files changed, 213 insertions, 0 deletions
diff --git a/azure/aria/aria-extension-cloudify/src/aria/tests/storage/test_model_storage.py b/azure/aria/aria-extension-cloudify/src/aria/tests/storage/test_model_storage.py new file mode 100644 index 0000000..518d624 --- /dev/null +++ b/azure/aria/aria-extension-cloudify/src/aria/tests/storage/test_model_storage.py @@ -0,0 +1,213 @@ +# 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 |