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