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 def test_get_params(fs_local):
  66     params = fs_local.get_params()
  67
  68     assert len(params) == 2
  69     assert "fs" in params
  70     assert "path" in params
  71     assert params["fs"] == "local"
  72     assert params["path"] == valid_path()
  73
  74 @pytest.mark.parametrize("config, exp_logger, exp_path", [
  75     ({'logger_name': 'fs_local', 'path': valid_path()}, 'fs_local', valid_path()),
  76     ({'logger_name': 'fs_local', 'path': valid_path()[:-1]}, 'fs_local', valid_path()),
  77     ({'path': valid_path()}, 'fs', valid_path()),
  78     ({'path': valid_path()[:-1]}, 'fs', valid_path())])
  79 def test_fs_connect_with_valid_config(config, exp_logger, exp_path):
  80     fs = FsLocal()
  81     fs.fs_connect(config)
  82
  83     assert fs.logger == logging.getLogger(exp_logger)
  84     assert fs.path == exp_path
  85
  86 @pytest.mark.parametrize("config, exp_exception_message", [
  87     ({'logger_name': 'fs_local', 'path': invalid_path()}, fs_connect_exception_message(invalid_path())),
  88     ({'logger_name': 'fs_local', 'path': invalid_path()[:-1]}, fs_connect_exception_message(invalid_path()[:-1])),
  89     ({'path': invalid_path()}, fs_connect_exception_message(invalid_path())),
  90     ({'path': invalid_path()[:-1]}, fs_connect_exception_message(invalid_path()[:-1]))])
  91 def test_fs_connect_with_invalid_path(config, exp_exception_message):
  92     fs = FsLocal()
  93
  94     with pytest.raises(FsException) as excinfo:
  95         fs.fs_connect(config)
  96     assert str(excinfo.value) == exp_exception_message
  97
  98 def test_fs_disconnect(fs_local):
  99     fs_local.fs_disconnect()
 100
 101 def test_mkdir_with_valid_path(fs_local):
 102     folder_name = str(uuid.uuid4())
 103     folder_path = valid_path() + folder_name
 104
 105     fs_local.mkdir(folder_name)
 106
 107     assert os.path.exists(folder_path)
 108
 109     os.rmdir(folder_path)
 110
 111 def test_mkdir_with_exception(fs_local):
 112     folder_name = str(uuid.uuid4())
 113     folder_path = valid_path() + folder_name
 114     os.mkdir(folder_path)
 115
 116     with pytest.raises(FsException) as excinfo:
 117         fs_local.mkdir(folder_name)
 118     assert excinfo.value.http_code == http.HTTPStatus.INTERNAL_SERVER_ERROR
 119
 120     os.rmdir(folder_path)
 121
 122 @pytest.mark.parametrize("storage, mode, expected", [
 123     (str(uuid.uuid4()), 'file', False),
 124     ([str(uuid.uuid4())], 'file', False),
 125     (str(uuid.uuid4()), 'dir', False),
 126     ([str(uuid.uuid4())], 'dir', False)])
 127 def test_file_exists_returns_false(fs_local, storage, mode, expected):
 128     assert fs_local.file_exists(storage, mode) == expected
 129
 130 @pytest.mark.parametrize("storage, mode, expected", [
 131     (str(uuid.uuid4()), 'file', True),
 132     ([str(uuid.uuid4())], 'file', True),
 133     (str(uuid.uuid4()), 'dir', True),
 134     ([str(uuid.uuid4())], 'dir', True)])
 135 def test_file_exists_returns_true(fs_local, storage, mode, expected):
 136     path = valid_path() + storage if isinstance(storage, str) else valid_path() + storage[0]
 137
 138     if mode == 'file':
 139         os.mknod(path)
 140     elif mode == 'dir':
 141         os.mkdir(path)
 142
 143     assert fs_local.file_exists(storage, mode) == expected
 144
 145     if mode == 'file':
 146         os.remove(path)
 147     elif mode == 'dir':
 148         os.rmdir(path)
 149
 150 @pytest.mark.parametrize("storage, mode", [
 151     (str(uuid.uuid4()), 'file'),
 152     ([str(uuid.uuid4())], 'file'),
 153     (str(uuid.uuid4()), 'dir'),
 154     ([str(uuid.uuid4())], 'dir')])
 155 def test_file_size(fs_local, storage, mode):
 156     path = valid_path() + storage if isinstance(storage, str) else valid_path() + storage[0]
 157
 158     if mode == 'file':
 159         os.mknod(path)
 160     elif mode == 'dir':
 161         os.mkdir(path)
 162
 163     size = os.path.getsize(path)
 164
 165     assert fs_local.file_size(storage) == size
 166
 167     if mode == 'file':
 168         os.remove(path)
 169     elif mode == 'dir':
 170         os.rmdir(path)
 171
 172 @pytest.mark.parametrize("files, path", [
 173     (['foo', 'bar', 'foobar'], str(uuid.uuid4())),
 174     (['foo', 'bar', 'foobar'], [str(uuid.uuid4())])])
 175 def test_file_extract(fs_local, files, path):
 176     for f in files:
 177         os.mknod(valid_path() + f)
 178
 179     tar_path = valid_path() + str(uuid.uuid4()) + '.tar'
 180     with tarfile.open(tar_path, 'w') as tar:
 181         for f in files:
 182             tar.add(valid_path() + f, arcname=f)
 183
 184     with tarfile.open(tar_path, 'r') as tar:
 185         fs_local.file_extract(tar, path)
 186
 187     extracted_path = valid_path() + (path if isinstance(path, str) else '/'.join(path))
 188     ls_dir = os.listdir(extracted_path)
 189
 190     assert len(ls_dir) == len(files)
 191     for f in files:
 192         assert f in ls_dir
 193
 194     os.remove(tar_path)
 195
 196     for f in files:
 197         os.remove(valid_path() + f)
 198
 199     shutil.rmtree(extracted_path)
 200
 201 @pytest.mark.parametrize("storage, mode", [
 202     (str(uuid.uuid4()), 'r'),
 203     (str(uuid.uuid4()), 'w'),
 204     (str(uuid.uuid4()), 'a'),
 205     (str(uuid.uuid4()), 'rb'),
 206     (str(uuid.uuid4()), 'wb'),
 207     (str(uuid.uuid4()), 'ab'),
 208     ([str(uuid.uuid4())], 'r'),
 209     ([str(uuid.uuid4())], 'w'),
 210     ([str(uuid.uuid4())], 'a'),
 211     ([str(uuid.uuid4())], 'rb'),
 212     ([str(uuid.uuid4())], 'wb'),
 213     ([str(uuid.uuid4())], 'ab')])
 214 def test_file_open(fs_local, storage, mode):
 215     path = valid_path() + storage if isinstance(storage, str) else valid_path() + storage[0]
 216
 217     os.mknod(path)
 218
 219     file_obj = fs_local.file_open(storage, mode)
 220
 221     assert isinstance(file_obj, io.IOBase)
 222     assert file_obj.closed == False
 223
 224     os.remove(path)
 225
 226 @pytest.mark.parametrize("storage, mode", [
 227     (str(uuid.uuid4()), 'r'),
 228     (str(uuid.uuid4()), 'rb'),
 229     ([str(uuid.uuid4())], 'r'),
 230     ([str(uuid.uuid4())], 'rb')])
 231 def test_file_open_file_not_found_exception(fs_local, storage, mode):
 232     with pytest.raises(FsException) as excinfo:
 233         fs_local.file_open(storage, mode)
 234     assert str(excinfo.value) == file_open_file_not_found_exception(storage)
 235     assert excinfo.value.http_code == http.HTTPStatus.NOT_FOUND
 236
 237 @pytest.mark.parametrize("storage, mode", [
 238     (str(uuid.uuid4()), 'r'),
 239     (str(uuid.uuid4()), 'w'),
 240     (str(uuid.uuid4()), 'a'),
 241     (str(uuid.uuid4()), 'rb'),
 242     (str(uuid.uuid4()), 'wb'),
 243     (str(uuid.uuid4()), 'ab'),
 244     ([str(uuid.uuid4())], 'r'),
 245     ([str(uuid.uuid4())], 'w'),
 246     ([str(uuid.uuid4())], 'a'),
 247     ([str(uuid.uuid4())], 'rb'),
 248     ([str(uuid.uuid4())], 'wb'),
 249     ([str(uuid.uuid4())], 'ab')])
 250 def test_file_open_io_error(fs_local, storage, mode):
 251     path = valid_path() + storage if isinstance(storage, str) else valid_path() + storage[0]
 252
 253     os.mknod(path)
 254     os.chmod(path, 0)
 255
 256     with pytest.raises(FsException) as excinfo:
 257         fs_local.file_open(storage, mode)
 258     assert str(excinfo.value) == file_open_io_exception(storage)
 259     assert excinfo.value.http_code == http.HTTPStatus.BAD_REQUEST
 260
 261     os.remove(path)
 262
 263 @pytest.mark.parametrize("storage, with_files", [
 264     (str(uuid.uuid4()), True),
 265     (str(uuid.uuid4()), False),
 266     ([str(uuid.uuid4())], True),
 267     ([str(uuid.uuid4())], False)])
 268 def test_dir_ls(fs_local, storage, with_files):
 269     path = valid_path() + storage if isinstance(storage, str) else valid_path() + storage[0]
 270
 271     os.mkdir(path)
 272
 273     if with_files == True:
 274         file_name = str(uuid.uuid4())
 275         file_path = path + '/' + file_name
 276         os.mknod(file_path)
 277
 278     result = fs_local.dir_ls(storage)
 279
 280     if with_files == True:
 281         assert len(result) == 1
 282         assert result[0] == file_name
 283     else:
 284         assert len(result) == 0
 285
 286     shutil.rmtree(path)
 287
 288 @pytest.mark.parametrize("storage", [
 289     (str(uuid.uuid4())),
 290     ([str(uuid.uuid4())])])
 291 def test_dir_ls_with_not_a_directory_error(fs_local, storage):
 292     path = valid_path() + storage if isinstance(storage, str) else valid_path() + storage[0]
 293
 294     os.mknod(path)
 295
 296     with pytest.raises(FsException) as excinfo:
 297         fs_local.dir_ls(storage)
 298     assert str(excinfo.value) == dir_ls_not_a_directory_exception(storage)
 299     assert excinfo.value.http_code == http.HTTPStatus.NOT_FOUND
 300
 301     os.remove(path)
 302
 303 @pytest.mark.parametrize("storage", [
 304     (str(uuid.uuid4())),
 305     ([str(uuid.uuid4())])])
 306 def test_dir_ls_with_io_error(fs_local, storage):
 307     path = valid_path() + storage if isinstance(storage, str) else valid_path() + storage[0]
 308
 309     os.mkdir(path)
 310     os.chmod(path, 0)
 311
 312     with pytest.raises(FsException) as excinfo:
 313         fs_local.dir_ls(storage)
 314     assert str(excinfo.value) == dir_ls_io_exception(storage)
 315     assert excinfo.value.http_code == http.HTTPStatus.BAD_REQUEST
 316
 317     os.rmdir(path)
 318
 319 @pytest.mark.parametrize("storage, with_files, ignore_non_exist", [
 320     (str(uuid.uuid4()), True, True),
 321     (str(uuid.uuid4()), False, True),
 322     (str(uuid.uuid4()), True, False),
 323     (str(uuid.uuid4()), False, False),
 324     ([str(uuid.uuid4())], True, True),
 325     ([str(uuid.uuid4())], False, True),
 326     ([str(uuid.uuid4())], True, False),
 327     ([str(uuid.uuid4())], False, False)])
 328 def test_file_delete_with_dir(fs_local, storage, with_files, ignore_non_exist):
 329     path = valid_path() + storage if isinstance(storage, str) else valid_path() + storage[0]
 330
 331     os.mkdir(path)
 332
 333     if with_files == True:
 334         file_path = path + '/' + str(uuid.uuid4())
 335         os.mknod(file_path)
 336
 337     fs_local.file_delete(storage, ignore_non_exist)
 338
 339     assert os.path.exists(path) == False
 340
 341 @pytest.mark.parametrize("storage", [
 342     (str(uuid.uuid4())),
 343     ([str(uuid.uuid4())])])
 344 def test_file_delete_expect_exception(fs_local, storage):
 345     with pytest.raises(FsException) as excinfo:
 346         fs_local.file_delete(storage)
 347     assert str(excinfo.value) == file_delete_exception_message(storage)
 348     assert excinfo.value.http_code == http.HTTPStatus.NOT_FOUND
 349
 350 @pytest.mark.parametrize("storage", [
 351     (str(uuid.uuid4())),
 352     ([str(uuid.uuid4())])])
 353 def test_file_delete_no_exception(fs_local, storage):
 354     path = valid_path() + storage if isinstance(storage, str) else valid_path() + storage[0]
 355
 356     fs_local.file_delete(storage, ignore_non_exist=True)
 357
 358     assert os.path.exists(path) == False