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())])
70 def test_fs_connect_with_valid_config(config
, exp_logger
, exp_path
):
74 assert fs
.logger
== logging
.getLogger(exp_logger
)
75 assert fs
.path
== exp_path
77 @pytest.mark
.parametrize("config, exp_exception_message", [
78 ({'logger_name': 'fs_local', 'path': invalid_path()}, fs_connect_exception_message(invalid_path())),
79 ({'logger_name': 'fs_local', 'path': invalid_path()[:-1]}, fs_connect_exception_message(invalid_path()[:-1])),
80 ({'path': invalid_path()}, fs_connect_exception_message(invalid_path())),
81 ({'path': invalid_path()[:-1]}, fs_connect_exception_message(invalid_path()[:-1]))])
82 def test_fs_connect_with_invalid_path(config
, exp_exception_message
):
85 with pytest
.raises(FsException
) as excinfo
:
87 assert str(excinfo
.value
) == exp_exception_message
89 def test_mkdir_with_valid_path(fs_local
):
90 folder_name
= str(uuid
.uuid4())
91 folder_path
= valid_path() + folder_name
93 fs_local
.mkdir(folder_name
)
95 assert os
.path
.exists(folder_path
)
99 def test_mkdir_with_exception(fs_local
):
100 folder_name
= str(uuid
.uuid4())
101 folder_path
= valid_path() + folder_name
102 os
.mkdir(folder_path
)
104 with pytest
.raises(FsException
) as excinfo
:
105 fs_local
.mkdir(folder_name
)
106 assert excinfo
.value
.http_code
== http
.HTTPStatus
.INTERNAL_SERVER_ERROR
108 os
.rmdir(folder_path
)
110 @pytest.mark
.parametrize("storage, mode, expected", [
111 (str(uuid
.uuid4()), 'file', False),
112 ([str(uuid
.uuid4())], 'file', False),
113 (str(uuid
.uuid4()), 'dir', False),
114 ([str(uuid
.uuid4())], 'dir', False)])
115 def test_file_exists_returns_false(fs_local
, storage
, mode
, expected
):
116 assert fs_local
.file_exists(storage
, mode
) == expected
118 @pytest.mark
.parametrize("storage, mode, expected", [
119 (str(uuid
.uuid4()), 'file', True),
120 ([str(uuid
.uuid4())], 'file', True),
121 (str(uuid
.uuid4()), 'dir', True),
122 ([str(uuid
.uuid4())], 'dir', True)])
123 def test_file_exists_returns_true(fs_local
, storage
, mode
, expected
):
124 path
= valid_path() + storage
if isinstance(storage
, str) else valid_path() + storage
[0]
131 assert fs_local
.file_exists(storage
, mode
) == expected
138 @pytest.mark
.parametrize("storage, mode", [
139 (str(uuid
.uuid4()), 'file'),
140 ([str(uuid
.uuid4())], 'file'),
141 (str(uuid
.uuid4()), 'dir'),
142 ([str(uuid
.uuid4())], 'dir')])
143 def test_file_size(fs_local
, storage
, mode
):
144 path
= valid_path() + storage
if isinstance(storage
, str) else valid_path() + storage
[0]
151 size
= os
.path
.getsize(path
)
153 assert fs_local
.file_size(storage
) == size
160 @pytest.mark
.parametrize("files, path", [
161 (['foo', 'bar', 'foobar'], str(uuid
.uuid4())),
162 (['foo', 'bar', 'foobar'], [str(uuid
.uuid4())])])
163 def test_file_extract(fs_local
, files
, path
):
165 os
.mknod(valid_path() + f
)
167 tar_path
= valid_path() + str(uuid
.uuid4()) + '.tar'
168 with tarfile
.open(tar_path
, 'w') as tar
:
170 tar
.add(valid_path() + f
, arcname
=f
)
172 with tarfile
.open(tar_path
, 'r') as tar
:
173 fs_local
.file_extract(tar
, path
)
175 extracted_path
= valid_path() + (path
if isinstance(path
, str) else '/'.join(path
))
176 ls_dir
= os
.listdir(extracted_path
)
178 assert len(ls_dir
) == len(files
)
185 os
.remove(valid_path() + f
)
187 shutil
.rmtree(extracted_path
)
189 @pytest.mark
.parametrize("storage, mode", [
190 (str(uuid
.uuid4()), 'r'),
191 (str(uuid
.uuid4()), 'w'),
192 (str(uuid
.uuid4()), 'a'),
193 (str(uuid
.uuid4()), 'rb'),
194 (str(uuid
.uuid4()), 'wb'),
195 (str(uuid
.uuid4()), 'ab'),
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 def test_file_open(fs_local
, storage
, mode
):
203 path
= valid_path() + storage
if isinstance(storage
, str) else valid_path() + storage
[0]
207 file_obj
= fs_local
.file_open(storage
, mode
)
209 assert isinstance(file_obj
, io
.IOBase
)
210 assert file_obj
.closed
== False
214 @pytest.mark
.parametrize("storage, mode", [
215 (str(uuid
.uuid4()), 'r'),
216 (str(uuid
.uuid4()), 'rb'),
217 ([str(uuid
.uuid4())], 'r'),
218 ([str(uuid
.uuid4())], 'rb')])
219 def test_file_open_file_not_found_exception(fs_local
, storage
, mode
):
220 with pytest
.raises(FsException
) as excinfo
:
221 fs_local
.file_open(storage
, mode
)
222 assert str(excinfo
.value
) == file_open_file_not_found_exception(storage
)
223 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
225 @pytest.mark
.parametrize("storage, mode", [
226 (str(uuid
.uuid4()), 'r'),
227 (str(uuid
.uuid4()), 'w'),
228 (str(uuid
.uuid4()), 'a'),
229 (str(uuid
.uuid4()), 'rb'),
230 (str(uuid
.uuid4()), 'wb'),
231 (str(uuid
.uuid4()), 'ab'),
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 def test_file_open_io_error(fs_local
, storage
, mode
):
239 path
= valid_path() + storage
if isinstance(storage
, str) else valid_path() + storage
[0]
244 with pytest
.raises(FsException
) as excinfo
:
245 fs_local
.file_open(storage
, mode
)
246 assert str(excinfo
.value
) == file_open_io_exception(storage
)
247 assert excinfo
.value
.http_code
== http
.HTTPStatus
.BAD_REQUEST
251 @pytest.mark
.parametrize("storage, with_files", [
252 (str(uuid
.uuid4()), True),
253 (str(uuid
.uuid4()), False),
254 ([str(uuid
.uuid4())], True),
255 ([str(uuid
.uuid4())], False)])
256 def test_dir_ls(fs_local
, storage
, with_files
):
257 path
= valid_path() + storage
if isinstance(storage
, str) else valid_path() + storage
[0]
261 if with_files
== True:
262 file_name
= str(uuid
.uuid4())
263 file_path
= path
+ '/' + file_name
266 result
= fs_local
.dir_ls(storage
)
268 if with_files
== True:
269 assert len(result
) == 1
270 assert result
[0] == file_name
272 assert len(result
) == 0
276 @pytest.mark
.parametrize("storage", [
278 ([str(uuid
.uuid4())])])
279 def test_dir_ls_with_not_a_directory_error(fs_local
, storage
):
280 path
= valid_path() + storage
if isinstance(storage
, str) else valid_path() + storage
[0]
284 with pytest
.raises(FsException
) as excinfo
:
285 fs_local
.dir_ls(storage
)
286 assert str(excinfo
.value
) == dir_ls_not_a_directory_exception(storage
)
287 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
291 @pytest.mark
.parametrize("storage", [
293 ([str(uuid
.uuid4())])])
294 def test_dir_ls_with_io_error(fs_local
, storage
):
295 path
= valid_path() + storage
if isinstance(storage
, str) else valid_path() + storage
[0]
300 with pytest
.raises(FsException
) as excinfo
:
301 fs_local
.dir_ls(storage
)
302 assert str(excinfo
.value
) == dir_ls_io_exception(storage
)
303 assert excinfo
.value
.http_code
== http
.HTTPStatus
.BAD_REQUEST
307 @pytest.mark
.parametrize("storage, with_files, ignore_non_exist", [
308 (str(uuid
.uuid4()), True, True),
309 (str(uuid
.uuid4()), False, True),
310 (str(uuid
.uuid4()), True, False),
311 (str(uuid
.uuid4()), False, False),
312 ([str(uuid
.uuid4())], True, True),
313 ([str(uuid
.uuid4())], False, True),
314 ([str(uuid
.uuid4())], True, False),
315 ([str(uuid
.uuid4())], False, False)])
316 def test_file_delete_with_dir(fs_local
, storage
, with_files
, ignore_non_exist
):
317 path
= valid_path() + storage
if isinstance(storage
, str) else valid_path() + storage
[0]
321 if with_files
== True:
322 file_path
= path
+ '/' + str(uuid
.uuid4())
325 fs_local
.file_delete(storage
, ignore_non_exist
)
327 assert os
.path
.exists(path
) == False
329 @pytest.mark
.parametrize("storage", [
331 ([str(uuid
.uuid4())])])
332 def test_file_delete_expect_exception(fs_local
, storage
):
333 with pytest
.raises(FsException
) as excinfo
:
334 fs_local
.file_delete(storage
)
335 assert str(excinfo
.value
) == file_delete_exception_message(storage
)
336 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
338 @pytest.mark
.parametrize("storage", [
340 ([str(uuid
.uuid4())])])
341 def test_file_delete_no_exception(fs_local
, storage
):
342 path
= valid_path() + storage
if isinstance(storage
, str) else valid_path() + storage
[0]
344 fs_local
.file_delete(storage
, ignore_non_exist
=True)
346 assert os
.path
.exists(path
) == False