| # 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 http |
| import io |
| import logging |
| import os |
| import shutil |
| import tarfile |
| import tempfile |
| import uuid |
| |
| from osm_common.fsbase import FsException |
| from osm_common.fslocal import FsLocal |
| import pytest |
| |
| __author__ = "Eduardo Sousa <eduardosousa@av.it.pt>" |
| |
| |
| def valid_path(): |
| return tempfile.gettempdir() + "/" |
| |
| |
| def invalid_path(): |
| return "/#tweeter/" |
| |
| |
| @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()), |
| ], |
| ) |
| 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_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()) |
| with pytest.raises(FsException) as excinfo: |
| fs_local.mkdir(folder_name + "/" + folder_name) |
| 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), |
| ], |
| ) |
| 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 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"), |
| ], |
| ) |
| 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 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 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())])]) |
| 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 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 |
| |
| |
| @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) is False |