def fs_connect_exception_message(path):
return "storage exception Invalid configuration param at '[storage]': path '{}' does not exist".format(path)
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_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 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_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 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 file_delete_exception_message(storage):
return "storage exception File {} does not exist".format(storage)
+
+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()),
@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()),
@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])),
@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()[:-1]}, fs_connect_exception_message(invalid_path()[:-1]))])
def test_fs_connect_with_invalid_path(config, exp_exception_message):
fs = FsLocal()
({'path': invalid_path()[:-1]}, fs_connect_exception_message(invalid_path()[:-1]))])
def test_fs_connect_with_invalid_path(config, exp_exception_message):
fs = FsLocal()
def test_mkdir_with_valid_path(fs_local):
folder_name = str(uuid.uuid4())
folder_path = valid_path() + folder_name
def test_mkdir_with_valid_path(fs_local):
folder_name = str(uuid.uuid4())
folder_path = valid_path() + folder_name
def test_mkdir_with_exception(fs_local):
folder_name = str(uuid.uuid4())
folder_path = valid_path() + folder_name
os.mkdir(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
with pytest.raises(FsException) as excinfo:
fs_local.mkdir(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),
@pytest.mark.parametrize("storage, mode, expected", [
(str(uuid.uuid4()), 'file', False),
([str(uuid.uuid4())], 'file', False),
def test_file_exists_returns_false(fs_local, storage, mode, expected):
assert fs_local.file_exists(storage, mode) == expected
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),
@pytest.mark.parametrize("storage, mode, expected", [
(str(uuid.uuid4()), 'file', True),
([str(uuid.uuid4())], 'file', 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]
([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]
@pytest.mark.parametrize("storage, mode", [
(str(uuid.uuid4()), 'file'),
([str(uuid.uuid4())], 'file'),
@pytest.mark.parametrize("storage, mode", [
(str(uuid.uuid4()), 'file'),
([str(uuid.uuid4())], 'file'),
([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]
([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]
@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)
@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)
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)
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)
extracted_path = valid_path() + (path if isinstance(path, str) else '/'.join(path))
ls_dir = os.listdir(extracted_path)
@pytest.mark.parametrize("storage, mode", [
(str(uuid.uuid4()), 'r'),
(str(uuid.uuid4()), 'w'),
@pytest.mark.parametrize("storage, mode", [
(str(uuid.uuid4()), 'r'),
(str(uuid.uuid4()), 'w'),
([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]
([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]
@pytest.mark.parametrize("storage, mode", [
(str(uuid.uuid4()), 'r'),
(str(uuid.uuid4()), 'rb'),
@pytest.mark.parametrize("storage, mode", [
(str(uuid.uuid4()), 'r'),
(str(uuid.uuid4()), 'rb'),
assert str(excinfo.value) == file_open_file_not_found_exception(storage)
assert excinfo.value.http_code == http.HTTPStatus.NOT_FOUND
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'),
@pytest.mark.parametrize("storage, mode", [
(str(uuid.uuid4()), 'r'),
(str(uuid.uuid4()), 'w'),
([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]
([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]
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
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
@pytest.mark.parametrize("storage, with_files", [
(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())], False)])
def test_dir_ls(fs_local, storage, with_files):
path = valid_path() + storage if isinstance(storage, str) else valid_path() + storage[0]
([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]
@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]
@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]
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
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
@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]
@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]
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
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
@pytest.mark.parametrize("storage, with_files, ignore_non_exist", [
(str(uuid.uuid4()), True, True),
(str(uuid.uuid4()), False, True),
@pytest.mark.parametrize("storage, with_files, ignore_non_exist", [
(str(uuid.uuid4()), True, True),
(str(uuid.uuid4()), False, True),
([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]
([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]
assert str(excinfo.value) == file_delete_exception_message(storage)
assert excinfo.value.http_code == http.HTTPStatus.NOT_FOUND
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]
@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]