11 from osm_common
.fsbase
import FsException
12 from osm_common
.fslocal
import FsLocal
14 __author__
= "Eduardo Sousa <eduardosousa@av.it.pt>"
17 return tempfile
.gettempdir() + '/'
25 fs
.fs_connect({'path': valid_path()})
29 def fs_connect_exception_message(path
):
30 return "storage exception Invalid configuration param at '[storage]': path '{}' does not exist".format(path
)
32 def file_open_file_not_found_exception(storage
):
33 f
= storage
if isinstance(storage
, str) else '/'.join(storage
)
34 return "storage exception File {} does not exist".format(f
)
36 def file_open_io_exception(storage
):
37 f
= storage
if isinstance(storage
, str) else '/'.join(storage
)
38 return "storage exception File {} cannot be opened".format(f
)
40 def dir_ls_not_a_directory_exception(storage
):
41 f
= storage
if isinstance(storage
, str) else '/'.join(storage
)
42 return "storage exception File {} does not exist".format(f
)
44 def dir_ls_io_exception(storage
):
45 f
= storage
if isinstance(storage
, str) else '/'.join(storage
)
46 return "storage exception File {} cannot be opened".format(f
)
48 def file_delete_exception_message(storage
):
49 return "storage exception File {} does not exist".format(storage
)
51 def test_constructor_without_logger():
54 assert fs
.logger
== logging
.getLogger('fs')
55 assert fs
.path
is None
57 def test_constructor_with_logger():
58 logger_name
= 'fs_local'
60 fs
= FsLocal(logger_name
=logger_name
)
62 assert fs
.logger
== logging
.getLogger(logger_name
)
63 assert fs
.path
is None
65 def test_get_params(fs_local
):
66 params
= fs_local
.get_params()
68 assert len(params
) == 2
70 assert "path" in params
71 assert params
["fs"] == "local"
72 assert params
["path"] == valid_path()
74 @pytest.mark
.parametrize("config, exp_logger, exp_path", [
75 ({'logger_name': 'fs_local', 'path': valid_path()}, 'fs_local', valid_path()),
76 ({'logger_name': 'fs_local', 'path': valid_path()[:-1]}, 'fs_local', valid_path()),
77 ({'path': valid_path()}, 'fs', valid_path()),
78 ({'path': valid_path()[:-1]}, 'fs', valid_path())])
79 def test_fs_connect_with_valid_config(config
, exp_logger
, exp_path
):
83 assert fs
.logger
== logging
.getLogger(exp_logger
)
84 assert fs
.path
== exp_path
86 @pytest.mark
.parametrize("config, exp_exception_message", [
87 ({'logger_name': 'fs_local', 'path': invalid_path()}, fs_connect_exception_message(invalid_path())),
88 ({'logger_name': 'fs_local', 'path': invalid_path()[:-1]}, fs_connect_exception_message(invalid_path()[:-1])),
89 ({'path': invalid_path()}, fs_connect_exception_message(invalid_path())),
90 ({'path': invalid_path()[:-1]}, fs_connect_exception_message(invalid_path()[:-1]))])
91 def test_fs_connect_with_invalid_path(config
, exp_exception_message
):
94 with pytest
.raises(FsException
) as excinfo
:
96 assert str(excinfo
.value
) == exp_exception_message
98 def test_fs_disconnect(fs_local
):
99 fs_local
.fs_disconnect()
101 def test_mkdir_with_valid_path(fs_local
):
102 folder_name
= str(uuid
.uuid4())
103 folder_path
= valid_path() + folder_name
105 fs_local
.mkdir(folder_name
)
107 assert os
.path
.exists(folder_path
)
109 os
.rmdir(folder_path
)
111 def test_mkdir_with_exception(fs_local
):
112 folder_name
= str(uuid
.uuid4())
113 folder_path
= valid_path() + folder_name
114 os
.mkdir(folder_path
)
116 with pytest
.raises(FsException
) as excinfo
:
117 fs_local
.mkdir(folder_name
)
118 assert excinfo
.value
.http_code
== http
.HTTPStatus
.INTERNAL_SERVER_ERROR
120 os
.rmdir(folder_path
)
122 @pytest.mark
.parametrize("storage, mode, expected", [
123 (str(uuid
.uuid4()), 'file', False),
124 ([str(uuid
.uuid4())], 'file', False),
125 (str(uuid
.uuid4()), 'dir', False),
126 ([str(uuid
.uuid4())], 'dir', False)])
127 def test_file_exists_returns_false(fs_local
, storage
, mode
, expected
):
128 assert fs_local
.file_exists(storage
, mode
) == expected
130 @pytest.mark
.parametrize("storage, mode, expected", [
131 (str(uuid
.uuid4()), 'file', True),
132 ([str(uuid
.uuid4())], 'file', True),
133 (str(uuid
.uuid4()), 'dir', True),
134 ([str(uuid
.uuid4())], 'dir', True)])
135 def test_file_exists_returns_true(fs_local
, storage
, mode
, expected
):
136 path
= valid_path() + storage
if isinstance(storage
, str) else valid_path() + storage
[0]
143 assert fs_local
.file_exists(storage
, mode
) == expected
150 @pytest.mark
.parametrize("storage, mode", [
151 (str(uuid
.uuid4()), 'file'),
152 ([str(uuid
.uuid4())], 'file'),
153 (str(uuid
.uuid4()), 'dir'),
154 ([str(uuid
.uuid4())], 'dir')])
155 def test_file_size(fs_local
, storage
, mode
):
156 path
= valid_path() + storage
if isinstance(storage
, str) else valid_path() + storage
[0]
163 size
= os
.path
.getsize(path
)
165 assert fs_local
.file_size(storage
) == size
172 @pytest.mark
.parametrize("files, path", [
173 (['foo', 'bar', 'foobar'], str(uuid
.uuid4())),
174 (['foo', 'bar', 'foobar'], [str(uuid
.uuid4())])])
175 def test_file_extract(fs_local
, files
, path
):
177 os
.mknod(valid_path() + f
)
179 tar_path
= valid_path() + str(uuid
.uuid4()) + '.tar'
180 with tarfile
.open(tar_path
, 'w') as tar
:
182 tar
.add(valid_path() + f
, arcname
=f
)
184 with tarfile
.open(tar_path
, 'r') as tar
:
185 fs_local
.file_extract(tar
, path
)
187 extracted_path
= valid_path() + (path
if isinstance(path
, str) else '/'.join(path
))
188 ls_dir
= os
.listdir(extracted_path
)
190 assert len(ls_dir
) == len(files
)
197 os
.remove(valid_path() + f
)
199 shutil
.rmtree(extracted_path
)
201 @pytest.mark
.parametrize("storage, mode", [
202 (str(uuid
.uuid4()), 'r'),
203 (str(uuid
.uuid4()), 'w'),
204 (str(uuid
.uuid4()), 'a'),
205 (str(uuid
.uuid4()), 'rb'),
206 (str(uuid
.uuid4()), 'wb'),
207 (str(uuid
.uuid4()), 'ab'),
208 ([str(uuid
.uuid4())], 'r'),
209 ([str(uuid
.uuid4())], 'w'),
210 ([str(uuid
.uuid4())], 'a'),
211 ([str(uuid
.uuid4())], 'rb'),
212 ([str(uuid
.uuid4())], 'wb'),
213 ([str(uuid
.uuid4())], 'ab')])
214 def test_file_open(fs_local
, storage
, mode
):
215 path
= valid_path() + storage
if isinstance(storage
, str) else valid_path() + storage
[0]
219 file_obj
= fs_local
.file_open(storage
, mode
)
221 assert isinstance(file_obj
, io
.IOBase
)
222 assert file_obj
.closed
== False
226 @pytest.mark
.parametrize("storage, mode", [
227 (str(uuid
.uuid4()), 'r'),
228 (str(uuid
.uuid4()), 'rb'),
229 ([str(uuid
.uuid4())], 'r'),
230 ([str(uuid
.uuid4())], 'rb')])
231 def test_file_open_file_not_found_exception(fs_local
, storage
, mode
):
232 with pytest
.raises(FsException
) as excinfo
:
233 fs_local
.file_open(storage
, mode
)
234 assert str(excinfo
.value
) == file_open_file_not_found_exception(storage
)
235 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
237 @pytest.mark
.parametrize("storage, mode", [
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 ([str(uuid
.uuid4())], 'r'),
245 ([str(uuid
.uuid4())], 'w'),
246 ([str(uuid
.uuid4())], 'a'),
247 ([str(uuid
.uuid4())], 'rb'),
248 ([str(uuid
.uuid4())], 'wb'),
249 ([str(uuid
.uuid4())], 'ab')])
250 def test_file_open_io_error(fs_local
, storage
, mode
):
251 path
= valid_path() + storage
if isinstance(storage
, str) else valid_path() + storage
[0]
256 with pytest
.raises(FsException
) as excinfo
:
257 fs_local
.file_open(storage
, mode
)
258 assert str(excinfo
.value
) == file_open_io_exception(storage
)
259 assert excinfo
.value
.http_code
== http
.HTTPStatus
.BAD_REQUEST
263 @pytest.mark
.parametrize("storage, with_files", [
264 (str(uuid
.uuid4()), True),
265 (str(uuid
.uuid4()), False),
266 ([str(uuid
.uuid4())], True),
267 ([str(uuid
.uuid4())], False)])
268 def test_dir_ls(fs_local
, storage
, with_files
):
269 path
= valid_path() + storage
if isinstance(storage
, str) else valid_path() + storage
[0]
273 if with_files
== True:
274 file_name
= str(uuid
.uuid4())
275 file_path
= path
+ '/' + file_name
278 result
= fs_local
.dir_ls(storage
)
280 if with_files
== True:
281 assert len(result
) == 1
282 assert result
[0] == file_name
284 assert len(result
) == 0
288 @pytest.mark
.parametrize("storage", [
290 ([str(uuid
.uuid4())])])
291 def test_dir_ls_with_not_a_directory_error(fs_local
, storage
):
292 path
= valid_path() + storage
if isinstance(storage
, str) else valid_path() + storage
[0]
296 with pytest
.raises(FsException
) as excinfo
:
297 fs_local
.dir_ls(storage
)
298 assert str(excinfo
.value
) == dir_ls_not_a_directory_exception(storage
)
299 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
303 @pytest.mark
.parametrize("storage", [
305 ([str(uuid
.uuid4())])])
306 def test_dir_ls_with_io_error(fs_local
, storage
):
307 path
= valid_path() + storage
if isinstance(storage
, str) else valid_path() + storage
[0]
312 with pytest
.raises(FsException
) as excinfo
:
313 fs_local
.dir_ls(storage
)
314 assert str(excinfo
.value
) == dir_ls_io_exception(storage
)
315 assert excinfo
.value
.http_code
== http
.HTTPStatus
.BAD_REQUEST
319 @pytest.mark
.parametrize("storage, with_files, ignore_non_exist", [
320 (str(uuid
.uuid4()), True, True),
321 (str(uuid
.uuid4()), False, True),
322 (str(uuid
.uuid4()), True, False),
323 (str(uuid
.uuid4()), False, False),
324 ([str(uuid
.uuid4())], True, True),
325 ([str(uuid
.uuid4())], False, True),
326 ([str(uuid
.uuid4())], True, False),
327 ([str(uuid
.uuid4())], False, False)])
328 def test_file_delete_with_dir(fs_local
, storage
, with_files
, ignore_non_exist
):
329 path
= valid_path() + storage
if isinstance(storage
, str) else valid_path() + storage
[0]
333 if with_files
== True:
334 file_path
= path
+ '/' + str(uuid
.uuid4())
337 fs_local
.file_delete(storage
, ignore_non_exist
)
339 assert os
.path
.exists(path
) == False
341 @pytest.mark
.parametrize("storage", [
343 ([str(uuid
.uuid4())])])
344 def test_file_delete_expect_exception(fs_local
, storage
):
345 with pytest
.raises(FsException
) as excinfo
:
346 fs_local
.file_delete(storage
)
347 assert str(excinfo
.value
) == file_delete_exception_message(storage
)
348 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
350 @pytest.mark
.parametrize("storage", [
352 ([str(uuid
.uuid4())])])
353 def test_file_delete_no_exception(fs_local
, storage
):
354 path
= valid_path() + storage
if isinstance(storage
, str) else valid_path() + storage
[0]
356 fs_local
.file_delete(storage
, ignore_non_exist
=True)
358 assert os
.path
.exists(path
) == False