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 @pytest.mark
.parametrize("config, exp_logger, exp_path", [
66 ({'logger_name': 'fs_local', 'path': valid_path()}, 'fs_local', valid_path()),
67 ({'logger_name': 'fs_local', 'path': valid_path()[:-1]}, 'fs_local', valid_path()),
68 ({'path': valid_path()}, 'fs', valid_path()),
69 ({'path': valid_path()[:-1]}, 'fs', valid_path()),
71 def test_fs_connect_with_valid_config(config
, exp_logger
, exp_path
):
75 assert fs
.logger
== logging
.getLogger(exp_logger
)
76 assert fs
.path
== exp_path
78 @pytest.mark
.parametrize("config, exp_exception_message", [
79 ({'logger_name': 'fs_local', 'path': invalid_path()}, fs_connect_exception_message(invalid_path())),
80 ({'logger_name': 'fs_local', 'path': invalid_path()[:-1]}, fs_connect_exception_message(invalid_path()[:-1])),
81 ({'path': invalid_path()}, fs_connect_exception_message(invalid_path())),
82 ({'path': invalid_path()[:-1]}, fs_connect_exception_message(invalid_path()[:-1])),
84 def test_fs_connect_with_invalid_path(config
, exp_exception_message
):
87 with pytest
.raises(FsException
) as excinfo
:
89 assert str(excinfo
.value
) == exp_exception_message
91 def test_mkdir_with_valid_path(fs_local
):
92 folder_name
= str(uuid
.uuid4())
93 folder_path
= valid_path() + folder_name
95 fs_local
.mkdir(folder_name
)
97 assert os
.path
.exists(folder_path
)
101 def test_mkdir_with_exception(fs_local
):
102 folder_name
= str(uuid
.uuid4())
103 folder_path
= valid_path() + folder_name
104 os
.mkdir(folder_path
)
106 with pytest
.raises(FsException
) as excinfo
:
107 fs_local
.mkdir(folder_name
)
108 assert excinfo
.value
.http_code
== http
.HTTPStatus
.INTERNAL_SERVER_ERROR
110 os
.rmdir(folder_path
)
112 @pytest.mark
.parametrize("storage, mode, expected", [
113 (str(uuid
.uuid4()), 'file', False),
114 ([str(uuid
.uuid4())], 'file', False),
115 (str(uuid
.uuid4()), 'dir', False),
116 ([str(uuid
.uuid4())], 'dir', False),
118 def test_file_exists_returns_false(fs_local
, storage
, mode
, expected
):
119 assert fs_local
.file_exists(storage
, mode
) == expected
121 @pytest.mark
.parametrize("storage, mode, expected", [
122 (str(uuid
.uuid4()), 'file', True),
123 ([str(uuid
.uuid4())], 'file', True),
124 (str(uuid
.uuid4()), 'dir', True),
125 ([str(uuid
.uuid4())], 'dir', True),
127 def test_file_exists_returns_true(fs_local
, storage
, mode
, expected
):
128 path
= valid_path() + storage
if isinstance(storage
, str) else valid_path() + storage
[0]
135 assert fs_local
.file_exists(storage
, mode
) == expected
142 @pytest.mark
.parametrize("storage, mode", [
143 (str(uuid
.uuid4()), 'file'),
144 ([str(uuid
.uuid4())], 'file'),
145 (str(uuid
.uuid4()), 'dir'),
146 ([str(uuid
.uuid4())], 'dir'),
148 def test_file_size(fs_local
, storage
, mode
):
149 path
= valid_path() + storage
if isinstance(storage
, str) else valid_path() + storage
[0]
156 size
= os
.path
.getsize(path
)
158 assert fs_local
.file_size(storage
) == size
165 @pytest.mark
.parametrize("files, path", [
166 (['foo', 'bar', 'foobar'], str(uuid
.uuid4())),
167 (['foo', 'bar', 'foobar'], [str(uuid
.uuid4())]),
169 def test_file_extract(fs_local
, files
, path
):
171 os
.mknod(valid_path() + f
)
173 tar_path
= valid_path() + str(uuid
.uuid4()) + '.tar'
174 with tarfile
.open(tar_path
, 'w') as tar
:
176 tar
.add(valid_path() + f
, arcname
=f
)
178 with tarfile
.open(tar_path
, 'r') as tar
:
179 fs_local
.file_extract(tar
, path
)
181 extracted_path
= valid_path() + (path
if isinstance(path
, str) else '/'.join(path
))
182 ls_dir
= os
.listdir(extracted_path
)
184 assert len(ls_dir
) == len(files
)
191 os
.remove(valid_path() + f
)
193 shutil
.rmtree(extracted_path
)
195 @pytest.mark
.parametrize("storage, mode", [
196 (str(uuid
.uuid4()), 'r'),
197 (str(uuid
.uuid4()), 'w'),
198 (str(uuid
.uuid4()), 'a'),
199 (str(uuid
.uuid4()), 'rb'),
200 (str(uuid
.uuid4()), 'wb'),
201 (str(uuid
.uuid4()), 'ab'),
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'),
209 def test_file_open(fs_local
, storage
, mode
):
210 path
= valid_path() + storage
if isinstance(storage
, str) else valid_path() + storage
[0]
214 file_obj
= fs_local
.file_open(storage
, mode
)
216 assert isinstance(file_obj
, io
.IOBase
)
217 assert file_obj
.closed
== False
221 @pytest.mark
.parametrize("storage, mode", [
222 (str(uuid
.uuid4()), 'r'),
223 (str(uuid
.uuid4()), 'rb'),
224 ([str(uuid
.uuid4())], 'r'),
225 ([str(uuid
.uuid4())], 'rb'),
227 def test_file_open_file_not_found_exception(fs_local
, storage
, mode
):
228 with pytest
.raises(FsException
) as excinfo
:
229 fs_local
.file_open(storage
, mode
)
230 assert str(excinfo
.value
) == file_open_file_not_found_exception(storage
)
231 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
233 @pytest.mark
.parametrize("storage, mode", [
234 (str(uuid
.uuid4()), 'r'),
235 (str(uuid
.uuid4()), 'w'),
236 (str(uuid
.uuid4()), 'a'),
237 (str(uuid
.uuid4()), 'rb'),
238 (str(uuid
.uuid4()), 'wb'),
239 (str(uuid
.uuid4()), 'ab'),
240 ([str(uuid
.uuid4())], 'r'),
241 ([str(uuid
.uuid4())], 'w'),
242 ([str(uuid
.uuid4())], 'a'),
243 ([str(uuid
.uuid4())], 'rb'),
244 ([str(uuid
.uuid4())], 'wb'),
245 ([str(uuid
.uuid4())], 'ab'),
247 def test_file_open_io_error(fs_local
, storage
, mode
):
248 path
= valid_path() + storage
if isinstance(storage
, str) else valid_path() + storage
[0]
253 with pytest
.raises(FsException
) as excinfo
:
254 fs_local
.file_open(storage
, mode
)
255 assert str(excinfo
.value
) == file_open_io_exception(storage
)
256 assert excinfo
.value
.http_code
== http
.HTTPStatus
.BAD_REQUEST
260 @pytest.mark
.parametrize("storage, with_files", [
261 (str(uuid
.uuid4()), True),
262 (str(uuid
.uuid4()), False),
263 ([str(uuid
.uuid4())], True),
264 ([str(uuid
.uuid4())], False),
266 def test_dir_ls(fs_local
, storage
, with_files
):
267 path
= valid_path() + storage
if isinstance(storage
, str) else valid_path() + storage
[0]
271 if with_files
== True:
272 file_name
= str(uuid
.uuid4())
273 file_path
= path
+ '/' + file_name
276 result
= fs_local
.dir_ls(storage
)
278 if with_files
== True:
279 assert len(result
) == 1
280 assert result
[0] == file_name
282 assert len(result
) == 0
286 @pytest.mark
.parametrize("storage", [
288 ([str(uuid
.uuid4())]),
290 def test_dir_ls_with_not_a_directory_error(fs_local
, storage
):
291 path
= valid_path() + storage
if isinstance(storage
, str) else valid_path() + storage
[0]
295 with pytest
.raises(FsException
) as excinfo
:
296 fs_local
.dir_ls(storage
)
297 assert str(excinfo
.value
) == dir_ls_not_a_directory_exception(storage
)
298 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
302 @pytest.mark
.parametrize("storage", [
304 ([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),
329 def test_file_delete_with_dir(fs_local
, storage
, with_files
, ignore_non_exist
):
330 path
= valid_path() + storage
if isinstance(storage
, str) else valid_path() + storage
[0]
334 if with_files
== True:
335 file_path
= path
+ '/' + str(uuid
.uuid4())
338 fs_local
.file_delete(storage
, ignore_non_exist
)
340 assert os
.path
.exists(path
) == False
342 @pytest.mark
.parametrize("storage", [
344 ([str(uuid
.uuid4())]),
346 def test_file_delete_expect_exception(fs_local
, storage
):
347 with pytest
.raises(FsException
) as excinfo
:
348 fs_local
.file_delete(storage
)
349 assert str(excinfo
.value
) == file_delete_exception_message(storage
)
350 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
352 @pytest.mark
.parametrize("storage", [
354 ([str(uuid
.uuid4())]),
356 def test_file_delete_no_exception(fs_local
, storage
):
357 path
= valid_path() + storage
if isinstance(storage
, str) else valid_path() + storage
[0]
359 fs_local
.file_delete(storage
, ignore_non_exist
=True)
361 assert os
.path
.exists(path
) == False