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.
20 from aria.storage import (
24 from aria import modeling
25 from aria.modeling.exceptions import ValueFormatException
27 from ..storage import (
28 release_sqlite_storage,
29 init_inmemory_model_storage
31 from . import MockModel
34 context as mock_context
40 base_storage = ModelStorage(sql_mapi.SQLAlchemyModelAPI,
41 initiator=init_inmemory_model_storage)
42 base_storage.register(MockModel)
44 release_sqlite_storage(base_storage)
47 @pytest.fixture(scope='module', autouse=True)
49 modeling.models.aria_declarative_base.metadata.remove(MockModel.__table__) # pylint: disable=no-member
54 ctx = mock_context.simple(str(tmpdir))
56 release_sqlite_storage(ctx.model)
59 def test_inner_dict_update(storage):
60 inner_dict = {'inner_value': 1}
62 mock_model = MockModel(model_dict={'inner_dict': inner_dict, 'value': 0})
63 storage.mock_model.put(mock_model)
65 storage_mm = storage.mock_model.get(mock_model.id)
66 assert storage_mm == mock_model
68 storage_mm.model_dict['inner_dict']['inner_value'] = 2
69 storage_mm.model_dict['value'] = -1
70 storage.mock_model.update(storage_mm)
71 storage_mm = storage.mock_model.get(storage_mm.id)
73 assert storage_mm.model_dict['inner_dict']['inner_value'] == 2
74 assert storage_mm.model_dict['value'] == -1
77 def test_inner_list_update(storage):
78 mock_model = MockModel(model_list=[0, [1]])
79 storage.mock_model.put(mock_model)
81 storage_mm = storage.mock_model.get(mock_model.id)
82 assert storage_mm == mock_model
84 storage_mm.model_list[1][0] = 'new_inner_value'
85 storage_mm.model_list[0] = 'new_value'
86 storage.mock_model.update(storage_mm)
87 storage_mm = storage.mock_model.get(storage_mm.id)
89 assert storage_mm.model_list[1][0] == 'new_inner_value'
90 assert storage_mm.model_list[0] == 'new_value'
93 def test_model_to_dict(context):
94 service = context.service
95 service = service.to_dict()
103 for expected_key in expected_keys:
104 assert expected_key in service
107 def test_relationship_model_ordering(context):
108 service = context.model.service.get_by_name(models.SERVICE_NAME)
109 source_node = context.model.node.get_by_name(models.DEPENDENT_NODE_NAME)
110 target_node = context.model.node.get_by_name(models.DEPENDENCY_NODE_NAME)
112 new_node_template = modeling.models.NodeTemplate(
113 name='new_node_template',
114 type=source_node.type,
115 service_template=service.service_template
118 new_node = modeling.models.Node(
120 type=source_node.type,
123 node_template=new_node_template,
124 state=modeling.models.Node.INITIAL,
127 source_node.outbound_relationships.append(modeling.models.Relationship(
128 source_node=source_node,
129 target_node=new_node,
132 new_node.outbound_relationships.append(modeling.models.Relationship( # pylint: disable=no-member
133 source_node=new_node,
134 target_node=target_node,
137 context.model.node_template.put(new_node_template)
138 context.model.node.put(new_node)
139 context.model.node.refresh(source_node)
140 context.model.node.refresh(target_node)
142 def flip_and_assert(node, direction):
144 Reversed the order of relationships and assert effects took place.
145 :param node: the node instance to operate on
146 :param direction: the type of relationships to flip (inbound/outbound)
149 assert direction in ('inbound', 'outbound')
151 def get_relationships():
152 return getattr(node, direction + '_relationships')
154 relationships = get_relationships()
155 assert len(relationships) == 2
157 reversed_relationship = list(reversed(relationships))
158 assert relationships != reversed_relationship
160 relationships[:] = reversed_relationship
161 context.model.node.update(node)
162 assert get_relationships() == reversed_relationship
164 flip_and_assert(source_node, 'outbound')
165 flip_and_assert(target_node, 'inbound')
168 class StrictClass(modeling.models.aria_declarative_base, modeling.mixins.ModelMixin):
169 __tablename__ = 'strict_class'
171 strict_dict = sqlalchemy.Column(modeling.types.StrictDict(basestring, basestring))
172 strict_list = sqlalchemy.Column(modeling.types.StrictList(basestring))
175 def test_strict_dict():
177 strict_class = StrictClass()
179 def assert_strict(sc):
180 with pytest.raises(ValueFormatException):
181 sc.strict_dict = {'key': 1}
183 with pytest.raises(ValueFormatException):
184 sc.strict_dict = {1: 'value'}
186 with pytest.raises(ValueFormatException):
187 sc.strict_dict = {1: 1}
189 assert_strict(strict_class)
190 strict_class.strict_dict = {'key': 'value'}
191 assert strict_class.strict_dict == {'key': 'value'}
193 assert_strict(strict_class)
194 with pytest.raises(ValueFormatException):
195 strict_class.strict_dict['key'] = 1
196 with pytest.raises(ValueFormatException):
197 strict_class.strict_dict[1] = 'value'
198 with pytest.raises(ValueFormatException):
199 strict_class.strict_dict[1] = 1
202 def test_strict_list():
203 strict_class = StrictClass()
205 def assert_strict(sc):
206 with pytest.raises(ValueFormatException):
209 assert_strict(strict_class)
210 strict_class.strict_list = ['item']
211 assert strict_class.strict_list == ['item']
213 assert_strict(strict_class)
214 with pytest.raises(ValueFormatException):
215 strict_class.strict_list[0] = 1