X-Git-Url: https://osm.etsi.org/gitweb/?p=osm%2Fcommon.git;a=blobdiff_plain;f=osm_common%2Ftests%2Ftest_fslocal.py;h=6f52984601ae90903da361d93043db6df95d1b41;hp=7416cfb4f0f50932027f8b23464a5a9499579ae1;hb=2644b76248a1b96f7a47013b414e31b4e3feecf8;hpb=fee977a304dbfadee797e25906df2db3cf51f1e3

diff --git a/osm_common/tests/test_fslocal.py b/osm_common/tests/test_fslocal.py
index 7416cfb..6f52984 100644
--- a/osm_common/tests/test_fslocal.py
+++ b/osm_common/tests/test_fslocal.py
@@ -34,41 +34,43 @@ __author__ = "Eduardo Sousa <eduardosousa@av.it.pt>"
 
 
 def valid_path():
-    return tempfile.gettempdir() + '/'
+    return tempfile.gettempdir() + "/"
 
 
 def invalid_path():
-    return '/#tweeter/'
+    return "/#tweeter/"
 
 
 @pytest.fixture(scope="function", params=[True, False])
 def fs_local(request):
     fs = FsLocal(lock=request.param)
-    fs.fs_connect({'path': valid_path()})
+    fs.fs_connect({"path": valid_path()})
     return fs
 
 
 def fs_connect_exception_message(path):
-    return "storage exception Invalid configuration param at '[storage]': path '{}' does not exist".format(path)
+    return "storage exception Invalid configuration param at '[storage]': path '{}' does not exist".format(
+        path
+    )
 
 
 def file_open_file_not_found_exception(storage):
-    f = storage if isinstance(storage, str) else '/'.join(storage)
+    f = storage if isinstance(storage, str) else "/".join(storage)
     return "storage exception File {} does not exist".format(f)
 
 
 def file_open_io_exception(storage):
-    f = storage if isinstance(storage, str) else '/'.join(storage)
+    f = storage if isinstance(storage, str) else "/".join(storage)
     return "storage exception File {} cannot be opened".format(f)
 
 
 def dir_ls_not_a_directory_exception(storage):
-    f = storage if isinstance(storage, str) else '/'.join(storage)
+    f = storage if isinstance(storage, str) else "/".join(storage)
     return "storage exception File {} does not exist".format(f)
 
 
 def dir_ls_io_exception(storage):
-    f = storage if isinstance(storage, str) else '/'.join(storage)
+    f = storage if isinstance(storage, str) else "/".join(storage)
     return "storage exception File {} cannot be opened".format(f)
 
 
@@ -78,12 +80,12 @@ def file_delete_exception_message(storage):
 
 def test_constructor_without_logger():
     fs = FsLocal()
-    assert fs.logger == logging.getLogger('fs')
+    assert fs.logger == logging.getLogger("fs")
     assert fs.path is None
 
 
 def test_constructor_with_logger():
-    logger_name = 'fs_local'
+    logger_name = "fs_local"
     fs = FsLocal(logger_name=logger_name)
     assert fs.logger == logging.getLogger(logger_name)
     assert fs.path is None
@@ -98,11 +100,19 @@ def test_get_params(fs_local):
     assert params["path"] == valid_path()
 
 
-@pytest.mark.parametrize("config, exp_logger, exp_path", [
-    ({'logger_name': 'fs_local', 'path': valid_path()}, 'fs_local', valid_path()),
-    ({'logger_name': 'fs_local', 'path': valid_path()[:-1]}, 'fs_local', valid_path()),
-    ({'path': valid_path()}, 'fs', valid_path()),
-    ({'path': valid_path()[:-1]}, 'fs', valid_path())])
+@pytest.mark.parametrize(
+    "config, exp_logger, exp_path",
+    [
+        ({"logger_name": "fs_local", "path": valid_path()}, "fs_local", valid_path()),
+        (
+            {"logger_name": "fs_local", "path": valid_path()[:-1]},
+            "fs_local",
+            valid_path(),
+        ),
+        ({"path": valid_path()}, "fs", valid_path()),
+        ({"path": valid_path()[:-1]}, "fs", valid_path()),
+    ],
+)
 def test_fs_connect_with_valid_config(config, exp_logger, exp_path):
     fs = FsLocal()
     fs.fs_connect(config)
