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