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
8 # http://www.apache.org/licenses/LICENSE-2.0
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.
18 from sqlalchemy import (
25 application_model_storage,
28 from aria.storage import (
36 storage as tests_storage,
37 modeling as tests_modeling
43 base_storage = ModelStorage(sql_mapi.SQLAlchemyModelAPI,
44 initiator=tests_storage.init_inmemory_model_storage)
45 base_storage.register(tests_modeling.MockModel)
47 tests_storage.release_sqlite_storage(base_storage)
50 @pytest.fixture(scope='module', autouse=True)
52 modeling.models.aria_declarative_base.metadata.remove(tests_modeling.MockModel.__table__) #pylint: disable=no-member
55 def test_storage_base(storage):
56 with pytest.raises(AttributeError):
57 storage.non_existent_attribute()
60 def test_model_storage(storage):
61 mock_model = tests_modeling.MockModel(value=0, name='model_name')
62 storage.mock_model.put(mock_model)
64 assert storage.mock_model.get_by_name('model_name') == mock_model
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]
69 storage.mock_model.delete(mock_model)
70 with pytest.raises(exceptions.StorageError):
71 storage.mock_model.get(mock_model.id)
74 def test_application_storage_factory():
75 storage = application_model_storage(sql_mapi.SQLAlchemyModelAPI,
76 initiator=tests_storage.init_inmemory_model_storage)
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
91 assert storage.service
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
103 assert storage.execution
104 assert storage.service_update
105 assert storage.service_update_step
106 assert storage.service_modification
107 assert storage.plugin
111 assert storage.output
112 assert storage.property
113 assert storage.attribute
116 assert storage.metadata
118 tests_storage.release_sqlite_storage(storage)
121 def test_cascade_deletion(context):
122 service = context.model.service.list()[0]
124 assert len(context.model.service_template.list()) == 1
125 assert len(service.nodes) == len(context.model.node.list()) == 2
127 context.model.service.delete(service)
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
136 result = mock.context.simple(str(tmpdir))
138 tests_storage.release_sqlite_storage(result.model)
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)
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)
151 assert service1 != service2
152 assert service1.service_template != service2.service_template
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
161 assert_include(service1)
162 assert_include(service2)
165 class MockModel(modeling.models.aria_declarative_base, modeling.mixins.ModelMixin): #pylint: disable=abstract-method
166 __tablename__ = 'op_mock_model'
169 value = Column(Integer)
172 class TestFilterOperands(object):
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))
182 tests_storage.release_sqlite_storage(model_storage)
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
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
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
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
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
204 def test_neq(self, storage):
205 assert len(storage.op_mock_model.list(filters=dict(value=dict(ne=2)))) == 3
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
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