@@ -110,11 +120,24 @@ def test_fs_connect_with_valid_config(config, exp_logger, exp_path):
     assert fs.path == exp_path
 
 
-@pytest.mark.parametrize("config, exp_exception_message", [
-    ({'logger_name': 'fs_local', 'path': invalid_path()}, fs_connect_exception_message(invalid_path())),
-    ({'logger_name': 'fs_local', 'path': invalid_path()[:-1]}, fs_connect_exception_message(invalid_path()[:-1])),
-    ({'path': invalid_path()}, fs_connect_exception_message(invalid_path())),
-    ({'path': invalid_path()[:-1]}, fs_connect_exception_message(invalid_path()[:-1]))])
+@pytest.mark.parametrize(
+    "config, exp_exception_message",
+    [
+        (
+            {"logger_name": "fs_local", "path": invalid_path()},
+            fs_connect_exception_message(invalid_path()),
+        ),
+        (
+            {"logger_name": "fs_local", "path": invalid_path()[:-1]},
+            fs_connect_exception_message(invalid_path()[:-1]),
+        ),
+        ({"path": invalid_path()}, fs_connect_exception_message(invalid_path())),
+        (
+            {"path": invalid_path()[:-1]},
+            fs_connect_exception_message(invalid_path()[:-1]),
+        ),
+    ],
+)
 def test_fs_connect_with_invalid_path(config, exp_exception_message):
     fs = FsLocal()
     with pytest.raises(FsException) as excinfo:
@@ -144,65 +167,89 @@ def test_mkdir_with_exception(fs_local):
     assert excinfo.value.http_code == http.HTTPStatus.INTERNAL_SERVER_ERROR
 
 
-@pytest.mark.parametrize("storage, mode, expected", [
-    (str(uuid.uuid4()), 'file', False),
-    ([str(uuid.uuid4())], 'file', False),
-    (str(uuid.uuid4()), 'dir', False),
-    ([str(uuid.uuid4())], 'dir', False)])
+@pytest.mark.parametrize(
+    "storage, mode, expected",
+    [
+        (str(uuid.uuid4()), "file", False),
+        ([str(uuid.uuid4())], "file", False),
+        (str(uuid.uuid4()), "dir", False),
+        ([str(uuid.uuid4())], "dir", False),
+    ],
+)
 def test_file_exists_returns_false(fs_local, storage, mode, expected):
     assert fs_local.file_exists(storage, mode) == expected
 
 
-@pytest.mark.parametrize("storage, mode, expected", [
-    (str(uuid.uuid4()), 'file', True),
-    ([str(uuid.uuid4())], 'file', True),
-    (str(uuid.uuid4()), 'dir', True),
-    ([str(uuid.uuid4())], 'dir', True)])
+@pytest.mark.parametrize(
+    "storage, mode, expected",
+    [
+        (str(uuid.uuid4()), "file", True),
+        ([str(uuid.uuid4())], "file", True),
+        (str(uuid.uuid4()), "dir", True),
+        ([str(uuid.uuid4())], "dir", True),
+    ],
+)
 def test_file_exists_returns_true(fs_local, storage, mode, expected):
-    path = valid_path() + storage if isinstance(storage, str) else valid_path() + storage[0]
-    if mode == 'file':
+    path = (
+        valid_path() + storage
+        if isinstance(storage, str)
+        else valid_path() + storage[0]
+    )
+    if mode == "file":
         os.mknod(path)
-    elif mode == 'dir':
+    elif mode == "dir":
         os.mkdir(path)
     assert fs_local.file_exists(storage, mode) == expected
-    if mode == 'file':
+    if mode == "file":
         os.remove(path)
-    elif mode == 'dir':
+    elif mode == "dir":
         os.rmdir(path)
 
 
-@pytest.mark.parametrize("storage, mode", [
-    (str(uuid.uuid4()), 'file'),
-    ([str(uuid.uuid4())], 'file'),
-    (str(uuid.uuid4()), 'dir'),
-    ([str(uuid.uuid4())], 'dir')])
+@pytest.mark.parametrize(
+    "storage, mode",
+    [
+        (str(uuid.uuid4()), "file"),
+        ([str(uuid.uuid4())], "file"),
+        (str(uuid.uuid4()), "dir"),
+        ([str(uuid.uuid4())], "dir"),
+    ],
+)
 def test_file_size(fs_local, storage, mode):
