ariatosca-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From mxm...@apache.org
Subject [2/4] incubator-ariatosca git commit: fixed the models and some tests
Date Sun, 27 Nov 2016 16:16:39 GMT
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')


Mime
View raw message