azure/aria/aria-extension-cloudify/src/aria/tests/storage/test_resource_storage.py

   1 # Licensed to the Apache ftware 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 os
  17 import tempfile
  18
  19 import pytest
  20
  21 from aria.storage.filesystem_rapi import FileSystemResourceAPI
  22 from aria.storage import (
  23     exceptions,
  24     ResourceStorage
  25 )
  26 from . import TestFileSystem
  27
  28
  29 class TestResourceStorage(TestFileSystem):
  30     def _create(self, storage):
  31         storage.register('service_template')
  32
  33     def _upload(self, storage, tmp_path, id):
  34         with open(tmp_path, 'w') as f:
  35             f.write('fake context')
  36
  37         storage.service_template.upload(entry_id=id, source=tmp_path)
  38
  39     def _upload_dir(self, storage, tmp_dir, tmp_file_name, id):
  40         file_source = os.path.join(tmp_dir, tmp_file_name)
  41         with open(file_source, 'w') as f:
  42             f.write('fake context')
  43
  44         storage.service_template.upload(entry_id=id, source=tmp_dir)
  45
  46     def _create_storage(self):
  47         return ResourceStorage(FileSystemResourceAPI,
  48                                api_kwargs=dict(directory=self.path))
  49
  50     def test_name(self):
  51         api = FileSystemResourceAPI
  52         storage = ResourceStorage(FileSystemResourceAPI,
  53                                   items=['service_template'],
  54                                   api_kwargs=dict(directory=self.path))
  55         assert repr(storage) == 'ResourceStorage(api={api})'.format(api=api)
  56         assert 'directory={resource_dir}'.format(resource_dir=self.path) in \
  57                repr(storage.registered['service_template'])
  58
  59     def test_create(self):
  60         storage = self._create_storage()
  61         self._create(storage)
  62         assert os.path.exists(os.path.join(self.path, 'service_template'))
  63
  64     def test_upload_file(self):
  65         storage = ResourceStorage(FileSystemResourceAPI, api_kwargs=dict(directory=self.path))
  66         self._create(storage)
  67         tmpfile_path = tempfile.mkstemp(suffix=self.__class__.__name__, dir=self.path)[1]
  68         self._upload(storage, tmpfile_path, id='service_template_id')
  69
  70         storage_path = os.path.join(
  71             self.path,
  72             'service_template',
  73             'service_template_id',
  74             os.path.basename(tmpfile_path))
  75         assert os.path.exists(storage_path)
  76
  77         with open(storage_path, 'rb') as f:
  78             assert f.read() == 'fake context'
  79
  80     def test_download_file(self):
  81         storage = self._create_storage()
  82         self._create(storage)
  83         tmpfile_path = tempfile.mkstemp(suffix=self.__class__.__name__, dir=self.path)[1]
  84         tmpfile_name = os.path.basename(tmpfile_path)
  85         self._upload(storage, tmpfile_path, 'service_template_id')
  86
  87         temp_dir = tempfile.mkdtemp(dir=self.path)
  88         storage.service_template.download(
  89             entry_id='service_template_id',
  90             destination=temp_dir,
  91             path=tmpfile_name)
  92
  93         with open(os.path.join(self.path, os.path.join(temp_dir, tmpfile_name))) as f:
  94             assert f.read() == 'fake context'
  95
  96     def test_download_non_existing_file(self):
  97         storage = self._create_storage()
  98         self._create(storage)
  99         with pytest.raises(exceptions.StorageError):
 100             storage.service_template.download(entry_id='service_template_id', destination='',
 101                                               path='fake_path')
 102
 103     def test_data_non_existing_file(self):
 104         storage = self._create_storage()
 105         self._create(storage)
 106         with pytest.raises(exceptions.StorageError):
 107             storage.service_template.read(entry_id='service_template_id', path='fake_path')
 108
 109     def test_data_file(self):
 110         storage = self._create_storage()
 111         self._create(storage)
 112         tmpfile_path = tempfile.mkstemp(suffix=self.__class__.__name__, dir=self.path)[1]
 113         self._upload(storage, tmpfile_path, 'service_template_id')
 114
 115         assert storage.service_template.read(entry_id='service_template_id',
 116                                              path=os.path.basename(tmpfile_path)) == 'fake context'
 117
 118     def test_upload_dir(self):
 119         storage = self._create_storage()
 120         self._create(storage)
 121         tmp_dir = tempfile.mkdtemp(suffix=self.__class__.__name__, dir=self.path)
 122         second_level_tmp_dir = tempfile.mkdtemp(dir=tmp_dir)
 123         tmp_filename = tempfile.mkstemp(dir=second_level_tmp_dir)[1]
 124         self._upload_dir(storage, tmp_dir, tmp_filename, id='service_template_id')
 125
 126         destination = os.path.join(
 127             self.path,
 128             'service_template',
 129             'service_template_id',
 130             os.path.basename(second_level_tmp_dir),
 131             os.path.basename(tmp_filename))
 132
 133         assert os.path.isfile(destination)
 134
 135     def test_upload_path_in_dir(self):
 136         storage = self._create_storage()
 137         self._create(storage)
 138         tmp_dir = tempfile.mkdtemp(suffix=self.__class__.__name__, dir=self.path)
 139         second_level_tmp_dir = tempfile.mkdtemp(dir=tmp_dir)
 140         tmp_filename = tempfile.mkstemp(dir=second_level_tmp_dir)[1]
 141         self._upload_dir(storage, tmp_dir, tmp_filename, id='service_template_id')
 142
 143         second_update_file = tempfile.mkstemp(dir=self.path)[1]
 144         with open(second_update_file, 'w') as f:
 145             f.write('fake context2')
 146
 147         storage.service_template.upload(
 148             entry_id='service_template_id',
 149             source=second_update_file,
 150             path=os.path.basename(second_level_tmp_dir))
 151
 152         assert os.path.isfile(os.path.join(
 153             self.path,
 154             'service_template',
 155             'service_template_id',
 156             os.path.basename(second_level_tmp_dir),
 157             os.path.basename(second_update_file)))
 158
 159     def test_download_dir(self):
 160         storage = self._create_storage()
 161         self._create(storage)
 162         tmp_dir = tempfile.mkdtemp(suffix=self.__class__.__name__, dir=self.path)
 163         second_level_tmp_dir = tempfile.mkdtemp(dir=tmp_dir)
 164         tmp_filename = tempfile.mkstemp(dir=second_level_tmp_dir)[1]
 165         self._upload_dir(storage, tmp_dir, tmp_filename, id='service_template_id')
 166
 167         temp_destination_dir = tempfile.mkdtemp(dir=self.path)
 168         storage.service_template.download(
 169             entry_id='service_template_id',
 170             destination=temp_destination_dir)
 171
 172         destination_file_path = os.path.join(
 173             temp_destination_dir,
 174             os.path.basename(second_level_tmp_dir),
 175             os.path.basename(tmp_filename))
 176
 177         assert os.path.isfile(destination_file_path)
 178
 179         with open(destination_file_path) as f:
 180             assert f.read() == 'fake context'
 181
 182     def test_data_dir(self):
 183         storage = self._create_storage()
 184         self._create(storage)
 185
 186         tmp_dir = tempfile.mkdtemp(suffix=self.__class__.__name__, dir=self.path)
 187         tempfile.mkstemp(dir=tmp_dir)
 188         tempfile.mkstemp(dir=tmp_dir)
 189
 190         storage.service_template.upload(entry_id='service_template_id', source=tmp_dir)
 191
 192         with pytest.raises(exceptions.StorageError):
 193             storage.service_template.read(entry_id='service_template_id', path='')
 194
 195     def test_delete_resource(self):
 196         storage = self._create_storage()
 197         self._create(storage)
 198         tmpfile_path = tempfile.mkstemp(suffix=self.__class__.__name__, dir=self.path)[1]
 199         self._upload(storage, tmpfile_path, 'service_template_id')
 200         tmpfile2_path = tempfile.mkstemp(suffix=self.__class__.__name__, dir=self.path)[1]
 201         self._upload(storage, tmpfile2_path, 'service_template_id')
 202
 203         # deleting the first resource and expecting an error on read
 204         storage.service_template.delete(entry_id='service_template_id',
 205                                         path=os.path.basename(tmpfile_path))
 206         with pytest.raises(exceptions.StorageError):
 207             storage.service_template.read(entry_id='service_template_id',
 208                                           path=os.path.basename(tmpfile_path))
 209         # the second resource should still be available for reading
 210         assert storage.service_template.read(
 211             entry_id='service_template_id',
 212             path=os.path.basename(tmpfile2_path)) == 'fake context'
 213
 214     def test_delete_directory(self):
 215         storage = self._create_storage()
 216         self._create(storage)
 217         temp_destination_dir = tempfile.mkdtemp(dir=self.path)
 218
 219         tmp_dir = tempfile.mkdtemp(suffix=self.__class__.__name__, dir=self.path)
 220         second_level_tmp_dir = tempfile.mkdtemp(dir=tmp_dir)
 221         tmp_filename = tempfile.mkstemp(dir=second_level_tmp_dir)[1]
 222         self._upload_dir(storage, tmp_dir, tmp_filename, id='service_template_id')
 223         file_path_in_dir = os.path.join(
 224             os.path.basename(second_level_tmp_dir),
 225             os.path.basename(tmp_filename))
 226
 227         # should be able to read the file and download the directory..
 228         assert storage.service_template.read(
 229             entry_id='service_template_id',
 230             path=file_path_in_dir) == 'fake context'
 231         storage.service_template.download(
 232             entry_id='service_template_id',
 233             path=os.path.basename(second_level_tmp_dir),
 234             destination=temp_destination_dir)
 235
 236         # after deletion, the file and directory should both be gone
 237         storage.service_template.delete(
 238             entry_id='service_template_id',
 239             path=os.path.basename(second_level_tmp_dir))
 240         with pytest.raises(exceptions.StorageError):
 241             assert storage.service_template.read(
 242                 entry_id='service_template_id',
 243                 path=file_path_in_dir) == 'fake context'
 244         with pytest.raises(exceptions.StorageError):
 245             storage.service_template.download(
 246                 entry_id='service_template_id',
 247                 path=os.path.basename(second_level_tmp_dir),
 248                 destination=temp_destination_dir)
 249
 250     def test_delete_all_resources(self):
 251         storage = self._create_storage()
 252         self._create(storage)
 253         temp_destination_dir = tempfile.mkdtemp(dir=self.path)
 254
 255         tmp_dir = tempfile.mkdtemp(suffix=self.__class__.__name__, dir=self.path)
 256         second_level_tmp_dir = tempfile.mkdtemp(dir=tmp_dir)
 257         tmp_filename = tempfile.mkstemp(dir=second_level_tmp_dir)[1]
 258         self._upload_dir(storage, tmp_dir, tmp_filename, id='service_template_id')
 259         file_path_in_dir = os.path.join(
 260             os.path.basename(second_level_tmp_dir),
 261             os.path.basename(tmp_filename))
 262
 263         # deleting without specifying a path - delete all resources of this entry
 264         storage.service_template.delete(entry_id='service_template_id')
 265         with pytest.raises(exceptions.StorageError):
 266             assert storage.service_template.read(
 267                 entry_id='service_template_id',
 268                 path=file_path_in_dir) == 'fake context'
 269         with pytest.raises(exceptions.StorageError):
 270             storage.service_template.download(
 271                 entry_id='service_template_id',
 272                 path=os.path.basename(second_level_tmp_dir),
 273                 destination=temp_destination_dir)
 274
 275     def test_delete_nonexisting_resource(self):
 276         storage = self._create_storage()
 277         self._create(storage)
 278         # deleting a nonexisting resource - no effect is expected to happen
 279         assert storage.service_template.delete(entry_id='service_template_id',
 280                                                path='fake-file') is False