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

diff --git a/osm_common/tests/test_fslocal.py b/osm_common/tests/test_fslocal.py
index 6a8e620..7416cfb 100644
--- a/osm_common/tests/test_fslocal.py
+++ b/osm_common/tests/test_fslocal.py
@@ -1,3 +1,22 @@
+# Copyright 2018 Whitestack, LLC
+# Copyright 2018 Telefonica S.A.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+#
+# For those usages not covered by the Apache License, Version 2.0 please
+# contact: esousa@whitestack.com or alfonso.tiernosepulveda@telefonica.com
+##
+
 import io
 import logging
 import http
@@ -13,185 +32,187 @@ from osm_common.fslocal import FsLocal
 
 __author__ = "Eduardo Sousa <eduardosousa@av.it.pt>"
 
+
 def valid_path():
     return tempfile.gettempdir() + '/'
 
+
 def invalid_path():
     return '/#tweeter/'
 
-@pytest.fixture
-def fs_local():
-    fs = FsLocal()
-    fs.fs_connect({'path': valid_path()})
 
+@pytest.fixture(scope="function", params=[True, False])
+def fs_local(request):
+    fs = FsLocal(lock=request.param)
+    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)
 
+
 def file_open_file_not_found_exception(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)
     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)
     return "storage exception File {} does not exist".format(f)
 
+
 def dir_ls_io_exception(storage):
     f = storage if isinstance(storage, str) else '/'.join(storage)
     return "storage exception File {} cannot be opened".format(f)
 
+
 def file_delete_exception_message(storage):
     return "storage exception File {} does not exist".format(storage)
 
+
 def test_constructor_without_logger():
     fs = FsLocal()
-
     assert fs.logger == logging.getLogger('fs')
     assert fs.path is None
 
+
 def test_constructor_with_logger():
     logger_name = 'fs_local'
-
     fs = FsLocal(logger_name=logger_name)
-
     assert fs.logger == logging.getLogger(logger_name)
     assert fs.path is None
 
+
+def test_get_params(fs_local):
+    params = fs_local.get_params()
+    assert len(params) == 2
+    assert "fs" in params
+    assert "path" in params
+    assert 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()),
-])
+    ({'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)
-
     assert fs.logger == logging.getLogger(exp_logger)
     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])),
