Repository: incubator-ariatosca Updated Branches: refs/heads/master 95177d0f7 -> 9841ca4ae
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/9841ca4a/tests/storage/test_resource_storage.py ---------------------------------------------------------------------- diff --git a/tests/storage/test_resource_storage.py b/tests/storage/test_resource_storage.py index 4d01a88..efacb2e 100644 --- a/tests/storage/test_resource_storage.py +++ b/tests/storage/test_resource_storage.py @@ -28,20 +28,20 @@ from . import TestFileSystem class TestResourceStorage(TestFileSystem): def _create(self, storage): - storage.register('blueprint') + storage.register('service_template') def _upload(self, storage, tmp_path, id): with open(tmp_path, 'w') as f: f.write('fake context') - storage.blueprint.upload(entry_id=id, source=tmp_path) + storage.service_template.upload(entry_id=id, source=tmp_path) def _upload_dir(self, storage, tmp_dir, tmp_file_name, id): file_source = os.path.join(tmp_dir, tmp_file_name) with open(file_source, 'w') as f: f.write('fake context') - storage.blueprint.upload(entry_id=id, source=tmp_dir) + storage.service_template.upload(entry_id=id, source=tmp_dir) def _create_storage(self): return ResourceStorage(FileSystemResourceAPI, @@ -50,27 +50,27 @@ class TestResourceStorage(TestFileSystem): def test_name(self): api = FileSystemResourceAPI storage = ResourceStorage(FileSystemResourceAPI, - items=['blueprint'], + items=['service_template'], api_kwargs=dict(directory=self.path)) assert repr(storage) == 'ResourceStorage(api={api})'.format(api=api) assert 'directory={resource_dir}'.format(resource_dir=self.path) in \ - repr(storage.registered['blueprint']) + repr(storage.registered['service_template']) def test_create(self): storage = self._create_storage() self._create(storage) - assert os.path.exists(os.path.join(self.path, 'blueprint')) + assert os.path.exists(os.path.join(self.path, 'service_template')) def test_upload_file(self): storage = ResourceStorage(FileSystemResourceAPI, api_kwargs=dict(directory=self.path)) self._create(storage) tmpfile_path = tempfile.mkstemp(suffix=self.__class__.__name__, dir=self.path)[1] - self._upload(storage, tmpfile_path, id='blueprint_id') + self._upload(storage, tmpfile_path, id='service_template_id') storage_path = os.path.join( self.path, - 'blueprint', - 'blueprint_id', + 'service_template', + 'service_template_id', os.path.basename(tmpfile_path)) assert os.path.exists(storage_path) @@ -82,11 +82,11 @@ class TestResourceStorage(TestFileSystem): self._create(storage) tmpfile_path = tempfile.mkstemp(suffix=self.__class__.__name__, dir=self.path)[1] tmpfile_name = os.path.basename(tmpfile_path) - self._upload(storage, tmpfile_path, 'blueprint_id') + self._upload(storage, tmpfile_path, 'service_template_id') temp_dir = tempfile.mkdtemp(dir=self.path) - storage.blueprint.download( - entry_id='blueprint_id', + storage.service_template.download( + entry_id='service_template_id', destination=temp_dir, path=tmpfile_name) @@ -97,23 +97,23 @@ class TestResourceStorage(TestFileSystem): storage = self._create_storage() self._create(storage) with pytest.raises(exceptions.StorageError): - storage.blueprint.download(entry_id='blueprint_id', destination='', path='fake_path') + storage.service_template.download(entry_id='service_template_id', destination='', + path='fake_path') def test_data_non_existing_file(self): storage = self._create_storage() self._create(storage) with pytest.raises(exceptions.StorageError): - storage.blueprint.read(entry_id='blueprint_id', path='fake_path') + storage.service_template.read(entry_id='service_template_id', path='fake_path') def test_data_file(self): storage = self._create_storage() self._create(storage) tmpfile_path = tempfile.mkstemp(suffix=self.__class__.__name__, dir=self.path)[1] - self._upload(storage, tmpfile_path, 'blueprint_id') + self._upload(storage, tmpfile_path, 'service_template_id') - assert storage.blueprint.read( - entry_id='blueprint_id', - path=os.path.basename(tmpfile_path)) == 'fake context' + assert storage.service_template.read(entry_id='service_template_id', + path=os.path.basename(tmpfile_path)) == 'fake context' def test_upload_dir(self): storage = self._create_storage() @@ -121,12 +121,12 @@ class TestResourceStorage(TestFileSystem): tmp_dir = tempfile.mkdtemp(suffix=self.__class__.__name__, dir=self.path) second_level_tmp_dir = tempfile.mkdtemp(dir=tmp_dir) tmp_filename = tempfile.mkstemp(dir=second_level_tmp_dir)[1] - self._upload_dir(storage, tmp_dir, tmp_filename, id='blueprint_id') + self._upload_dir(storage, tmp_dir, tmp_filename, id='service_template_id') destination = os.path.join( self.path, - 'blueprint', - 'blueprint_id', + 'service_template', + 'service_template_id', os.path.basename(second_level_tmp_dir), os.path.basename(tmp_filename)) @@ -138,21 +138,21 @@ class TestResourceStorage(TestFileSystem): tmp_dir = tempfile.mkdtemp(suffix=self.__class__.__name__, dir=self.path) second_level_tmp_dir = tempfile.mkdtemp(dir=tmp_dir) tmp_filename = tempfile.mkstemp(dir=second_level_tmp_dir)[1] - self._upload_dir(storage, tmp_dir, tmp_filename, id='blueprint_id') + self._upload_dir(storage, tmp_dir, tmp_filename, id='service_template_id') second_update_file = tempfile.mkstemp(dir=self.path)[1] with open(second_update_file, 'w') as f: f.write('fake context2') - storage.blueprint.upload( - entry_id='blueprint_id', + storage.service_template.upload( + entry_id='service_template_id', source=second_update_file, path=os.path.basename(second_level_tmp_dir)) assert os.path.isfile(os.path.join( self.path, - 'blueprint', - 'blueprint_id', + 'service_template', + 'service_template_id', os.path.basename(second_level_tmp_dir), os.path.basename(second_update_file))) @@ -162,11 +162,11 @@ class TestResourceStorage(TestFileSystem): tmp_dir = tempfile.mkdtemp(suffix=self.__class__.__name__, dir=self.path) second_level_tmp_dir = tempfile.mkdtemp(dir=tmp_dir) tmp_filename = tempfile.mkstemp(dir=second_level_tmp_dir)[1] - self._upload_dir(storage, tmp_dir, tmp_filename, id='blueprint_id') + self._upload_dir(storage, tmp_dir, tmp_filename, id='service_template_id') temp_destination_dir = tempfile.mkdtemp(dir=self.path) - storage.blueprint.download( - entry_id='blueprint_id', + storage.service_template.download( + entry_id='service_template_id', destination=temp_destination_dir) destination_file_path = os.path.join( @@ -187,26 +187,28 @@ class TestResourceStorage(TestFileSystem): tempfile.mkstemp(dir=tmp_dir) tempfile.mkstemp(dir=tmp_dir) - storage.blueprint.upload(entry_id='blueprint_id', source=tmp_dir) + storage.service_template.upload(entry_id='service_template_id', source=tmp_dir) with pytest.raises(exceptions.StorageError): - storage.blueprint.read(entry_id='blueprint_id', path='') + storage.service_template.read(entry_id='service_template_id', path='') def test_delete_resource(self): storage = self._create_storage() self._create(storage) tmpfile_path = tempfile.mkstemp(suffix=self.__class__.__name__, dir=self.path)[1] - self._upload(storage, tmpfile_path, 'blueprint_id') + self._upload(storage, tmpfile_path, 'service_template_id') tmpfile2_path = tempfile.mkstemp(suffix=self.__class__.__name__, dir=self.path)[1] - self._upload(storage, tmpfile2_path, 'blueprint_id') + self._upload(storage, tmpfile2_path, 'service_template_id') # deleting the first resource and expecting an error on read - storage.blueprint.delete(entry_id='blueprint_id', path=os.path.basename(tmpfile_path)) + storage.service_template.delete(entry_id='service_template_id', + path=os.path.basename(tmpfile_path)) with pytest.raises(exceptions.StorageError): - storage.blueprint.read(entry_id='blueprint_id', path=os.path.basename(tmpfile_path)) + storage.service_template.read(entry_id='service_template_id', + path=os.path.basename(tmpfile_path)) # the second resource should still be available for reading - assert storage.blueprint.read( - entry_id='blueprint_id', + assert storage.service_template.read( + entry_id='service_template_id', path=os.path.basename(tmpfile2_path)) == 'fake context' def test_delete_directory(self): @@ -217,31 +219,31 @@ class TestResourceStorage(TestFileSystem): tmp_dir = tempfile.mkdtemp(suffix=self.__class__.__name__, dir=self.path) second_level_tmp_dir = tempfile.mkdtemp(dir=tmp_dir) tmp_filename = tempfile.mkstemp(dir=second_level_tmp_dir)[1] - self._upload_dir(storage, tmp_dir, tmp_filename, id='blueprint_id') + self._upload_dir(storage, tmp_dir, tmp_filename, id='service_template_id') file_path_in_dir = os.path.join( os.path.basename(second_level_tmp_dir), os.path.basename(tmp_filename)) # should be able to read the file and download the directory.. - assert storage.blueprint.read( - entry_id='blueprint_id', + assert storage.service_template.read( + entry_id='service_template_id', path=file_path_in_dir) == 'fake context' - storage.blueprint.download( - entry_id='blueprint_id', + storage.service_template.download( + entry_id='service_template_id', path=os.path.basename(second_level_tmp_dir), destination=temp_destination_dir) # after deletion, the file and directory should both be gone - storage.blueprint.delete( - entry_id='blueprint_id', + storage.service_template.delete( + entry_id='service_template_id', path=os.path.basename(second_level_tmp_dir)) with pytest.raises(exceptions.StorageError): - assert storage.blueprint.read( - entry_id='blueprint_id', + assert storage.service_template.read( + entry_id='service_template_id', path=file_path_in_dir) == 'fake context' with pytest.raises(exceptions.StorageError): - storage.blueprint.download( - entry_id='blueprint_id', + storage.service_template.download( + entry_id='service_template_id', path=os.path.basename(second_level_tmp_dir), destination=temp_destination_dir) @@ -253,20 +255,20 @@ class TestResourceStorage(TestFileSystem): tmp_dir = tempfile.mkdtemp(suffix=self.__class__.__name__, dir=self.path) second_level_tmp_dir = tempfile.mkdtemp(dir=tmp_dir) tmp_filename = tempfile.mkstemp(dir=second_level_tmp_dir)[1] - self._upload_dir(storage, tmp_dir, tmp_filename, id='blueprint_id') + self._upload_dir(storage, tmp_dir, tmp_filename, id='service_template_id') file_path_in_dir = os.path.join( os.path.basename(second_level_tmp_dir), os.path.basename(tmp_filename)) # deleting without specifying a path - delete all resources of this entry - storage.blueprint.delete(entry_id='blueprint_id') + storage.service_template.delete(entry_id='service_template_id') with pytest.raises(exceptions.StorageError): - assert storage.blueprint.read( - entry_id='blueprint_id', + assert storage.service_template.read( + entry_id='service_template_id', path=file_path_in_dir) == 'fake context' with pytest.raises(exceptions.StorageError): - storage.blueprint.download( - entry_id='blueprint_id', + storage.service_template.download( + entry_id='service_template_id', path=os.path.basename(second_level_tmp_dir), destination=temp_destination_dir) @@ -274,4 +276,5 @@ class TestResourceStorage(TestFileSystem): storage = self._create_storage() self._create(storage) # deleting a nonexisting resource - no effect is expected to happen - assert storage.blueprint.delete(entry_id='blueprint_id', path='fake-file') is False + assert storage.service_template.delete(entry_id='service_template_id', + path='fake-file') is False http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/9841ca4a/tests/storage/test_structures.py ---------------------------------------------------------------------- diff --git a/tests/storage/test_structures.py b/tests/storage/test_structures.py deleted file mode 100644 index 27e99d7..0000000 --- a/tests/storage/test_structures.py +++ /dev/null @@ -1,218 +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 - -import sqlalchemy - -from aria.storage import ( - ModelStorage, - sql_mapi, - exceptions, - modeling, -) -from aria.storage.modeling import type - -from ..storage import release_sqlite_storage, structure, init_inmemory_model_storage -from . import MockModel -from ..mock import ( - models, - context as mock_context -) - - [email protected] -def storage(): - base_storage = ModelStorage(sql_mapi.SQLAlchemyModelAPI, - initiator=init_inmemory_model_storage) - base_storage.register(MockModel) - yield base_storage - release_sqlite_storage(base_storage) - - [email protected](scope='module', autouse=True) -def module_cleanup(): - modeling.model.aria_declarative_base.metadata.remove(MockModel.__table__) #pylint: disable=no-member - - [email protected] -def context(tmpdir): - ctx = mock_context.simple(str(tmpdir)) - yield ctx - release_sqlite_storage(ctx.model) - - -def test_inner_dict_update(storage): - inner_dict = {'inner_value': 1} - - mock_model = MockModel(model_dict={'inner_dict': inner_dict, 'value': 0}) - storage.mock_model.put(mock_model) - - storage_mm = storage.mock_model.get(mock_model.id) - assert storage_mm == mock_model - - storage_mm.model_dict['inner_dict']['inner_value'] = 2 - storage_mm.model_dict['value'] = -1 - storage.mock_model.update(storage_mm) - storage_mm = storage.mock_model.get(storage_mm.id) - - assert storage_mm.model_dict['inner_dict']['inner_value'] == 2 - assert storage_mm.model_dict['value'] == -1 - - -def test_inner_list_update(storage): - mock_model = MockModel(model_list=[0, [1]]) - storage.mock_model.put(mock_model) - - storage_mm = storage.mock_model.get(mock_model.id) - assert storage_mm == mock_model - - storage_mm.model_list[1][0] = 'new_inner_value' - storage_mm.model_list[0] = 'new_value' - storage.mock_model.update(storage_mm) - storage_mm = storage.mock_model.get(storage_mm.id) - - assert storage_mm.model_list[1][0] == 'new_inner_value' - assert storage_mm.model_list[0] == 'new_value' - - -def test_model_to_dict(context): - service_instance = context.service_instance - service_instance = service_instance.to_dict() - - expected_keys = [ - 'description', - '_metadata', - 'created_at', - 'permalink', - 'policy_triggers', - 'policy_types', - 'scaling_groups', - 'updated_at', - 'workflows', - ] - - for expected_key in expected_keys: - assert expected_key in service_instance - - -def test_relationship_model_ordering(context): - service_instance = context.model.service_instance.get_by_name(models.DEPLOYMENT_NAME) - source_node = context.model.node.get_by_name(models.DEPENDENT_NODE_INSTANCE_NAME) - target_node = context.model.node.get_by_name(models.DEPENDENCY_NODE_INSTANCE_NAME) - new_node_template = modeling.model.NodeTemplate( - name='new_node', - type_name='test_node_type', - type_hierarchy=[], - default_instances=1, - min_instances=1, - max_instances=1, - service_template=service_instance.service_template - ) - new_node = modeling.model.Node( - name='new_node_instance', - runtime_properties={}, - service_instance=service_instance, - node_template=new_node_template, - state='', - scaling_groups=[] - ) - - source_to_new_relationship = modeling.model.Relationship( - target_node=new_node, - source_node=source_node, - ) - - new_to_target_relationship = modeling.model.Relationship( - source_node=new_node, - target_node=target_node, - ) - - context.model.node_template.put(new_node_template) - context.model.node.put(new_node) - context.model.relationship.put(source_to_new_relationship) - context.model.relationship.put(new_to_target_relationship) - - def flip_and_assert(node, direction): - """ - Reversed the order of relationships and assert effects took place. - :param node: the node instance to operatate on - :param direction: the type of relationships to flip (inbound/outbount) - :return: - """ - assert direction in ('inbound', 'outbound') - - relationships = getattr(node, direction + '_relationships').all() - assert len(relationships) == 2 - - reversed_relationship_instances = list(reversed(relationships)) - assert relationships != reversed_relationship_instances - - relationships[:] = reversed_relationship_instances - context.model.node.update(node) - assert relationships == reversed_relationship_instances - - flip_and_assert(source_node, 'outbound') - flip_and_assert(target_node, 'inbound') - - -class StrictClass(modeling.model.aria_declarative_base, structure.ModelMixin): - __tablename__ = 'strict_class' - - strict_dict = sqlalchemy.Column(type.StrictDict(basestring, basestring)) - strict_list = sqlalchemy.Column(type.StrictList(basestring)) - - -def test_strict_dict(): - - strict_class = StrictClass() - - def assert_strict(sc): - with pytest.raises(exceptions.StorageError): - sc.strict_dict = {'key': 1} - - with pytest.raises(exceptions.StorageError): - sc.strict_dict = {1: 'value'} - - with pytest.raises(exceptions.StorageError): - sc.strict_dict = {1: 1} - - assert_strict(strict_class) - strict_class.strict_dict = {'key': 'value'} - assert strict_class.strict_dict == {'key': 'value'} - - assert_strict(strict_class) - with pytest.raises(exceptions.StorageError): - strict_class.strict_dict['key'] = 1 - with pytest.raises(exceptions.StorageError): - strict_class.strict_dict[1] = 'value' - with pytest.raises(exceptions.StorageError): - strict_class.strict_dict[1] = 1 - - -def test_strict_list(): - strict_class = StrictClass() - - def assert_strict(sc): - with pytest.raises(exceptions.StorageError): - sc.strict_list = [1] - - assert_strict(strict_class) - strict_class.strict_list = ['item'] - assert strict_class.strict_list == ['item'] - - assert_strict(strict_class) - with pytest.raises(exceptions.StorageError): - strict_class.strict_list[0] = 1
