azure/aria/aria-extension-cloudify/src/aria/tests/modeling/test_mixins.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 import sqlalchemy
  19
  20 from aria.storage import (
  21     ModelStorage,
  22     sql_mapi
  23 )
  24 from aria import modeling
  25 from aria.modeling.exceptions import ValueFormatException
  26
  27 from ..storage import (
  28     release_sqlite_storage,
  29     init_inmemory_model_storage
  30 )
  31 from . import MockModel
  32 from ..mock import (
  33     models,
  34     context as mock_context
  35 )
  36
  37
  38 @pytest.fixture
  39 def storage():
  40     base_storage = ModelStorage(sql_mapi.SQLAlchemyModelAPI,
  41                                 initiator=init_inmemory_model_storage)
  42     base_storage.register(MockModel)
  43     yield base_storage
  44     release_sqlite_storage(base_storage)
  45
  46
  47 @pytest.fixture(scope='module', autouse=True)
  48 def module_cleanup():
  49     modeling.models.aria_declarative_base.metadata.remove(MockModel.__table__)                      # pylint: disable=no-member
  50
  51
  52 @pytest.fixture
  53 def context(tmpdir):
  54     ctx = mock_context.simple(str(tmpdir))
  55     yield ctx
  56     release_sqlite_storage(ctx.model)
  57
  58
  59 def test_inner_dict_update(storage):
  60     inner_dict = {'inner_value': 1}
  61
  62     mock_model = MockModel(model_dict={'inner_dict': inner_dict, 'value': 0})
  63     storage.mock_model.put(mock_model)
  64
  65     storage_mm = storage.mock_model.get(mock_model.id)
  66     assert storage_mm == mock_model
  67
  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)
  72
  73     assert storage_mm.model_dict['inner_dict']['inner_value'] == 2
  74     assert storage_mm.model_dict['value'] == -1
  75
  76
  77 def test_inner_list_update(storage):
  78     mock_model = MockModel(model_list=[0, [1]])
  79     storage.mock_model.put(mock_model)
  80
  81     storage_mm = storage.mock_model.get(mock_model.id)
  82     assert storage_mm == mock_model
  83
  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)
  88
  89     assert storage_mm.model_list[1][0] == 'new_inner_value'
  90     assert storage_mm.model_list[0] == 'new_value'
  91
  92
  93 def test_model_to_dict(context):
  94     service = context.service
  95     service = service.to_dict()
  96
  97     expected_keys = [
  98         'description',
  99         'created_at',
 100         'updated_at'
 101     ]
 102
 103     for expected_key in expected_keys:
 104         assert expected_key in service
 105
 106
 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)
 111
 112     new_node_template = modeling.models.NodeTemplate(
 113         name='new_node_template',
 114         type=source_node.type,
 115         service_template=service.service_template
 116     )
 117
 118     new_node = modeling.models.Node(
 119         name='new_node',
 120         type=source_node.type,
 121         service=service,
 122         version=None,
 123         node_template=new_node_template,
 124         state=modeling.models.Node.INITIAL,
 125     )
 126
 127     source_node.outbound_relationships.append(modeling.models.Relationship(
 128         source_node=source_node,
 129         target_node=new_node,
 130     ))
 131
 132     new_node.outbound_relationships.append(modeling.models.Relationship(                            # pylint: disable=no-member
 133         source_node=new_node,
 134         target_node=target_node,
 135     ))
 136
 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)
 141
 142     def flip_and_assert(node, direction):
 143         """
 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)
 147         :return:
 148         """
 149         assert direction in ('inbound', 'outbound')
 150
 151         def get_relationships():
 152             return getattr(node, direction + '_relationships')
 153
 154         relationships = get_relationships()
 155         assert len(relationships) == 2
 156
 157         reversed_relationship = list(reversed(relationships))
 158         assert relationships != reversed_relationship
 159
 160         relationships[:] = reversed_relationship
 161         context.model.node.update(node)
 162         assert get_relationships() == reversed_relationship
 163
 164     flip_and_assert(source_node, 'outbound')
 165     flip_and_assert(target_node, 'inbound')
 166
 167
 168 class StrictClass(modeling.models.aria_declarative_base, modeling.mixins.ModelMixin):
 169     __tablename__ = 'strict_class'
 170
 171     strict_dict = sqlalchemy.Column(modeling.types.StrictDict(basestring, basestring))
 172     strict_list = sqlalchemy.Column(modeling.types.StrictList(basestring))
 173
 174
 175 def test_strict_dict():
 176
 177     strict_class = StrictClass()
 178
 179     def assert_strict(sc):
 180         with pytest.raises(ValueFormatException):
 181             sc.strict_dict = {'key': 1}
 182
 183         with pytest.raises(ValueFormatException):
 184             sc.strict_dict = {1: 'value'}
 185
 186         with pytest.raises(ValueFormatException):
 187             sc.strict_dict = {1: 1}
 188
 189     assert_strict(strict_class)
 190     strict_class.strict_dict = {'key': 'value'}
 191     assert strict_class.strict_dict == {'key': 'value'}
 192
 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
 200
 201
 202 def test_strict_list():
 203     strict_class = StrictClass()
 204
 205     def assert_strict(sc):
 206         with pytest.raises(ValueFormatException):
 207             sc.strict_list = [1]
 208
 209     assert_strict(strict_class)
 210     strict_class.strict_list = ['item']
 211     assert strict_class.strict_list == ['item']
 212
 213     assert_strict(strict_class)
 214     with pytest.raises(ValueFormatException):
 215         strict_class.strict_list[0] = 1