-])
+    ({'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:
         fs.fs_connect(config)
     assert str(excinfo.value) == exp_exception_message
 
+
+def test_fs_disconnect(fs_local):
+    fs_local.fs_disconnect()
+
+
 def test_mkdir_with_valid_path(fs_local):
     folder_name = str(uuid.uuid4())
     folder_path = valid_path() + folder_name
-
     fs_local.mkdir(folder_name)
-
     assert os.path.exists(folder_path)
-
+    # test idempotency
+    fs_local.mkdir(folder_name)
+    assert os.path.exists(folder_path)
     os.rmdir(folder_path)
 
+
 def test_mkdir_with_exception(fs_local):
     folder_name = str(uuid.uuid4())
-    folder_path = valid_path() + folder_name
-    os.mkdir(folder_path)
-
     with pytest.raises(FsException) as excinfo:
-        fs_local.mkdir(folder_name)
+        fs_local.mkdir(folder_name + "/" + folder_name)
     assert excinfo.value.http_code == http.HTTPStatus.INTERNAL_SERVER_ERROR
 
-    os.rmdir(folder_path)
 
 @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),
-])
+    ([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),
-])
+    ([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':
         os.mknod(path)
     elif mode == 'dir':
         os.mkdir(path)
-
     assert fs_local.file_exists(storage, mode) == expected
-
     if mode == 'file':
         os.remove(path)
     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'),
-])
+    ([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':
         os.mknod(path)
     elif mode == 'dir':
         os.mkdir(path)
-
     size = os.path.getsize(path)
-
     assert fs_local.file_size(storage) == size
-
     if mode == 'file':
         os.remove(path)
     elif mode == 'dir':
         os.rmdir(path)
 
+
 @pytest.mark.parametrize("files, path", [
     (['foo', 'bar', 'foobar'], str(uuid.uuid4())),
-    (['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:
         for f in files:
             tar.add(valid_path() + f, arcname=f)
-    
     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))
     ls_dir = os.listdir(extracted_path)
-
     assert len(ls_dir) == len(files)
     for f in files:
         assert f in ls_dir
-    
     os.remove(tar_path)
-
     for f in files:
         os.remove(valid_path() + f)
-
     shutil.rmtree(extracted_path)
 
+
 @pytest.mark.parametrize("storage, mode", [
     (str(uuid.uuid4()), 'r'),
     (str(uuid.uuid4()), 'w'),
@@ -204,32 +225,28 @@ def test_file_extract(fs_local, files, path):
     ([str(uuid.uuid4())], 'a'),
     ([str(uuid.uuid4())], 'rb'),
     ([str(uuid.uuid4())], 'wb'),
-    ([str(uuid.uuid4())], 'ab'),
-])
+    ([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]
-
     os.mknod(path)
-
     file_obj = fs_local.file_open(storage, mode)
-
     assert isinstance(file_obj, io.IOBase)
-    assert file_obj.closed == False
-
+    assert file_obj.closed is False
     os.remove(path)
 
+
 @pytest.mark.parametrize("storage, mode", [
     (str(uuid.uuid4()), 'r'),
     (str(uuid.uuid4()), 'rb'),
     ([str(uuid.uuid4())], 'r'),
-    ([str(uuid.uuid4())], 'rb'),
-])
+    ([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)
     assert str(excinfo.value) == file_open_file_not_found_exception(storage)
     assert excinfo.value.http_code == http.HTTPStatus.NOT_FOUND
 
+
 @pytest.mark.parametrize("storage, mode", [
     (str(uuid.uuid4()), 'r'),
     (str(uuid.uuid4()), 'w'),
@@ -242,80 +259,67 @@ def test_file_open_file_not_found_exception(fs_local, storage, mode):
     ([str(uuid.uuid4())], 'a'),
     ([str(uuid.uuid4())], 'rb'),
     ([str(uuid.uuid4())], 'wb'),
-    ([str(uuid.uuid4())], 'ab'),
-])
+    ([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]
-
     os.mknod(path)
     os.chmod(path, 0)
-
     with pytest.raises(FsException) as excinfo:
         fs_local.file_open(storage, mode)
     assert str(excinfo.value) == file_open_io_exception(storage)
     assert excinfo.value.http_code == http.HTTPStatus.BAD_REQUEST
-
     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),
-])
+    ([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]
-
     os.mkdir(path)
-
-    if with_files == True:
+    if with_files is True:
         file_name = str(uuid.uuid4())
         file_path = path + '/' + file_name
         os.mknod(file_path)
-    
     result = fs_local.dir_ls(storage)
 
-    if with_files == True:
+    if with_files is True:
         assert len(result) == 1
         assert result[0] == file_name
     else:
         assert len(result) == 0
-    
     shutil.rmtree(path)
 
+
 @pytest.mark.parametrize("storage", [
     (str(uuid.uuid4())),
-    ([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]
-
     os.mknod(path)
-
     with pytest.raises(FsException) as excinfo:
         fs_local.dir_ls(storage)
     assert str(excinfo.value) == dir_ls_not_a_directory_exception(storage)
     assert excinfo.value.http_code == http.HTTPStatus.NOT_FOUND
-
     os.remove(path)
 
+
 @pytest.mark.parametrize("storage", [
     (str(uuid.uuid4())),
-    ([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]
-
     os.mkdir(path)
     os.chmod(path, 0)
-
     with pytest.raises(FsException) as excinfo:
         fs_local.dir_ls(storage)
     assert str(excinfo.value) == dir_ls_io_exception(storage)
     assert excinfo.value.http_code == http.HTTPStatus.BAD_REQUEST
-
     os.rmdir(path)
 
+
 @pytest.mark.parametrize("storage, with_files, ignore_non_exist", [
     (str(uuid.uuid4()), True, True),
     (str(uuid.uuid4()), False, True),
@@ -324,39 +328,31 @@ def test_dir_ls_with_io_error(fs_local, storage):
     ([str(uuid.uuid4())], True, True),
     ([str(uuid.uuid4())], False, True),
     ([str(uuid.uuid4())], True, False),
-    ([str(uuid.uuid4())], False, 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]
-
     os.mkdir(path)
-
-    if with_files == True:
+    if with_files is True:
         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
 
-    assert os.path.exists(path) == False
 
 @pytest.mark.parametrize("storage", [
     (str(uuid.uuid4())),
-    ([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)
     assert str(excinfo.value) == file_delete_exception_message(storage)
     assert excinfo.value.http_code == http.HTTPStatus.NOT_FOUND
 
+
 @pytest.mark.parametrize("storage", [
     (str(uuid.uuid4())),
-    ([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]
-
     fs_local.file_delete(storage, ignore_non_exist=True)
-
-    assert os.path.exists(path) == False
-    
\ No newline at end of file
+    assert os.path.exists(path) is False