azure/aria/aria-extension-cloudify/src/aria/tests/storage/test_model_storage.py

   1 # Licensed to the Apache Software Foundation (ASF) under one or more
   2 # contributor license agreements.  See the NOTICE file distributed with
   3 # this work for additional information regarding copyright ownership.
   4 # The ASF licenses this file to You under the Apache License, Version 2.0
   5 # (the "License"); you may not use this file except in compliance with
   6 # the License.  You may obtain a copy of the License at
   7 #
   8 #     http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 # Unless required by applicable law or agreed to in writing, software
  11 # distributed under the License is distributed on an "AS IS" BASIS,
  12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 # See the License for the specific language governing permissions and
  14 # limitations under the License.
  15
  16 import pytest
  17
  18 from sqlalchemy import (
  19     Column,
  20     Integer,
  21     Text
  22 )
  23
  24 from aria import (
  25     application_model_storage,
  26     modeling
  27 )
  28 from aria.storage import (
  29     ModelStorage,
  30     exceptions,
  31     sql_mapi,
  32 )
  33
  34 from tests import (
  35     mock,
  36     storage as tests_storage,
  37     modeling as tests_modeling
  38 )
  39
  40
  41 @pytest.fixture
  42 def storage():
  43     base_storage = ModelStorage(sql_mapi.SQLAlchemyModelAPI,
  44                                 initiator=tests_storage.init_inmemory_model_storage)
  45     base_storage.register(tests_modeling.MockModel)
  46     yield base_storage
  47     tests_storage.release_sqlite_storage(base_storage)
  48
  49
  50 @pytest.fixture(scope='module', autouse=True)
  51 def module_cleanup():
  52     modeling.models.aria_declarative_base.metadata.remove(tests_modeling.MockModel.__table__)  #pylint: disable=no-member
  53
  54
  55 def test_storage_base(storage):
  56     with pytest.raises(AttributeError):
  57         storage.non_existent_attribute()
  58
  59
  60 def test_model_storage(storage):
  61     mock_model = tests_modeling.MockModel(value=0, name='model_name')
  62     storage.mock_model.put(mock_model)
  63
  64     assert storage.mock_model.get_by_name('model_name') == mock_model
  65
  66     assert [mm_from_storage for mm_from_storage in storage.mock_model.iter()] == [mock_model]
  67     assert [mm_from_storage for mm_from_storage in storage.mock_model] == [mock_model]
  68
  69     storage.mock_model.delete(mock_model)
  70     with pytest.raises(exceptions.StorageError):
  71         storage.mock_model.get(mock_model.id)
  72
  73
  74 def test_application_storage_factory():
  75     storage = application_model_storage(sql_mapi.SQLAlchemyModelAPI,
  76                                         initiator=tests_storage.init_inmemory_model_storage)
  77
  78     assert storage.service_template
  79     assert storage.node_template
  80     assert storage.group_template
  81     assert storage.policy_template
  82     assert storage.substitution_template
  83     assert storage.substitution_template_mapping
  84     assert storage.requirement_template
  85     assert storage.relationship_template
  86     assert storage.capability_template
  87     assert storage.interface_template
  88     assert storage.operation_template
  89     assert storage.artifact_template
  90
  91     assert storage.service
  92     assert storage.node
  93     assert storage.group
  94     assert storage.policy
  95     assert storage.substitution
  96     assert storage.substitution_mapping
  97     assert storage.relationship
  98     assert storage.capability
  99     assert storage.interface
 100     assert storage.operation
 101     assert storage.artifact
 102
 103     assert storage.execution
 104     assert storage.service_update
 105     assert storage.service_update_step
 106     assert storage.service_modification
 107     assert storage.plugin
 108     assert storage.task
 109
 110     assert storage.input
 111     assert storage.output
 112     assert storage.property
 113     assert storage.attribute
 114
 115     assert storage.type
 116     assert storage.metadata
 117
 118     tests_storage.release_sqlite_storage(storage)
 119
 120
 121 def test_cascade_deletion(context):
 122     service = context.model.service.list()[0]
 123
 124     assert len(context.model.service_template.list()) == 1
 125     assert len(service.nodes) == len(context.model.node.list()) == 2
 126
 127     context.model.service.delete(service)
 128
 129     assert len(context.model.service_template.list()) == 1
 130     assert len(context.model.service.list()) == 0
 131     assert len(context.model.node.list()) == 0
 132
 133
 134 @pytest.fixture
 135 def context(tmpdir):
 136     result = mock.context.simple(str(tmpdir))
 137     yield result
 138     tests_storage.release_sqlite_storage(result.model)
 139
 140
 141 def test_mapi_include(context):
 142     service1 = context.model.service.list()[0]
 143     service1.name = 'service1'
 144     service1.service_template.name = 'service_template1'
 145     context.model.service.update(service1)
 146
 147     service_template2 = mock.models.create_service_template('service_template2')
 148     service2 = mock.models.create_service(service_template2, 'service2')
 149     context.model.service.put(service2)
 150
 151     assert service1 != service2
 152     assert service1.service_template != service2.service_template
 153
 154     def assert_include(service):
 155         st_name = context.model.service.get(service.id, include=('service_template_name',))
 156         st_name_list = context.model.service.list(filters={'id': service.id},
 157                                                   include=('service_template_name', ))
 158         assert len(st_name) == len(st_name_list) == 1
 159         assert st_name[0] == st_name_list[0][0] == service.service_template.name
 160
 161     assert_include(service1)
 162     assert_include(service2)
 163
 164
 165 class MockModel(modeling.models.aria_declarative_base, modeling.mixins.ModelMixin): #pylint: disable=abstract-method
 166     __tablename__ = 'op_mock_model'
 167
 168     name = Column(Text)
 169     value = Column(Integer)
 170
 171
 172 class TestFilterOperands(object):
 173
 174     @pytest.fixture()
 175     def storage(self):
 176         model_storage = application_model_storage(
 177             sql_mapi.SQLAlchemyModelAPI, initiator=tests_storage.init_inmemory_model_storage)
 178         model_storage.register(MockModel)
 179         for value in (1, 2, 3, 4):
 180             model_storage.op_mock_model.put(MockModel(value=value))
 181         yield model_storage
 182         tests_storage.release_sqlite_storage(model_storage)
 183
 184     def test_gt(self, storage):
 185         assert len(storage.op_mock_model.list(filters=dict(value=dict(gt=3)))) == 1
 186         assert len(storage.op_mock_model.list(filters=dict(value=dict(gt=4)))) == 0
 187
 188     def test_ge(self, storage):
 189         assert len(storage.op_mock_model.list(filters=dict(value=dict(ge=3)))) == 2
 190         assert len(storage.op_mock_model.list(filters=dict(value=dict(ge=5)))) == 0
 191
 192     def test_lt(self, storage):
 193         assert len(storage.op_mock_model.list(filters=dict(value=dict(lt=2)))) == 1
 194         assert len(storage.op_mock_model.list(filters=dict(value=dict(lt=1)))) == 0
 195
 196     def test_le(self, storage):
 197         assert len(storage.op_mock_model.list(filters=dict(value=dict(le=2)))) == 2
 198         assert len(storage.op_mock_model.list(filters=dict(value=dict(le=0)))) == 0
 199
 200     def test_eq(self, storage):
 201         assert len(storage.op_mock_model.list(filters=dict(value=dict(eq=2)))) == 1
 202         assert len(storage.op_mock_model.list(filters=dict(value=dict(eq=0)))) == 0
 203
 204     def test_neq(self, storage):
 205         assert len(storage.op_mock_model.list(filters=dict(value=dict(ne=2)))) == 3
 206
 207     def test_gt_and_lt(self, storage):
 208         assert len(storage.op_mock_model.list(filters=dict(value=dict(gt=1, lt=3)))) == 1
 209         assert len(storage.op_mock_model.list(filters=dict(value=dict(gt=2, lt=2)))) == 0
 210
 211     def test_eq_and_ne(self, storage):
 212         assert len(storage.op_mock_model.list(filters=dict(value=dict(eq=1, ne=3)))) == 1
 213         assert len(storage.op_mock_model.list(filters=dict(value=dict(eq=1, ne=1)))) == 0