-    path = valid_path() + storage if isinstance(storage, str) else valid_path() + storage[0]
-    if mode == 'file':
+    path = (
+        valid_path() + storage
+        if isinstance(storage, str)
+        else valid_path() + storage[0]
+    )
+    if mode == "file":
         os.mknod(path)
-    elif mode == 'dir':
+    elif mode == "dir":
         os.mkdir(path)
     size = os.path.getsize(path)
     assert fs_local.file_size(storage) == size
-    if mode == 'file':
+    if mode == "file":
         os.remove(path)
-    elif mode == 'dir':
+    elif mode == "dir":
         os.rmdir(path)
 
 
-@pytest.mark.parametrize("files, path", [
-    (['foo', 'bar', 'foobar'], str(uuid.uuid4())),
-    (['foo', 'bar', 'foobar'], [str(uuid.uuid4())])])
+@pytest.mark.parametrize(
+    "files, path",
+    [
+        (["foo", "bar", "foobar"], str(uuid.uuid4())),
+        (["foo", "bar", "foobar"], [str(uuid.uuid4())]),
+    ],
+)
 def test_file_extract(fs_local, files, path):
     for f in files:
         os.mknod(valid_path() + f)
-    tar_path = valid_path() + str(uuid.uuid4()) + '.tar'
-    with tarfile.open(tar_path, 'w') as tar:
+    tar_path = valid_path() + str(uuid.uuid4()) + ".tar"
+    with tarfile.open(tar_path, "w") as tar:
         for f in files:
             tar.add(valid_path() + f, arcname=f)
-    with tarfile.open(tar_path, 'r') as tar:
+    with tarfile.open(tar_path, "r") as tar:
         fs_local.file_extract(tar, path)
-    extracted_path = valid_path() + (path if isinstance(path, str) else '/'.join(path))
+    extracted_path = valid_path() + (path if isinstance(path, str) else "/".join(path))
     ls_dir = os.listdir(extracted_path)
     assert len(ls_dir) == len(files)
     for f in files:
@@ -213,21 +260,29 @@ def test_file_extract(fs_local, files, path):
     shutil.rmtree(extracted_path)
 
 
-@pytest.mark.parametrize("storage, mode", [
-    (str(uuid.uuid4()), 'r'),
-    (str(uuid.uuid4()), 'w'),
-    (str(uuid.uuid4()), 'a'),
-    (str(uuid.uuid4()), 'rb'),
-    (str(uuid.uuid4()), 'wb'),
-    (str(uuid.uuid4()), 'ab'),
-    ([str(uuid.uuid4())], 'r'),
-    ([str(uuid.uuid4())], 'w'),
-    ([str(uuid.uuid4())], 'a'),
-    ([str(uuid.uuid4())], 'rb'),
-    ([str(uuid.uuid4())], 'wb'),
-    ([str(uuid.uuid4())], 'ab')])
+@pytest.mark.parametrize(
+    "storage, mode",
+    [
+        (str(uuid.uuid4()), "r"),
+        (str(uuid.uuid4()), "w"),
+        (str(uuid.uuid4()), "a"),
+        (str(uuid.uuid4()), "rb"),
+        (str(uuid.uuid4()), "wb"),
+        (str(uuid.uuid4()), "ab"),
+        ([str(uuid.uuid4())], "r"),
+        ([str(uuid.uuid4())], "w"),
+        ([str(uuid.uuid4())], "a"),
+        ([str(uuid.uuid4())], "rb"),
+        ([str(uuid.uuid4())], "wb"),
+        ([str(uuid.uuid4())], "ab"),
+    ],
+)
 def test_file_open(fs_local, storage, mode):
-    path = valid_path() + storage if isinstance(storage, str) else valid_path() + storage[0]
+    path = (
+        valid_path() + storage
+        if isinstance(storage, str)
+        else valid_path() + storage[0]
+    )
     os.mknod(path)
     file_obj = fs_local.file_open(storage, mode)
     assert isinstance(file_obj, io.IOBase)
@@ -235,11 +290,15 @@ def test_file_open(fs_local, storage, mode):
     os.remove(path)
 
 
