11 from osm_common
.fsbase
import FsException
12 from osm_common
.fslocal
import FsLocal
14 __author__
= "Eduardo Sousa <eduardosousa@av.it.pt>"
18 return tempfile
.gettempdir() + '/'
25 @pytest.fixture(scope
="function", params
=[True, False])
26 def fs_local(request
):
27 fs
= FsLocal(lock
=request
.param
)
28 fs
.fs_connect({'path': valid_path()})
32 def fs_connect_exception_message(path
):
33 return "storage exception Invalid configuration param at '[storage]': path '{}' does not exist".format(path
)
36 def file_open_file_not_found_exception(storage
):
37 f
= storage
if isinstance(storage
, str) else '/'.join(storage
)
38 return "storage exception File {} does not exist".format(f
)
41 def file_open_io_exception(storage
):
42 f
= storage
if isinstance(storage
, str) else '/'.join(storage
)
43 return "storage exception File {} cannot be opened".format(f
)
46 def dir_ls_not_a_directory_exception(storage
):
47 f
= storage
if isinstance(storage
, str) else '/'.join(storage
)
48 return "storage exception File {} does not exist".format(f
)
51 def dir_ls_io_exception(storage
):
52 f
= storage
if isinstance(storage
, str) else '/'.join(storage
)
53 return "storage exception File {} cannot be opened".format(f
)
56 def file_delete_exception_message(storage
):
57 return "storage exception File {} does not exist".format(storage
)
60 def test_constructor_without_logger():
62 assert fs
.logger
== logging
.getLogger('fs')
63 assert fs
.path
is None
66 def test_constructor_with_logger():
67 logger_name
= 'fs_local'
68 fs
= FsLocal(logger_name
=logger_name
)
69 assert fs
.logger
== logging
.getLogger(logger_name
)
70 assert fs
.path
is None
73 def test_get_params(fs_local
):
74 params
= fs_local
.get_params()
75 assert len(params
) == 2
77 assert "path" in params
78 assert params
["fs"] == "local"
79 assert params
["path"] == valid_path()
82 @pytest.mark
.parametrize("config, exp_logger, exp_path", [
83 ({'logger_name': 'fs_local', 'path': valid_path()}, 'fs_local', valid_path()),
84 ({'logger_name': 'fs_local', 'path': valid_path()[:-1]}, 'fs_local', valid_path()),
85 ({'path': valid_path()}, 'fs', valid_path()),
86 ({'path': valid_path()[:-1]}, 'fs', valid_path())])
87 def test_fs_connect_with_valid_config(config
, exp_logger
, exp_path
):
90 assert fs
.logger
== logging
.getLogger(exp_logger
)
91 assert fs
.path
== exp_path
94 @pytest.mark
.parametrize("config, exp_exception_message", [
95 ({'logger_name': 'fs_local', 'path': invalid_path()}, fs_connect_exception_message(invalid_path())),
96 ({'logger_name': 'fs_local', 'path': invalid_path()[:-1]}, fs_connect_exception_message(invalid_path()[:-1])),
97 ({'path': invalid_path()}, fs_connect_exception_message(invalid_path())),
98 ({'path': invalid_path()[:-1]}, fs_connect_exception_message(invalid_path()[:-1]))])
99 def test_fs_connect_with_invalid_path(config
, exp_exception_message
):
101 with pytest
.raises(FsException
) as excinfo
:
102 fs
.fs_connect(config
)
103 assert str(excinfo
.value
) == exp_exception_message
106 def test_fs_disconnect(fs_local
):
107 fs_local
.fs_disconnect()
110 def test_mkdir_with_valid_path(fs_local
):
111 folder_name
= str(uuid
.uuid4())
112 folder_path
= valid_path() + folder_name
113 fs_local
.mkdir(folder_name
)
114 assert os
.path
.exists(folder_path
)
115 os
.rmdir(folder_path
)
118 def test_mkdir_with_exception(fs_local
):
119 folder_name
= str(uuid
.uuid4())
120 folder_path
= valid_path() + folder_name
121 os
.mkdir(folder_path
)
122 with pytest
.raises(FsException
) as excinfo
:
123 fs_local
.mkdir(folder_name
)
124 assert excinfo
.value
.http_code
== http
.HTTPStatus
.INTERNAL_SERVER_ERROR
125 os
.rmdir(folder_path
)
128 @pytest.mark
.parametrize("storage, mode, expected", [
129 (str(uuid
.uuid4()), 'file', False),
130 ([str(uuid
.uuid4())], 'file', False),
131 (str(uuid
.uuid4()), 'dir', False),
132 ([str(uuid
.uuid4())], 'dir', False)])
133 def test_file_exists_returns_false(fs_local
, storage
, mode
, expected
):
134 assert fs_local
.file_exists(storage
, mode
) == expected
137 @pytest.mark
.parametrize("storage, mode, expected", [
138 (str(uuid
.uuid4()), 'file', True),
139 ([str(uuid
.uuid4())], 'file', True),
140 (str(uuid
.uuid4()), 'dir', True),
141 ([str(uuid
.uuid4())], 'dir', True)])
142 def test_file_exists_returns_true(fs_local
, storage
, mode
, expected
):
143 path
= valid_path() + storage
if isinstance(storage
, str) else valid_path() + storage
[0]
148 assert fs_local
.file_exists(storage
, mode
) == expected
155 @pytest.mark
.parametrize("storage, mode", [
156 (str(uuid
.uuid4()), 'file'),
157 ([str(uuid
.uuid4())], 'file'),
158 (str(uuid
.uuid4()), 'dir'),
159 ([str(uuid
.uuid4())], 'dir')])
160 def test_file_size(fs_local
, storage
, mode
):
161 path
= valid_path() + storage
if isinstance(storage
, str) else valid_path() + storage
[0]
166 size
= os
.path
.getsize(path
)
167 assert fs_local
.file_size(storage
) == size
174 @pytest.mark
.parametrize("files, path", [
175 (['foo', 'bar', 'foobar'], str(uuid
.uuid4())),
176 (['foo', 'bar', 'foobar'], [str(uuid
.uuid4())])])
177 def test_file_extract(fs_local
, files
, path
):
179 os
.mknod(valid_path() + f
)
180 tar_path
= valid_path() + str(uuid
.uuid4()) + '.tar'
181 with tarfile
.open(tar_path
, 'w') as tar
:
183 tar
.add(valid_path() + f
, arcname
=f
)
184 with tarfile
.open(tar_path
, 'r') as tar
:
185 fs_local
.file_extract(tar
, path
)
186 extracted_path
= valid_path() + (path
if isinstance(path
, str) else '/'.join(path
))
187 ls_dir
= os
.listdir(extracted_path
)
188 assert len(ls_dir
) == len(files
)
193 os
.remove(valid_path() + f
)
194 shutil
.rmtree(extracted_path
)
197 @pytest.mark
.parametrize("storage, mode", [
198 (str(uuid
.uuid4()), 'r'),
199 (str(uuid
.uuid4()), 'w'),
200 (str(uuid
.uuid4()), 'a'),
201 (str(uuid
.uuid4()), 'rb'),
202 (str(uuid
.uuid4()), 'wb'),
203 (str(uuid
.uuid4()), 'ab'),
204 ([str(uuid
.uuid4())], 'r'),
205 ([str(uuid
.uuid4())], 'w'),
206 ([str(uuid
.uuid4())], 'a'),
207 ([str(uuid
.uuid4())], 'rb'),
208 ([str(uuid
.uuid4())], 'wb'),
209 ([str(uuid
.uuid4())], 'ab')])
210 def test_file_open(fs_local
, storage
, mode
):
211 path
= valid_path() + storage
if isinstance(storage
, str) else valid_path() + storage
[0]
213 file_obj
= fs_local
.file_open(storage
, mode
)
214 assert isinstance(file_obj
, io
.IOBase
)
215 assert file_obj
.closed
is False
219 @pytest.mark
.parametrize("storage, mode", [
220 (str(uuid
.uuid4()), 'r'),
221 (str(uuid
.uuid4()), 'rb'),
222 ([str(uuid
.uuid4())], 'r'),
223 ([str(uuid
.uuid4())], 'rb')])
224 def test_file_open_file_not_found_exception(fs_local
, storage
, mode
):
225 with pytest
.raises(FsException
) as excinfo
:
226 fs_local
.file_open(storage
, mode
)
227 assert str(excinfo
.value
) == file_open_file_not_found_exception(storage
)
228 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
231 @pytest.mark
.parametrize("storage, mode", [
232 (str(uuid
.uuid4()), 'r'),
233 (str(uuid
.uuid4()), 'w'),
234 (str(uuid
.uuid4()), 'a'),
235 (str(uuid
.uuid4()), 'rb'),
236 (str(uuid
.uuid4()), 'wb'),
237 (str(uuid
.uuid4()), 'ab'),
238 ([str(uuid
.uuid4())], 'r'),
239 ([str(uuid
.uuid4())], 'w'),
240 ([str(uuid
.uuid4())], 'a'),
241 ([str(uuid
.uuid4())], 'rb'),
242 ([str(uuid
.uuid4())], 'wb'),
243 ([str(uuid
.uuid4())], 'ab')])
244 def test_file_open_io_error(fs_local
, storage
, mode
):
245 path
= valid_path() + storage
if isinstance(storage
, str) else valid_path() + storage
[0]
248 with pytest
.raises(FsException
) as excinfo
:
249 fs_local
.file_open(storage
, mode
)
250 assert str(excinfo
.value
) == file_open_io_exception(storage
)
251 assert excinfo
.value
.http_code
== http
.HTTPStatus
.BAD_REQUEST
255 @pytest.mark
.parametrize("storage, with_files", [
256 (str(uuid
.uuid4()), True),
257 (str(uuid
.uuid4()), False),
258 ([str(uuid
.uuid4())], True),
259 ([str(uuid
.uuid4())], False)])
260 def test_dir_ls(fs_local
, storage
, with_files
):
261 path
= valid_path() + storage
if isinstance(storage
, str) else valid_path() + storage
[0]
263 if with_files
is True:
264 file_name
= str(uuid
.uuid4())
265 file_path
= path
+ '/' + file_name
267 result
= fs_local
.dir_ls(storage
)
269 if with_files
is True:
270 assert len(result
) == 1
271 assert result
[0] == file_name
273 assert len(result
) == 0
277 @pytest.mark
.parametrize("storage", [
279 ([str(uuid
.uuid4())])])
280 def test_dir_ls_with_not_a_directory_error(fs_local
, storage
):
281 path
= valid_path() + storage
if isinstance(storage
, str) else valid_path() + storage
[0]
283 with pytest
.raises(FsException
) as excinfo
:
284 fs_local
.dir_ls(storage
)
285 assert str(excinfo
.value
) == dir_ls_not_a_directory_exception(storage
)
286 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
290 @pytest.mark
.parametrize("storage", [
292 ([str(uuid
.uuid4())])])
293 def test_dir_ls_with_io_error(fs_local
, storage
):
294 path
= valid_path() + storage
if isinstance(storage
, str) else valid_path() + storage
[0]
297 with pytest
.raises(FsException
) as excinfo
:
298 fs_local
.dir_ls(storage
)
299 assert str(excinfo
.value
) == dir_ls_io_exception(storage
)
300 assert excinfo
.value
.http_code
== http
.HTTPStatus
.BAD_REQUEST
304 @pytest.mark
.parametrize("storage, with_files, ignore_non_exist", [
305 (str(uuid
.uuid4()), True, True),
306 (str(uuid
.uuid4()), False, True),
307 (str(uuid
.uuid4()), True, False),
308 (str(uuid
.uuid4()), False, False),
309 ([str(uuid
.uuid4())], True, True),
310 ([str(uuid
.uuid4())], False, True),
311 ([str(uuid
.uuid4())], True, False),
312 ([str(uuid
.uuid4())], False, False)])
313 def test_file_delete_with_dir(fs_local
, storage
, with_files
, ignore_non_exist
):
314 path
= valid_path() + storage
if isinstance(storage
, str) else valid_path() + storage
[0]
316 if with_files
is True:
317 file_path
= path
+ '/' + str(uuid
.uuid4())
319 fs_local
.file_delete(storage
, ignore_non_exist
)
320 assert os
.path
.exists(path
) is False
323 @pytest.mark
.parametrize("storage", [
325 ([str(uuid
.uuid4())])])
326 def test_file_delete_expect_exception(fs_local
, storage
):
327 with pytest
.raises(FsException
) as excinfo
:
328 fs_local
.file_delete(storage
)
329 assert str(excinfo
.value
) == file_delete_exception_message(storage
)
330 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
333 @pytest.mark
.parametrize("storage", [
335 ([str(uuid
.uuid4())])])
336 def test_file_delete_no_exception(fs_local
, storage
):
337 path
= valid_path() + storage
if isinstance(storage
, str) else valid_path() + storage
[0]
338 fs_local
.file_delete(storage
, ignore_non_exist
=True)
339 assert os
.path
.exists(path
) is False