--- /dev/null
+# Licensed to the Apache ftware 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 os
+import tempfile
+
+import pytest
+
+from aria.storage.filesystem_rapi import FileSystemResourceAPI
+from aria.storage import (
+ exceptions,
+ ResourceStorage
+)
+from . import TestFileSystem
+
+
+class TestResourceStorage(TestFileSystem):
+ def _create(self, storage):
+ storage.register('service_template')
+
+ def _upload(self, storage, tmp_path, id):
+ with open(tmp_path, 'w') as f:
+ f.write('fake context')
+
+ 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.service_template.upload(entry_id=id, source=tmp_dir)
+
+ def _create_storage(self):
+ return ResourceStorage(FileSystemResourceAPI,
+ api_kwargs=dict(directory=self.path))
+
+ def test_name(self):
+ api = FileSystemResourceAPI
+ storage = ResourceStorage(FileSystemResourceAPI,
+ 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['service_template'])
+
+ def test_create(self):
+ storage = self._create_storage()
+ self._create(storage)
+ 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='service_template_id')
+
+ storage_path = os.path.join(
+ self.path,
+ 'service_template',
+ 'service_template_id',
+ os.path.basename(tmpfile_path))
+ assert os.path.exists(storage_path)
+
+ with open(storage_path, 'rb') as f:
+ assert f.read() == 'fake context'
+
+ def test_download_file(self):
+ storage = self._create_storage()
+ 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, 'service_template_id')
+
+ temp_dir = tempfile.mkdtemp(dir=self.path)
+ storage.service_template.download(
+ entry_id='service_template_id',
+ destination=temp_dir,
+ path=tmpfile_name)
+
+ with open(os.path.join(self.path, os.path.join(temp_dir, tmpfile_name))) as f:
+ assert f.read() == 'fake context'
+
+ def test_download_non_existing_file(self):
+ storage = self._create_storage()
+ self._create(storage)
+ with pytest.raises(exceptions.StorageError):
+ 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.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, 'service_template_id')
+
+ 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()
+ self._create(storage)
+ 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='service_template_id')
+
+ destination = os.path.join(
+ self.path,
+ 'service_template',
+ 'service_template_id',
+ os.path.basename(second_level_tmp_dir),
+ os.path.basename(tmp_filename))
+
+ assert os.path.isfile(destination)
+
+ def test_upload_path_in_dir(self):
+ storage = self._create_storage()
+ self._create(storage)
+ 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='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.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,
+ 'service_template',
+ 'service_template_id',
+ os.path.basename(second_level_tmp_dir),
+ os.path.basename(second_update_file)))
+
+ def test_download_dir(self):
+ storage = self._create_storage()
+ self._create(storage)
+ 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='service_template_id')
+
+ temp_destination_dir = tempfile.mkdtemp(dir=self.path)
+ storage.service_template.download(
+ entry_id='service_template_id',
+ destination=temp_destination_dir)
+
+ destination_file_path = os.path.join(
+ temp_destination_dir,
+ os.path.basename(second_level_tmp_dir),
+ os.path.basename(tmp_filename))
+
+ assert os.path.isfile(destination_file_path)
+
+ with open(destination_file_path) as f:
+ assert f.read() == 'fake context'
+
+ def test_data_dir(self):
+ storage = self._create_storage()
+ self._create(storage)
+
+ tmp_dir = tempfile.mkdtemp(suffix=self.__class__.__name__, dir=self.path)
+ tempfile.mkstemp(dir=tmp_dir)
+ tempfile.mkstemp(dir=tmp_dir)
+
+ storage.service_template.upload(entry_id='service_template_id', source=tmp_dir)
+
+ with pytest.raises(exceptions.StorageError):
+ 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, 'service_template_id')
+ tmpfile2_path = tempfile.mkstemp(suffix=self.__class__.__name__, dir=self.path)[1]
+ self._upload(storage, tmpfile2_path, 'service_template_id')
+
+ # deleting the first resource and expecting an error on read
+ storage.service_template.delete(entry_id='service_template_id',
+ path=os.path.basename(tmpfile_path))
+ with pytest.raises(exceptions.StorageError):
+ 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.service_template.read(
+ entry_id='service_template_id',
+ path=os.path.basename(tmpfile2_path)) == 'fake context'
+
+ def test_delete_directory(self):
+ storage = self._create_storage()
+ self._create(storage)
+ temp_destination_dir = tempfile.mkdtemp(dir=self.path)
+
+ 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='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.service_template.read(
+ entry_id='service_template_id',
+ path=file_path_in_dir) == 'fake context'
+ 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.service_template.delete(
+ entry_id='service_template_id',
+ path=os.path.basename(second_level_tmp_dir))
+ with pytest.raises(exceptions.StorageError):
+ assert storage.service_template.read(
+ entry_id='service_template_id',
+ path=file_path_in_dir) == 'fake context'
+ with pytest.raises(exceptions.StorageError):
+ storage.service_template.download(
+ entry_id='service_template_id',
+ path=os.path.basename(second_level_tmp_dir),
+ destination=temp_destination_dir)
+
+ def test_delete_all_resources(self):
+ storage = self._create_storage()
+ self._create(storage)
+ temp_destination_dir = tempfile.mkdtemp(dir=self.path)
+
+ 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='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.service_template.delete(entry_id='service_template_id')
+ with pytest.raises(exceptions.StorageError):
+ assert storage.service_template.read(
+ entry_id='service_template_id',
+ path=file_path_in_dir) == 'fake context'
+ with pytest.raises(exceptions.StorageError):
+ storage.service_template.download(
+ entry_id='service_template_id',
+ path=os.path.basename(second_level_tmp_dir),
+ destination=temp_destination_dir)
+
+ def test_delete_nonexisting_resource(self):
+ storage = self._create_storage()
+ self._create(storage)
+ # deleting a nonexisting resource - no effect is expected to happen
+ assert storage.service_template.delete(entry_id='service_template_id',
+ path='fake-file') is False
Code Review / multicloud / azure.git / blobdiff