From f5119a8712de131f05096505ca4363f4b5f01122 Mon Sep 17 00:00:00 2001 From: Eduardo Sousa Date: Wed, 25 Apr 2018 17:55:17 +0100 Subject: [PATCH] Introducing unit tests for FsLocal Change-Id: I94c91e229d6e45cd527cea6d91aec8b004382090 Signed-off-by: Eduardo Sousa --- osm_common/tests/test_fslocal.py | 362 +++++++++++++++++++++++++++++++ 1 file changed, 362 insertions(+) create mode 100644 osm_common/tests/test_fslocal.py diff --git a/osm_common/tests/test_fslocal.py b/osm_common/tests/test_fslocal.py new file mode 100644 index 0000000..6a8e620 --- /dev/null +++ b/osm_common/tests/test_fslocal.py @@ -0,0 +1,362 @@ +import io +import logging +import http +import os +import pytest +import tarfile +import tempfile +import uuid +import shutil + +from osm_common.fsbase import FsException +from osm_common.fslocal import FsLocal + +__author__ = "Eduardo Sousa " + +def valid_path(): + return tempfile.gettempdir() + '/' + +def invalid_path(): + return '/#tweeter/' + +@pytest.fixture +def fs_local(): + fs = FsLocal() + 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 + +@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) + + 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])), +]) +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_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) + + 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) + 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), +]) +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), +]) +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'), +]) +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())]), +]) +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'), + (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] + + os.mknod(path) + + file_obj = fs_local.file_open(storage, mode) + + assert isinstance(file_obj, io.IOBase) + assert file_obj.closed == 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'), +]) +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'), + (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] + + 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), +]) +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: + file_name = str(uuid.uuid4()) + file_path = path + '/' + file_name + os.mknod(file_path) + + result = fs_local.dir_ls(storage) + + if with_files == 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())]), +]) +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())]), +]) +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), + (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] + + os.mkdir(path) + + if with_files == True: + file_path = path + '/' + str(uuid.uuid4()) + os.mknod(file_path) + + fs_local.file_delete(storage, ignore_non_exist) + + assert os.path.exists(path) == False + +@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) + 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())]), +]) +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 -- 2.25.1