osm_common/tests/test_fslocal.py

   1 # Copyright 2018 Whitestack, LLC
   2 # Copyright 2018 Telefonica S.A.
   3 #
   4 # Licensed under the Apache License, Version 2.0 (the "License"); you may
   5 # not use this file except in compliance with the License. You may obtain
   6 # 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, WITHOUT
  12 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  13 # License for the specific language governing permissions and limitations
  14 # under the License.
  15 #
  16 # For those usages not covered by the Apache License, Version 2.0 please
  17 # contact: esousa@whitestack.com or alfonso.tiernosepulveda@telefonica.com
  18 ##
  19
  20 import io
  21 import logging
  22 import http
  23 import os
  24 import pytest
  25 import tarfile
  26 import tempfile
  27 import uuid
  28 import shutil
  29
  30 from osm_common.fsbase import FsException
  31 from osm_common.fslocal import FsLocal
  32
  33 __author__ = "Eduardo Sousa <eduardosousa@av.it.pt>"
  34
  35
  36 def valid_path():
  37     return tempfile.gettempdir() + "/"
  38
  39
  40 def invalid_path():
  41     return "/#tweeter/"
  42
  43
  44 @pytest.fixture(scope="function", params=[True, False])
  45 def fs_local(request):
  46     fs = FsLocal(lock=request.param)
  47     fs.fs_connect({"path": valid_path()})
  48     return fs
  49
  50
  51 def fs_connect_exception_message(path):
  52     return "storage exception Invalid configuration param at '[storage]': path '{}' does not exist".format(
  53         path
  54     )
  55
  56
  57 def file_open_file_not_found_exception(storage):
  58     f = storage if isinstance(storage, str) else "/".join(storage)
  59     return "storage exception File {} does not exist".format(f)
  60
  61
  62 def file_open_io_exception(storage):
  63     f = storage if isinstance(storage, str) else "/".join(storage)
  64     return "storage exception File {} cannot be opened".format(f)
  65
  66
  67 def dir_ls_not_a_directory_exception(storage):
  68     f = storage if isinstance(storage, str) else "/".join(storage)
  69     return "storage exception File {} does not exist".format(f)
  70
  71
  72 def dir_ls_io_exception(storage):
  73     f = storage if isinstance(storage, str) else "/".join(storage)
  74     return "storage exception File {} cannot be opened".format(f)
  75
  76
  77 def file_delete_exception_message(storage):
  78     return "storage exception File {} does not exist".format(storage)
  79
  80
  81 def test_constructor_without_logger():
  82     fs = FsLocal()
  83     assert fs.logger == logging.getLogger("fs")
  84     assert fs.path is None
  85
  86
  87 def test_constructor_with_logger():
  88     logger_name = "fs_local"
  89     fs = FsLocal(logger_name=logger_name)
  90     assert fs.logger == logging.getLogger(logger_name)
  91     assert fs.path is None
  92
  93
  94 def test_get_params(fs_local):
  95     params = fs_local.get_params()
  96     assert len(params) == 2
  97     assert "fs" in params
  98     assert "path" in params
  99     assert params["fs"] == "local"
 100     assert params["path"] == valid_path()
 101
 102
 103 @pytest.mark.parametrize(
 104     "config, exp_logger, exp_path",
 105     [
 106         ({"logger_name": "fs_local", "path": valid_path()}, "fs_local", valid_path()),
 107         (
 108             {"logger_name": "fs_local", "path": valid_path()[:-1]},
 109             "fs_local",
 110             valid_path(),
 111         ),
 112         ({"path": valid_path()}, "fs", valid_path()),
 113         ({"path": valid_path()[:-1]}, "fs", valid_path()),
 114     ],
 115 )
 116 def test_fs_connect_with_valid_config(config, exp_logger, exp_path):
 117     fs = FsLocal()
 118     fs.fs_connect(config)
 119     assert fs.logger == logging.getLogger(exp_logger)
 120     assert fs.path == exp_path
 121
 122
 123 @pytest.mark.parametrize(
 124     "config, exp_exception_message",
 125     [
 126         (
 127             {"logger_name": "fs_local", "path": invalid_path()},
 128             fs_connect_exception_message(invalid_path()),
 129         ),
 130         (
 131             {"logger_name": "fs_local", "path": invalid_path()[:-1]},
 132             fs_connect_exception_message(invalid_path()[:-1]),
 133         ),
 134         ({"path": invalid_path()}, fs_connect_exception_message(invalid_path())),
 135         (
 136             {"path": invalid_path()[:-1]},
 137             fs_connect_exception_message(invalid_path()[:-1]),
 138         ),
 139     ],
 140 )
 141 def test_fs_connect_with_invalid_path(config, exp_exception_message):
 142     fs = FsLocal()
 143     with pytest.raises(FsException) as excinfo:
 144         fs.fs_connect(config)
 145     assert str(excinfo.value) == exp_exception_message
 146
 147
 148 def test_fs_disconnect(fs_local):
 149     fs_local.fs_disconnect()
 150
 151
 152 def test_mkdir_with_valid_path(fs_local):
 153     folder_name = str(uuid.uuid4())
 154     folder_path = valid_path() + folder_name
 155     fs_local.mkdir(folder_name)
 156     assert os.path.exists(folder_path)
 157     # test idempotency
 158     fs_local.mkdir(folder_name)
 159     assert os.path.exists(folder_path)
 160     os.rmdir(folder_path)
 161
 162
 163 def test_mkdir_with_exception(fs_local):
 164     folder_name = str(uuid.uuid4())
 165     with pytest.raises(FsException) as excinfo:
 166         fs_local.mkdir(folder_name + "/" + folder_name)
 167     assert excinfo.value.http_code == http.HTTPStatus.INTERNAL_SERVER_ERROR
 168
 169
 170 @pytest.mark.parametrize(
 171     "storage, mode, expected",
 172     [
 173         (str(uuid.uuid4()), "file", False),
 174         ([str(uuid.uuid4())], "file", False),
 175         (str(uuid.uuid4()), "dir", False),
 176         ([str(uuid.uuid4())], "dir", False),
 177     ],
 178 )
 179 def test_file_exists_returns_false(fs_local, storage, mode, expected):
 180     assert fs_local.file_exists(storage, mode) == expected
 181
 182
 183 @pytest.mark.parametrize(
 184     "storage, mode, expected",
 185     [
 186         (str(uuid.uuid4()), "file", True),
 187         ([str(uuid.uuid4())], "file", True),
 188         (str(uuid.uuid4()), "dir", True),
 189         ([str(uuid.uuid4())], "dir", True),
 190     ],
 191 )
 192 def test_file_exists_returns_true(fs_local, storage, mode, expected):
 193     path = (
 194         valid_path() + storage
 195         if isinstance(storage, str)
 196         else valid_path() + storage[0]
 197     )
 198     if mode == "file":
 199         os.mknod(path)
 200     elif mode == "dir":
 201         os.mkdir(path)
 202     assert fs_local.file_exists(storage, mode) == expected
 203     if mode == "file":
 204         os.remove(path)
 205     elif mode == "dir":
 206         os.rmdir(path)
 207
 208
 209 @pytest.mark.parametrize(
 210     "storage, mode",
 211     [
 212         (str(uuid.uuid4()), "file"),
 213         ([str(uuid.uuid4())], "file"),
 214         (str(uuid.uuid4()), "dir"),
 215         ([str(uuid.uuid4())], "dir"),
 216     ],
 217 )
 218 def test_file_size(fs_local, storage, mode):
 219     path = (
 220         valid_path() + storage
 221         if isinstance(storage, str)
 222         else valid_path() + storage[0]
 223     )
 224     if mode == "file":
 225         os.mknod(path)
 226     elif mode == "dir":
 227         os.mkdir(path)
 228     size = os.path.getsize(path)
 229     assert fs_local.file_size(storage) == size
 230     if mode == "file":
 231         os.remove(path)
 232     elif mode == "dir":
 233         os.rmdir(path)
 234
 235
 236 @pytest.mark.parametrize(
 237     "files, path",
 238     [
 239         (["foo", "bar", "foobar"], str(uuid.uuid4())),
 240         (["foo", "bar", "foobar"], [str(uuid.uuid4())]),
 241     ],
 242 )
 243 def test_file_extract(fs_local, files, path):
 244     for f in files:
 245         os.mknod(valid_path() + f)
 246     tar_path = valid_path() + str(uuid.uuid4()) + ".tar"
 247     with tarfile.open(tar_path, "w") as tar:
 248         for f in files:
 249             tar.add(valid_path() + f, arcname=f)
 250     with tarfile.open(tar_path, "r") as tar:
 251         fs_local.file_extract(tar, path)
 252     extracted_path = valid_path() + (path if isinstance(path, str) else "/".join(path))
 253     ls_dir = os.listdir(extracted_path)
 254     assert len(ls_dir) == len(files)
 255     for f in files:
 256         assert f in ls_dir
 257     os.remove(tar_path)
 258     for f in files:
 259         os.remove(valid_path() + f)
 260     shutil.rmtree(extracted_path)
 261
 262
 263 @pytest.mark.parametrize(
 264     "storage, mode",
 265     [
 266         (str(uuid.uuid4()), "r"),
 267         (str(uuid.uuid4()), "w"),
 268         (str(uuid.uuid4()), "a"),
 269         (str(uuid.uuid4()), "rb"),
 270         (str(uuid.uuid4()), "wb"),
 271         (str(uuid.uuid4()), "ab"),
 272         ([str(uuid.uuid4())], "r"),
 273         ([str(uuid.uuid4())], "w"),
 274         ([str(uuid.uuid4())], "a"),
 275         ([str(uuid.uuid4())], "rb"),
 276         ([str(uuid.uuid4())], "wb"),
 277         ([str(uuid.uuid4())], "ab"),
 278     ],
 279 )
 280 def test_file_open(fs_local, storage, mode):
 281     path = (
 282         valid_path() + storage
 283         if isinstance(storage, str)
 284         else valid_path() + storage[0]
 285     )
 286     os.mknod(path)
 287     file_obj = fs_local.file_open(storage, mode)
 288     assert isinstance(file_obj, io.IOBase)
 289     assert file_obj.closed is False
 290     os.remove(path)
 291
 292
 293 @pytest.mark.parametrize(
 294     "storage, mode",
 295     [
 296         (str(uuid.uuid4()), "r"),
 297         (str(uuid.uuid4()), "rb"),
 298         ([str(uuid.uuid4())], "r"),
 299         ([str(uuid.uuid4())], "rb"),
 300     ],
 301 )
 302 def test_file_open_file_not_found_exception(fs_local, storage, mode):
 303     with pytest.raises(FsException) as excinfo:
 304         fs_local.file_open(storage, mode)
 305     assert str(excinfo.value) == file_open_file_not_found_exception(storage)
 306     assert excinfo.value.http_code == http.HTTPStatus.NOT_FOUND
 307
 308
 309 @pytest.mark.parametrize(
 310     "storage, mode",
 311     [
 312         (str(uuid.uuid4()), "r"),
 313         (str(uuid.uuid4()), "w"),
 314         (str(uuid.uuid4()), "a"),
 315         (str(uuid.uuid4()), "rb"),
 316         (str(uuid.uuid4()), "wb"),
 317         (str(uuid.uuid4()), "ab"),
 318         ([str(uuid.uuid4())], "r"),
 319         ([str(uuid.uuid4())], "w"),
 320         ([str(uuid.uuid4())], "a"),
 321         ([str(uuid.uuid4())], "rb"),
 322         ([str(uuid.uuid4())], "wb"),
 323         ([str(uuid.uuid4())], "ab"),
 324     ],
 325 )
 326 def test_file_open_io_error(fs_local, storage, mode):
 327     path = (
 328         valid_path() + storage
 329         if isinstance(storage, str)
 330         else valid_path() + storage[0]
 331     )
 332     os.mknod(path)
 333     os.chmod(path, 0)
 334     with pytest.raises(FsException) as excinfo:
 335         fs_local.file_open(storage, mode)
 336     assert str(excinfo.value) == file_open_io_exception(storage)
 337     assert excinfo.value.http_code == http.HTTPStatus.BAD_REQUEST
 338     os.remove(path)
 339
 340
 341 @pytest.mark.parametrize(
 342     "storage, with_files",
 343     [
 344         (str(uuid.uuid4()), True),
 345         (str(uuid.uuid4()), False),
 346         ([str(uuid.uuid4())], True),
 347         ([str(uuid.uuid4())], False),
 348     ],
 349 )
 350 def test_dir_ls(fs_local, storage, with_files):
 351     path = (
 352         valid_path() + storage
 353         if isinstance(storage, str)
 354         else valid_path() + storage[0]
 355     )
 356     os.mkdir(path)
 357     if with_files is True:
 358         file_name = str(uuid.uuid4())
 359         file_path = path + "/" + file_name
 360         os.mknod(file_path)
 361     result = fs_local.dir_ls(storage)
 362
 363     if with_files is True:
 364         assert len(result) == 1
 365         assert result[0] == file_name
 366     else:
 367         assert len(result) == 0
 368     shutil.rmtree(path)
 369
 370
 371 @pytest.mark.parametrize("storage", [(str(uuid.uuid4())), ([str(uuid.uuid4())])])
 372 def test_dir_ls_with_not_a_directory_error(fs_local, storage):
 373     path = (
 374         valid_path() + storage
 375         if isinstance(storage, str)
 376         else valid_path() + storage[0]
 377     )
 378     os.mknod(path)
 379     with pytest.raises(FsException) as excinfo:
 380         fs_local.dir_ls(storage)
 381     assert str(excinfo.value) == dir_ls_not_a_directory_exception(storage)
 382     assert excinfo.value.http_code == http.HTTPStatus.NOT_FOUND
 383     os.remove(path)
 384
 385
 386 @pytest.mark.parametrize("storage", [(str(uuid.uuid4())), ([str(uuid.uuid4())])])
 387 def test_dir_ls_with_io_error(fs_local, storage):
 388     path = (
 389         valid_path() + storage
 390         if isinstance(storage, str)
 391         else valid_path() + storage[0]
 392     )
 393     os.mkdir(path)
 394     os.chmod(path, 0)
 395     with pytest.raises(FsException) as excinfo:
 396         fs_local.dir_ls(storage)
 397     assert str(excinfo.value) == dir_ls_io_exception(storage)
 398     assert excinfo.value.http_code == http.HTTPStatus.BAD_REQUEST
 399     os.rmdir(path)
 400
 401
 402 @pytest.mark.parametrize(
 403     "storage, with_files, ignore_non_exist",
 404     [
 405         (str(uuid.uuid4()), True, True),
 406         (str(uuid.uuid4()), False, True),
 407         (str(uuid.uuid4()), True, False),
 408         (str(uuid.uuid4()), False, False),
 409         ([str(uuid.uuid4())], True, True),
 410         ([str(uuid.uuid4())], False, True),
 411         ([str(uuid.uuid4())], True, False),
 412         ([str(uuid.uuid4())], False, False),
 413     ],
 414 )
 415 def test_file_delete_with_dir(fs_local, storage, with_files, ignore_non_exist):
 416     path = (
 417         valid_path() + storage
 418         if isinstance(storage, str)
 419         else valid_path() + storage[0]
 420     )
 421     os.mkdir(path)
 422     if with_files is True:
 423         file_path = path + "/" + str(uuid.uuid4())
 424         os.mknod(file_path)
 425     fs_local.file_delete(storage, ignore_non_exist)
 426     assert os.path.exists(path) is False
 427
 428
 429 @pytest.mark.parametrize("storage", [(str(uuid.uuid4())), ([str(uuid.uuid4())])])
 430 def test_file_delete_expect_exception(fs_local, storage):
 431     with pytest.raises(FsException) as excinfo:
 432         fs_local.file_delete(storage)
 433     assert str(excinfo.value) == file_delete_exception_message(storage)
 434     assert excinfo.value.http_code == http.HTTPStatus.NOT_FOUND
 435
 436
 437 @pytest.mark.parametrize("storage", [(str(uuid.uuid4())), ([str(uuid.uuid4())])])
 438 def test_file_delete_no_exception(fs_local, storage):
 439     path = (
 440         valid_path() + storage
 441         if isinstance(storage, str)
 442         else valid_path() + storage[0]
 443     )
 444     fs_local.file_delete(storage, ignore_non_exist=True)
 445     assert os.path.exists(path) is False