-@pytest.mark.parametrize("storage, mode", [
-    (str(uuid.uuid4()), 'r'),
-    (str(uuid.uuid4()), 'rb'),
-    ([str(uuid.uuid4())], 'r'),
-    ([str(uuid.uuid4())], 'rb')])
+@pytest.mark.parametrize(
+    "storage, mode",
+    [
+        (str(uuid.uuid4()), "r"),
+        (str(uuid.uuid4()), "rb"),
+        ([str(uuid.uuid4())], "r"),
+        ([str(uuid.uuid4())], "rb"),
+    ],
+)
 def test_file_open_file_not_found_exception(fs_local, storage, mode):
     with pytest.raises(FsException) as excinfo:
         fs_local.file_open(storage, mode)
@@ -247,21 +306,29 @@ def test_file_open_file_not_found_exception(fs_local, storage, mode):
     assert excinfo.value.http_code == http.HTTPStatus.NOT_FOUND
 
 
-@pytest.mark.parametrize("storage, mode", [
-    (str(uuid.uuid4()), 'r'),
-    (str(uuid.uuid4()), 'w'),
-    (str(uuid.uuid4()), 'a'),
-    (str(uuid.uuid4()), 'rb'),
-    (str(uuid.uuid4()), 'wb'),
-    (str(uuid.uuid4()), 'ab'),
-    ([str(uuid.uuid4())], 'r'),
-    ([str(uuid.uuid4())], 'w'),
-    ([str(uuid.uuid4())], 'a'),
-    ([str(uuid.uuid4())], 'rb'),
-    ([str(uuid.uuid4())], 'wb'),
-    ([str(uuid.uuid4())], 'ab')])
+@pytest.mark.parametrize(
+    "storage, mode",
+    [
+        (str(uuid.uuid4()), "r"),
+        (str(uuid.uuid4()), "w"),
+        (str(uuid.uuid4()), "a"),
+        (str(uuid.uuid4()), "rb"),
+        (str(uuid.uuid4()), "wb"),
+        (str(uuid.uuid4()), "ab"),
+        ([str(uuid.uuid4())], "r"),
+        ([str(uuid.uuid4())], "w"),
+        ([str(uuid.uuid4())], "a"),
+        ([str(uuid.uuid4())], "rb"),
+        ([str(uuid.uuid4())], "wb"),
+        ([str(uuid.uuid4())], "ab"),
+    ],
+)
 def test_file_open_io_error(fs_local, storage, mode):
-    path = valid_path() + storage if isinstance(storage, str) else valid_path() + storage[0]
+    path = (
+        valid_path() + storage
+        if isinstance(storage, str)
+        else valid_path() + storage[0]
+    )
     os.mknod(path)
     os.chmod(path, 0)
     with pytest.raises(FsException) as excinfo:
@@ -271,17 +338,25 @@ def test_file_open_io_error(fs_local, storage, mode):
     os.remove(path)
 
 
-@pytest.mark.parametrize("storage, with_files", [
-    (str(uuid.uuid4()), True),
-    (str(uuid.uuid4()), False),
-    ([str(uuid.uuid4())], True),
-    ([str(uuid.uuid4())], False)])
+@pytest.mark.parametrize(
+    "storage, with_files",
+    [
+        (str(uuid.uuid4()), True),
+        (str(uuid.uuid4()), False),
+        ([str(uuid.uuid4())], True),
+        ([str(uuid.uuid4())], False),
+    ],
+)
 def test_dir_ls(fs_local, storage, with_files):
-    path = valid_path() + storage if isinstance(storage, str) else valid_path() + storage[0]
+    path = (
+        valid_path() + storage
+        if isinstance(storage, str)
+        else valid_path() + storage[0]
+    )
     os.mkdir(path)
     if with_files is True:
         file_name = str(uuid.uuid4())
-        file_path = path + '/' + file_name
+        file_path = path + "/" + file_name
         os.mknod(file_path)
     result = fs_local.dir_ls(storage)
 
@@ -293,11 +368,13 @@ def test_dir_ls(fs_local, storage, with_files):
     shutil.rmtree(path)
 
 
-@pytest.mark.parametrize("storage", [
-    (str(uuid.uuid4())),
-    ([str(uuid.uuid4())])])
+@pytest.mark.parametrize("storage", [(str(uuid.uuid4())), ([str(uuid.uuid4())])])
 def test_dir_ls_with_not_a_directory_error(fs_local, storage):
