You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@ariatosca.apache.org by mx...@apache.org on 2016/11/27 16:16:39 UTC
[2/4] incubator-ariatosca git commit: fixed the models and some tests
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/7b473b63/tests/storage/test_model_storage.py
----------------------------------------------------------------------
diff --git a/tests/storage/test_model_storage.py b/tests/storage/test_model_storage.py
index 4b83fd0..c5879ae 100644
--- a/tests/storage/test_model_storage.py
+++ b/tests/storage/test_model_storage.py
@@ -15,32 +15,29 @@
import pytest
+from aria import application_model_storage
+
from aria.storage import (
Storage,
- ModelStorage,
models,
+ exceptions,
+ api_driver as storage_api,
)
-from aria.storage import structures
-from aria.storage.exceptions import StorageError
-from aria import application_model_storage
-
-from . import InMemoryModelDriver
def test_storage_base():
- driver = InMemoryModelDriver()
- storage = Storage(driver)
+ api = storage_api.InMemoryModelAPI
+ storage = Storage(api)
- assert storage.driver == driver
+ assert storage.api == api
with pytest.raises(AttributeError):
storage.non_existent_attribute()
def test_model_storage():
- storage = ModelStorage(InMemoryModelDriver())
+ storage = Storage(storage_api.InMemoryModelAPI)
storage.register(models.ProviderContext)
- storage.setup()
pc = models.ProviderContext(context={}, name='context_name', id='id1')
storage.provider_context.store(pc)
@@ -54,34 +51,30 @@ def test_model_storage():
assert storage.provider_context.get('id1').context == {'update_key': 0}
storage.provider_context.delete('id1')
- with pytest.raises(StorageError):
+ with pytest.raises(exceptions.StorageError):
storage.provider_context.get('id1')
def test_storage_driver():
- storage = ModelStorage(InMemoryModelDriver())
+ storage = Storage(storage_api.InMemoryModelAPI)
storage.register(models.ProviderContext)
- storage.setup()
+
pc = models.ProviderContext(context={}, name='context_name', id='id2')
- storage.driver.store(name='provider_context', entry=pc.fields_dict, entry_id=pc.id)
+ storage.registered['provider_context'].store(entry=pc)
- assert storage.driver.get(
- name='provider_context',
- entry_id='id2',
- model_cls=models.ProviderContext) == pc.fields_dict
+ assert storage.registered['provider_context'].get(entry_id='id2').fields_dict == pc.fields_dict
- assert next(i for i in storage.driver.iter(name='provider_context')) == pc
- # assert [i for i in storage.provider_context] == [pc]
+ assert next(i.fields_dict for i in storage.registered['provider_context'].iter()) == pc.fields_dict
+ assert [i for i in storage.provider_context] == [pc]
- storage.provider_context.delete('id2')
+ storage.registered['provider_context'].delete('id2')
- with pytest.raises(StorageError):
- storage.provider_context.get('id2')
+ with pytest.raises(exceptions.StorageError):
+ storage.registered['provider_context'].get('id2')
def test_application_storage_factory():
- driver = InMemoryModelDriver()
- storage = application_model_storage(driver)
+ storage = application_model_storage(storage_api.InMemoryModelAPI)
assert storage.node
assert storage.node_instance
assert storage.plugin
@@ -94,68 +87,68 @@ def test_application_storage_factory():
assert storage.execution
assert storage.provider_context
- reused_storage = application_model_storage(driver)
+ reused_storage = application_model_storage(storage_api.InMemoryModelAPI)
assert reused_storage == storage
-def test_storage_pointers():
- class PointedModel(Model):
- id = Field()
-
- class PointingModel(Model):
- id = Field()
- pointing_field = PointerField(type=PointedModel)
-
- storage = ModelStorage(InMemoryModelDriver(), model_classes=[PointingModel])
- storage.setup()
-
- assert storage.pointed_model
- assert storage.pointing_model
-
- pointed_model = PointedModel(id='pointed_id')
-
- pointing_model = PointingModel(id='pointing_id', pointing_field=pointed_model)
- storage.pointing_model.store(pointing_model)
-
- assert storage.pointed_model.get('pointed_id') == pointed_model
- assert storage.pointing_model.get('pointing_id') == pointing_model
-
- storage.pointing_model.delete('pointing_id')
-
- with pytest.raises(StorageError):
- assert storage.pointed_model.get('pointed_id')
- assert storage.pointing_model.get('pointing_id')
-
-
-def test_storage_iter_pointers():
- class PointedIterModel(models.Model):
- id = structures.Field()
-
- class PointingIterModel(models.Model):
- id = models.Field()
- pointing_field = structures.IterPointerField(type=PointedIterModel)
-
- storage = ModelStorage(InMemoryModelDriver(), model_classes=[PointingIterModel])
- storage.setup()
-
- assert storage.pointed_iter_model
- assert storage.pointing_iter_model
-
- pointed_iter_model1 = PointedIterModel(id='pointed_id1')
- pointed_iter_model2 = PointedIterModel(id='pointed_id2')
-
- pointing_iter_model = PointingIterModel(
- id='pointing_id',
- pointing_field=[pointed_iter_model1, pointed_iter_model2])
- storage.pointing_iter_model.store(pointing_iter_model)
-
- assert storage.pointed_iter_model.get('pointed_id1') == pointed_iter_model1
- assert storage.pointed_iter_model.get('pointed_id2') == pointed_iter_model2
- assert storage.pointing_iter_model.get('pointing_id') == pointing_iter_model
+# def test_storage_pointers():
+# class PointedModel(Model):
+# id = Field()
+#
+# class PointingModel(Model):
+# id = Field()
+# pointing_field = PointerField(type=PointedModel)
+#
+# storage = ModelStorage(InMemoryModelDriver(), model_classes=[PointingModel])
+# storage.setup()
+#
+# assert storage.pointed_model
+# assert storage.pointing_model
+#
+# pointed_model = PointedModel(id='pointed_id')
+#
+# pointing_model = PointingModel(id='pointing_id', pointing_field=pointed_model)
+# storage.pointing_model.store(pointing_model)
+#
+# assert storage.pointed_model.get('pointed_id') == pointed_model
+# assert storage.pointing_model.get('pointing_id') == pointing_model
+#
+# storage.pointing_model.delete('pointing_id')
+#
+# with pytest.raises(StorageError):
+# assert storage.pointed_model.get('pointed_id')
+# assert storage.pointing_model.get('pointing_id')
- storage.pointing_iter_model.delete('pointing_id')
- with pytest.raises(StorageError):
- assert storage.pointed_iter_model.get('pointed_id1')
- assert storage.pointed_iter_model.get('pointed_id2')
- assert storage.pointing_iter_model.get('pointing_id')
+# def test_storage_iter_pointers():
+# class PointedIterModel(models.Model):
+# id = structures.Field()
+#
+# class PointingIterModel(models.Model):
+# id = models.Field()
+# pointing_field = structures.IterPointerField(type=PointedIterModel)
+#
+# storage = ModelStorage(InMemoryModelDriver(), model_classes=[PointingIterModel])
+# storage.setup()
+#
+# assert storage.pointed_iter_model
+# assert storage.pointing_iter_model
+#
+# pointed_iter_model1 = PointedIterModel(id='pointed_id1')
+# pointed_iter_model2 = PointedIterModel(id='pointed_id2')
+#
+# pointing_iter_model = PointingIterModel(
+# id='pointing_id',
+# pointing_field=[pointed_iter_model1, pointed_iter_model2])
+# storage.pointing_iter_model.store(pointing_iter_model)
+#
+# assert storage.pointed_iter_model.get('pointed_id1') == pointed_iter_model1
+# assert storage.pointed_iter_model.get('pointed_id2') == pointed_iter_model2
+# assert storage.pointing_iter_model.get('pointing_id') == pointing_iter_model
+#
+# storage.pointing_iter_model.delete('pointing_id')
+#
+# with pytest.raises(StorageError):
+# assert storage.pointed_iter_model.get('pointed_id1')
+# assert storage.pointed_iter_model.get('pointed_id2')
+# assert storage.pointing_iter_model.get('pointing_id')
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/7b473b63/tests/storage/test_models.py
----------------------------------------------------------------------
diff --git a/tests/storage/test_models.py b/tests/storage/test_models.py
deleted file mode 100644
index 414c7c4..0000000
--- a/tests/storage/test_models.py
+++ /dev/null
@@ -1,363 +0,0 @@
-# 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 json
-from datetime import datetime
-
-import pytest
-
-from aria.storage.exceptions import StorageError
-from aria.storage.models import (
- DeploymentUpdateStep,
- Relationship,
- RelationshipInstance,
- Node,
- NodeInstance,
- Blueprint,
- Execution,
- Task
-)
-from tests.mock import models
-
-# TODO: add tests per model
-
-
-def test_base_model_without_fields():
- with pytest.raises(StorageError, message="Id field has to be in model fields"):
- Model()
-
-
-def test_base_model_members():
- _test_field = Field()
-
- class TestModel1(Model):
- test_field = _test_field
- id = Field(default='test_id')
-
- assert _test_field is TestModel1.test_field
-
- test_model = TestModel1(test_field='test_field_value', id='test_id')
-
- assert repr(test_model) == "TestModel1(fields=['id', 'test_field'])"
- expected = {'test_field': 'test_field_value', 'id': 'test_id'}
- assert json.loads(test_model.json) == expected
- assert test_model.fields_dict == expected
-
- with pytest.raises(StorageError):
- TestModel1()
-
- with pytest.raises(StorageError):
- TestModel1(test_field='test_field_value', id='test_id', unsupported_field='value')
-
- class TestModel2(Model):
- test_field = Field()
- id = Field()
-
- with pytest.raises(StorageError):
- TestModel2()
-
-
-def test_blueprint_model():
- Blueprint(
- plan={},
- id='id',
- description='description',
- created_at=datetime.utcnow(),
- updated_at=datetime.utcnow(),
- main_file_name='/path',
- )
- with pytest.raises(TypeError):
- Blueprint(
- plan=None,
- id='id',
- description='description',
- created_at=datetime.utcnow(),
- updated_at=datetime.utcnow(),
- main_file_name='/path',
- )
- with pytest.raises(TypeError):
- Blueprint(
- plan={},
- id=999,
- description='description',
- created_at=datetime.utcnow(),
- updated_at=datetime.utcnow(),
- main_file_name='/path',
- )
- with pytest.raises(TypeError):
- Blueprint(
- plan={},
- id='id',
- description=999,
- created_at=datetime.utcnow(),
- updated_at=datetime.utcnow(),
- main_file_name='/path',
- )
- with pytest.raises(TypeError):
- Blueprint(
- plan={},
- id='id',
- description='description',
- created_at='error',
- updated_at=datetime.utcnow(),
- main_file_name='/path',
- )
- with pytest.raises(TypeError):
- Blueprint(
- plan={},
- id='id',
- description='description',
- created_at=datetime.utcnow(),
- updated_at=None,
- main_file_name='/path',
- )
- with pytest.raises(TypeError):
- Blueprint(
- plan={},
- id='id',
- description='description',
- created_at=datetime.utcnow(),
- updated_at=None,
- main_file_name=88,
- )
- Blueprint(
- plan={},
- description='description',
- created_at=datetime.utcnow(),
- updated_at=datetime.utcnow(),
- main_file_name='/path',
- )
-
-
-def test_deployment_update_step_model():
- add_node = DeploymentUpdateStep(
- id='add_step',
- action='add',
- entity_type='node',
- entity_id='node_id')
-
- modify_node = DeploymentUpdateStep(
- id='modify_step',
- action='modify',
- entity_type='node',
- entity_id='node_id')
-
- remove_node = DeploymentUpdateStep(
- id='remove_step',
- action='remove',
- entity_type='node',
- entity_id='node_id')
-
- for step in (add_node, modify_node, remove_node):
- assert hash((step.id, step.entity_id)) == hash(step)
-
- assert remove_node < modify_node < add_node
- assert not remove_node > modify_node > add_node
-
- add_rel = DeploymentUpdateStep(
- id='add_step',
- action='add',
- entity_type='relationship',
- entity_id='relationship_id')
-
- # modify_rel = DeploymentUpdateStep(
- # id='modify_step',
- # action='modify',
- # entity_type='relationship',
- # entity_id='relationship_id')
-
- remove_rel = DeploymentUpdateStep(
- id='remove_step',
- action='remove',
- entity_type='relationship',
- entity_id='relationship_id')
-
- assert remove_rel < remove_node < add_node < add_rel
- assert not add_node < None
- # TODO fix logic here so that pylint is happy
- # assert not modify_node < modify_rel and not modify_rel < modify_node
-
-
-def _relationship(id=''):
- return Relationship(
- id='rel{0}'.format(id),
- target_id='target{0}'.format(id),
- source_id='source{0}'.format(id),
- source_interfaces={},
- source_operations={},
- target_interfaces={},
- target_operations={},
- type='type{0}'.format(id),
- type_hierarchy=[],
- properties={})
-
-
-def test_relationships():
- relationships = [_relationship(index) for index in xrange(3)]
-
- node = Node(
- blueprint_id='blueprint_id',
- type='type',
- type_hierarchy=None,
- number_of_instances=1,
- planned_number_of_instances=1,
- deploy_number_of_instances=1,
- properties={},
- operations={},
- relationships=relationships,
- min_number_of_instances=1,
- max_number_of_instances=1)
-
- for index in xrange(3):
- assert relationships[index] is \
- next(node.relationships_by_target('target{0}'.format(index)))
-
- relationship = _relationship()
-
- node = Node(
- blueprint_id='blueprint_id',
- type='type',
- type_hierarchy=None,
- number_of_instances=1,
- planned_number_of_instances=1,
- deploy_number_of_instances=1,
- properties={},
- operations={},
- relationships=[relationship, relationship, relationship],
- min_number_of_instances=1,
- max_number_of_instances=1)
-
- for node_relationship in node.relationships_by_target('target'):
- assert relationship is node_relationship
-
-
-def test_relationship_instance():
- relationship = _relationship()
- relationship_instances = [RelationshipInstance(
- id='rel{0}'.format(index),
- target_id='target_{0}'.format(index % 2),
- source_id='source_{0}'.format(index % 2),
- source_name='',
- target_name='',
- relationship=relationship,
- type='type{0}'.format(index)) for index in xrange(3)]
-
- node_instance = NodeInstance(
- deployment_id='deployment_id',
- runtime_properties={},
- version='1',
- relationship_instances=relationship_instances,
- node=Node(
- blueprint_id='blueprint_id',
- type='type',
- type_hierarchy=None,
- number_of_instances=1,
- planned_number_of_instances=1,
- deploy_number_of_instances=1,
- properties={},
- operations={},
- relationships=[],
- min_number_of_instances=1,
- max_number_of_instances=1),
- scaling_groups=()
- )
-
- from itertools import chain
-
- assert set(relationship_instances) == set(chain(
- node_instance.relationships_by_target('target_0'),
- node_instance.relationships_by_target('target_1')))
-
-
-def test_execution_status_transition():
- def create_execution(status):
- return Execution(
- id='e_id',
- deployment_id='d_id',
- workflow_id='w_id',
- blueprint_id='b_id',
- status=status,
- parameters={}
- )
-
- valid_transitions = {
- Execution.PENDING: [Execution.STARTED,
- Execution.CANCELLED,
- Execution.PENDING],
- Execution.STARTED: [Execution.FAILED,
- Execution.TERMINATED,
- Execution.CANCELLED,
- Execution.CANCELLING,
- Execution.STARTED],
- Execution.CANCELLING: [Execution.FAILED,
- Execution.TERMINATED,
- Execution.CANCELLED,
- Execution.CANCELLING],
- Execution.FAILED: [Execution.FAILED],
- Execution.TERMINATED: [Execution.TERMINATED],
- Execution.CANCELLED: [Execution.CANCELLED]
- }
-
- invalid_transitions = {
- Execution.PENDING: [Execution.FAILED,
- Execution.TERMINATED,
- Execution.CANCELLING],
- Execution.STARTED: [Execution.PENDING],
- Execution.CANCELLING: [Execution.PENDING,
- Execution.STARTED],
- Execution.FAILED: [Execution.PENDING,
- Execution.STARTED,
- Execution.TERMINATED,
- Execution.CANCELLED,
- Execution.CANCELLING],
- Execution.TERMINATED: [Execution.PENDING,
- Execution.STARTED,
- Execution.FAILED,
- Execution.CANCELLED,
- Execution.CANCELLING],
- Execution.CANCELLED: [Execution.PENDING,
- Execution.STARTED,
- Execution.FAILED,
- Execution.TERMINATED,
- Execution.CANCELLING],
- }
-
- for current_status, valid_transitioned_statues in valid_transitions.items():
- for transitioned_status in valid_transitioned_statues:
- execution = create_execution(current_status)
- execution.status = transitioned_status
-
- for current_status, invalid_transitioned_statues in invalid_transitions.items():
- for transitioned_status in invalid_transitioned_statues:
- execution = create_execution(current_status)
- with pytest.raises(ValueError):
- execution.status = transitioned_status
-
-
-def test_task_max_attempts_validation():
- def create_task(max_attempts):
- Task(execution_id='eid',
- name='name',
- operation_mapping='',
- inputs={},
- actor=models.get_dependency_node_instance(),
- max_attempts=max_attempts)
- create_task(max_attempts=1)
- create_task(max_attempts=2)
- create_task(max_attempts=Task.INFINITE_RETRIES)
- with pytest.raises(ValueError):
- create_task(max_attempts=0)
- with pytest.raises(ValueError):
- create_task(max_attempts=-2)
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/7b473b63/tests/storage/test_models_api.py
----------------------------------------------------------------------
diff --git a/tests/storage/test_models_api.py b/tests/storage/test_models_api.py
deleted file mode 100644
index 2b92820..0000000
--- a/tests/storage/test_models_api.py
+++ /dev/null
@@ -1,70 +0,0 @@
-# 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 aria.storage import _ModelApi, models
-from aria.storage.exceptions import StorageError
-
-from . import InMemoryModelDriver
-
-
-def test_models_api_base():
- driver = InMemoryModelDriver()
- driver.create('provider_context')
- table = _ModelApi('provider_context', driver, models.ProviderContext)
- assert repr(table) == (
- '{table.name}(driver={table.driver}, '
- 'model={table.model_cls})'.format(table=table))
- provider_context = models.ProviderContext(context={}, name='context_name', id='id')
-
- table.store(provider_context)
- assert table.get('id') == provider_context
-
- assert [i for i in table.iter()] == [provider_context]
- assert [i for i in table] == [provider_context]
-
- table.delete('id')
-
- with pytest.raises(StorageError):
- table.get('id')
-
-
-def test_iterable_model_api():
- driver = InMemoryModelDriver()
- driver.create('deployment_update')
- driver.create('deployment_update_step')
- model_api = _ModelApi('deployment_update', driver, models.DeploymentUpdate)
- deployment_update = models.DeploymentUpdate(
- id='id',
- deployment_id='deployment_id',
- deployment_plan={},
- execution_id='execution_id',
- steps=[models.DeploymentUpdateStep(
- id='step_id',
- action='add',
- entity_type='node',
- entity_id='node_id'
- )]
- )
-
- model_api.store(deployment_update)
- assert [i for i in model_api.iter()] == [deployment_update]
- assert [i for i in model_api] == [deployment_update]
-
- model_api.delete('id')
-
- with pytest.raises(StorageError):
- model_api.get('id')