osm_common/tests/test_fslocal.py

   1 import io
   2 import logging
   3 import http
   4 import os
   5 import pytest
   6 import tarfile
   7 import tempfile
   8 import uuid
   9 import shutil
  10
  11 from osm_common.fsbase import FsException
  12 from osm_common.fslocal import FsLocal
  13
  14 __author__ = "Eduardo Sousa <eduardosousa@av.it.pt>"
  15
  16 def valid_path():
  17     return tempfile.gettempdir() + '/'
  18
  19 def invalid_path():
  20     return '/#tweeter/'
  21
  22 @pytest.fixture
  23 def fs_local():
  24     fs = FsLocal()
  25     fs.fs_connect({'path': valid_path()})
  26
  27     return fs
  28
  29 def fs_connect_exception_message(path):
  30     return "storage exception Invalid configuration param at '[storage]': path '{}' does not exist".format(path)
  31
  32 def file_open_file_not_found_exception(storage):
  33     f = storage if isinstance(storage, str) else '/'.join(storage)
  34     return "storage exception File {} does not exist".format(f)
  35
  36 def file_open_io_exception(storage):
  37     f = storage if isinstance(storage, str) else '/'.join(storage)
  38     return "storage exception File {} cannot be opened".format(f)
  39
  40 def dir_ls_not_a_directory_exception(storage):
  41     f = storage if isinstance(storage, str) else '/'.join(storage)
  42     return "storage exception File {} does not exist".format(f)
  43
  44 def dir_ls_io_exception(storage):
  45     f = storage if isinstance(storage, str) else '/'.join(storage)
  46     return "storage exception File {} cannot be opened".format(f)
  47
  48 def file_delete_exception_message(storage):
  49     return "storage exception File {} does not exist".format(storage)
  50
  51 def test_constructor_without_logger():
  52     fs = FsLocal()
  53
  54     assert fs.logger == logging.getLogger('fs')
  55     assert fs.path is None
  56
  57 def test_constructor_with_logger():
  58     logger_name = 'fs_local'
  59
  60     fs = FsLocal(logger_name=logger_name)
  61
  62     assert fs.logger == logging.getLogger(logger_name)
  63     assert fs.path is None
  64
  65 @pytest.mark.parametrize("config, exp_logger, exp_path", [
  66     ({'logger_name': 'fs_local', 'path': valid_path()}, 'fs_local', valid_path()),
  67     ({'logger_name': 'fs_local', 'path': valid_path()[:-1]}, 'fs_local', valid_path()),
  68     ({'path': valid_path()}, 'fs', valid_path()),
  69     ({'path': valid_path()[:-1]}, 'fs', valid_path())])
  70 def test_fs_connect_with_valid_config(config, exp_logger, exp_path):
  71     fs = FsLocal()
  72     fs.fs_connect(config)
  73
  74     assert fs.logger == logging.getLogger(exp_logger)
  75     assert fs.path == exp_path
  76
  77 @pytest.mark.parametrize("config, exp_exception_message", [
  78     ({'logger_name': 'fs_local', 'path': invalid_path()}, fs_connect_exception_message(invalid_path())),
  79     ({'logger_name': 'fs_local', 'path': invalid_path()[:-1]}, fs_connect_exception_message(invalid_path()[:-1])),
  80     ({'path': invalid_path()}, fs_connect_exception_message(invalid_path())),
  81     ({'path': invalid_path()[:-1]}, fs_connect_exception_message(invalid_path()[:-1]))])
  82 def test_fs_connect_with_invalid_path(config, exp_exception_message):
  83     fs = FsLocal()
  84
  85     with pytest.raises(FsException) as excinfo:
  86         fs.fs_connect(config)
  87     assert str(excinfo.value) == exp_exception_message
  88
  89 def test_mkdir_with_valid_path(fs_local):
  90     folder_name = str(uuid.uuid4())
  91     folder_path = valid_path() + folder_name
  92
  93     fs_local.mkdir(folder_name)
  94
  95     assert os.path.exists(folder_path)
  96
  97     os.rmdir(folder_path)
  98
  99 def test_mkdir_with_exception(fs_local):
 100     folder_name = str(uuid.uuid4())
 101     folder_path = valid_path() + folder_name
 102     os.mkdir(folder_path)
 103
 104     with pytest.raises(FsException) as excinfo:
 105         fs_local.mkdir(folder_name)
 106     assert excinfo.value.http_code == http.HTTPStatus.INTERNAL_SERVER_ERROR
 107
 108     os.rmdir(folder_path)
 109
 110 @pytest.mark.parametrize("storage, mode, expected", [
 111     (str(uuid.uuid4()), 'file', False),
 112     ([str(uuid.uuid4())], 'file', False),
 113     (str(uuid.uuid4()), 'dir', False),
 114     ([str(uuid.uuid4())], 'dir', False)])
 115 def test_file_exists_returns_false(fs_local, storage, mode, expected):
 116     assert fs_local.file_exists(storage, mode) == expected
 117
 118 @pytest.mark.parametrize("storage, mode, expected", [
 119     (str(uuid.uuid4()), 'file', True),
 120     ([str(uuid.uuid4())], 'file', True),
 121     (str(uuid.uuid4()), 'dir', True),
 122     ([str(uuid.uuid4())], 'dir', True)])
 123 def test_file_exists_returns_true(fs_local, storage, mode, expected):
 124     path = valid_path() + storage if isinstance(storage, str) else valid_path() + storage[0]
 125
 126     if mode == 'file':
 127         os.mknod(path)
 128     elif mode == 'dir':
 129         os.mkdir(path)
 130
 131     assert fs_local.file_exists(storage, mode) == expected
 132
 133     if mode == 'file':
 134         os.remove(path)
 135     elif mode == 'dir':
 136         os.rmdir(path)
 137
 138 @pytest.mark.parametrize("storage, mode", [
 139     (str(uuid.uuid4()), 'file'),
 140     ([str(uuid.uuid4())], 'file'),
 141     (str(uuid.uuid4()), 'dir'),
 142     ([str(uuid.uuid4())], 'dir')])
 143 def test_file_size(fs_local, storage, mode):
 144     path = valid_path() + storage if isinstance(storage, str) else valid_path() + storage[0]
 145
 146     if mode == 'file':
 147         os.mknod(path)
 148     elif mode == 'dir':
 149         os.mkdir(path)
 150
 151     size = os.path.getsize(path)
 152
 153     assert fs_local.file_size(storage) == size
 154
 155     if mode == 'file':
 156         os.remove(path)
 157     elif mode == 'dir':
 158         os.rmdir(path)
 159
 160 @pytest.mark.parametrize("files, path", [
 161     (['foo', 'bar', 'foobar'], str(uuid.uuid4())),
 162     (['foo', 'bar', 'foobar'], [str(uuid.uuid4())])])
 163 def test_file_extract(fs_local, files, path):
 164     for f in files:
 165         os.mknod(valid_path() + f)
 166
 167     tar_path = valid_path() + str(uuid.uuid4()) + '.tar'
 168     with tarfile.open(tar_path, 'w') as tar:
 169         for f in files:
 170             tar.add(valid_path() + f, arcname=f)
 171
 172     with tarfile.open(tar_path, 'r') as tar:
 173         fs_local.file_extract(tar, path)
 174
 175     extracted_path = valid_path() + (path if isinstance(path, str) else '/'.join(path))
 176     ls_dir = os.listdir(extracted_path)
 177
 178     assert len(ls_dir) == len(files)
 179     for f in files:
 180         assert f in ls_dir
 181
 182     os.remove(tar_path)
 183
 184     for f in files:
 185         os.remove(valid_path() + f)
 186
 187     shutil.rmtree(extracted_path)
 188
 189 @pytest.mark.parametrize("storage, mode", [
 190     (str(uuid.uuid4()), 'r'),
 191     (str(uuid.uuid4()), 'w'),
 192     (str(uuid.uuid4()), 'a'),
 193     (str(uuid.uuid4()), 'rb'),
 194     (str(uuid.uuid4()), 'wb'),
 195     (str(uuid.uuid4()), 'ab'),
 196     ([str(uuid.uuid4())], 'r'),
 197     ([str(uuid.uuid4())], 'w'),
 198     ([str(uuid.uuid4())], 'a'),
 199     ([str(uuid.uuid4())], 'rb'),
 200     ([str(uuid.uuid4())], 'wb'),
 201     ([str(uuid.uuid4())], 'ab')])
 202 def test_file_open(fs_local, storage, mode):
 203     path = valid_path() + storage if isinstance(storage, str) else valid_path() + storage[0]
 204
 205     os.mknod(path)
 206
 207     file_obj = fs_local.file_open(storage, mode)
 208
 209     assert isinstance(file_obj, io.IOBase)
 210     assert file_obj.closed == False
 211
 212     os.remove(path)
 213
 214 @pytest.mark.parametrize("storage, mode", [
 215     (str(uuid.uuid4()), 'r'),
 216     (str(uuid.uuid4()), 'rb'),
 217     ([str(uuid.uuid4())], 'r'),
 218     ([str(uuid.uuid4())], 'rb')])
 219 def test_file_open_file_not_found_exception(fs_local, storage, mode):
 220     with pytest.raises(FsException) as excinfo:
 221         fs_local.file_open(storage, mode)
 222     assert str(excinfo.value) == file_open_file_not_found_exception(storage)
 223     assert excinfo.value.http_code == http.HTTPStatus.NOT_FOUND
 224
 225 @pytest.mark.parametrize("storage, mode", [
 226     (str(uuid.uuid4()), 'r'),
 227     (str(uuid.uuid4()), 'w'),
 228     (str(uuid.uuid4()), 'a'),
 229     (str(uuid.uuid4()), 'rb'),
 230     (str(uuid.uuid4()), 'wb'),
 231     (str(uuid.uuid4()), 'ab'),
 232     ([str(uuid.uuid4())], 'r'),
 233     ([str(uuid.uuid4())], 'w'),
 234     ([str(uuid.uuid4())], 'a'),
 235     ([str(uuid.uuid4())], 'rb'),
 236     ([str(uuid.uuid4())], 'wb'),
 237     ([str(uuid.uuid4())], 'ab')])
 238 def test_file_open_io_error(fs_local, storage, mode):
 239     path = valid_path() + storage if isinstance(storage, str) else valid_path() + storage[0]
 240
 241     os.mknod(path)
 242     os.chmod(path, 0)
 243
 244     with pytest.raises(FsException) as excinfo:
 245         fs_local.file_open(storage, mode)
 246     assert str(excinfo.value) == file_open_io_exception(storage)
 247     assert excinfo.value.http_code == http.HTTPStatus.BAD_REQUEST
 248
 249     os.remove(path)
 250
 251 @pytest.mark.parametrize("storage, with_files", [
 252     (str(uuid.uuid4()), True),
 253     (str(uuid.uuid4()), False),
 254     ([str(uuid.uuid4())], True),
 255     ([str(uuid.uuid4())], False)])
 256 def test_dir_ls(fs_local, storage, with_files):
 257     path = valid_path() + storage if isinstance(storage, str) else valid_path() + storage[0]
 258
 259     os.mkdir(path)
 260
 261     if with_files == True:
 262         file_name = str(uuid.uuid4())
 263         file_path = path + '/' + file_name
 264         os.mknod(file_path)
 265
 266     result = fs_local.dir_ls(storage)
 267
 268     if with_files == True:
 269         assert len(result) == 1
 270         assert result[0] == file_name
 271     else:
 272         assert len(result) == 0
 273
 274     shutil.rmtree(path)
 275
 276 @pytest.mark.parametrize("storage", [
 277     (str(uuid.uuid4())),
 278     ([str(uuid.uuid4())])])
 279 def test_dir_ls_with_not_a_directory_error(fs_local, storage):
 280     path = valid_path() + storage if isinstance(storage, str) else valid_path() + storage[0]
 281
 282     os.mknod(path)
 283
 284     with pytest.raises(FsException) as excinfo:
 285         fs_local.dir_ls(storage)
 286     assert str(excinfo.value) == dir_ls_not_a_directory_exception(storage)
 287     assert excinfo.value.http_code == http.HTTPStatus.NOT_FOUND
 288
 289     os.remove(path)
 290
 291 @pytest.mark.parametrize("storage", [
 292     (str(uuid.uuid4())),
 293     ([str(uuid.uuid4())])])
 294 def test_dir_ls_with_io_error(fs_local, storage):
 295     path = valid_path() + storage if isinstance(storage, str) else valid_path() + storage[0]
 296
 297     os.mkdir(path)
 298     os.chmod(path, 0)
 299
 300     with pytest.raises(FsException) as excinfo:
 301         fs_local.dir_ls(storage)
 302     assert str(excinfo.value) == dir_ls_io_exception(storage)
 303     assert excinfo.value.http_code == http.HTTPStatus.BAD_REQUEST
 304
 305     os.rmdir(path)
 306
 307 @pytest.mark.parametrize("storage, with_files, ignore_non_exist", [
 308     (str(uuid.uuid4()), True, True),
 309     (str(uuid.uuid4()), False, True),
 310     (str(uuid.uuid4()), True, False),
 311     (str(uuid.uuid4()), False, False),
 312     ([str(uuid.uuid4())], True, True),
 313     ([str(uuid.uuid4())], False, True),
 314     ([str(uuid.uuid4())], True, False),
 315     ([str(uuid.uuid4())], False, False)])
 316 def test_file_delete_with_dir(fs_local, storage, with_files, ignore_non_exist):
 317     path = valid_path() + storage if isinstance(storage, str) else valid_path() + storage[0]
 318
 319     os.mkdir(path)
 320
 321     if with_files == True:
 322         file_path = path + '/' + str(uuid.uuid4())
 323         os.mknod(file_path)
 324
 325     fs_local.file_delete(storage, ignore_non_exist)
 326
 327     assert os.path.exists(path) == False
 328
 329 @pytest.mark.parametrize("storage", [
 330     (str(uuid.uuid4())),
 331     ([str(uuid.uuid4())])])
 332 def test_file_delete_expect_exception(fs_local, storage):
 333     with pytest.raises(FsException) as excinfo:
 334         fs_local.file_delete(storage)
 335     assert str(excinfo.value) == file_delete_exception_message(storage)
 336     assert excinfo.value.http_code == http.HTTPStatus.NOT_FOUND
 337
 338 @pytest.mark.parametrize("storage", [
 339     (str(uuid.uuid4())),
 340     ([str(uuid.uuid4())])])
 341 def test_file_delete_no_exception(fs_local, storage):
 342     path = valid_path() + storage if isinstance(storage, str) else valid_path() + storage[0]
 343
 344     fs_local.file_delete(storage, ignore_non_exist=True)
 345
 346     assert os.path.exists(path) == False