-    path = valid_path() + storage if isinstance(storage, str) else valid_path() + storage[0]
+    path = (
+        valid_path() + storage
+        if isinstance(storage, str)
+        else valid_path() + storage[0]
+    )
     os.mknod(path)
     with pytest.raises(FsException) as excinfo:
         fs_local.dir_ls(storage)
@@ -306,11 +383,13 @@ def test_dir_ls_with_not_a_directory_error(fs_local, storage):
     os.remove(path)
 
 
-@pytest.mark.parametrize("storage", [
-    (str(uuid.uuid4())),
-    ([str(uuid.uuid4())])])
+@pytest.mark.parametrize("storage", [(str(uuid.uuid4())), ([str(uuid.uuid4())])])
 def test_dir_ls_with_io_error(fs_local, storage):
-    path = valid_path() + storage if isinstance(storage, str) else valid_path() + storage[0]
+    path = (
+        valid_path() + storage
+        if isinstance(storage, str)
+        else valid_path() + storage[0]
+    )
     os.mkdir(path)
     os.chmod(path, 0)
     with pytest.raises(FsException) as excinfo:
@@ -320,28 +399,34 @@ def test_dir_ls_with_io_error(fs_local, storage):
     os.rmdir(path)
 
 
-@pytest.mark.parametrize("storage, with_files, ignore_non_exist", [
-    (str(uuid.uuid4()), True, True),
-    (str(uuid.uuid4()), False, True),
-    (str(uuid.uuid4()), True, False),
-    (str(uuid.uuid4()), False, False),
-    ([str(uuid.uuid4())], True, True),
-    ([str(uuid.uuid4())], False, True),
-    ([str(uuid.uuid4())], True, False),
-    ([str(uuid.uuid4())], False, False)])
+@pytest.mark.parametrize(
+    "storage, with_files, ignore_non_exist",
+    [
+        (str(uuid.uuid4()), True, True),
+        (str(uuid.uuid4()), False, True),
+        (str(uuid.uuid4()), True, False),
+        (str(uuid.uuid4()), False, False),
+        ([str(uuid.uuid4())], True, True),
+        ([str(uuid.uuid4())], False, True),
+        ([str(uuid.uuid4())], True, False),
+        ([str(uuid.uuid4())], False, False),
+    ],
+)
 def test_file_delete_with_dir(fs_local, storage, with_files, ignore_non_exist):
-    path = valid_path() + storage if isinstance(storage, str) else valid_path() + storage[0]
+    path = (
+        valid_path() + storage
+        if isinstance(storage, str)
+        else valid_path() + storage[0]
+    )
     os.mkdir(path)
     if with_files is True:
-        file_path = path + '/' + str(uuid.uuid4())
+        file_path = path + "/" + str(uuid.uuid4())
         os.mknod(file_path)
     fs_local.file_delete(storage, ignore_non_exist)
     assert os.path.exists(path) is False
 
 
-@pytest.mark.parametrize("storage", [
-    (str(uuid.uuid4())),
-    ([str(uuid.uuid4())])])
+@pytest.mark.parametrize("storage", [(str(uuid.uuid4())), ([str(uuid.uuid4())])])
 def test_file_delete_expect_exception(fs_local, storage):
     with pytest.raises(FsException) as excinfo:
         fs_local.file_delete(storage)
@@ -349,10 +434,12 @@ def test_file_delete_expect_exception(fs_local, storage):
     assert excinfo.value.http_code == http.HTTPStatus.NOT_FOUND
 
 
-@pytest.mark.parametrize("storage", [
-    (str(uuid.uuid4())),
-    ([str(uuid.uuid4())])])
+@pytest.mark.parametrize("storage", [(str(uuid.uuid4())), ([str(uuid.uuid4())])])
 def test_file_delete_no_exception(fs_local, storage):
-    path = valid_path() + storage if isinstance(storage, str) else valid_path() + storage[0]
+    path = (
+        valid_path() + storage
+        if isinstance(storage, str)
+        else valid_path() + storage[0]
+    )
     fs_local.file_delete(storage, ignore_non_exist=True)
     assert os.path.exists(path) is False