1 # Copyright 2018 Whitestack, LLC
2 # Copyright 2018 Telefonica S.A.
4 # Licensed under the Apache License, Version 2.0 (the "License"); you may
5 # not use this file except in compliance with the License. You may obtain
6 # a copy of the License at
8 # http://www.apache.org/licenses/LICENSE-2.0
10 # Unless required by applicable law or agreed to in writing, software
11 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
12 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
13 # License for the specific language governing permissions and limitations
16 # For those usages not covered by the Apache License, Version 2.0 please
17 # contact: esousa@whitestack.com or alfonso.tiernosepulveda@telefonica.com
30 from osm_common
.fsbase
import FsException
31 from osm_common
.fslocal
import FsLocal
33 __author__
= "Eduardo Sousa <eduardosousa@av.it.pt>"
37 return tempfile
.gettempdir() + "/"
44 @pytest.fixture(scope
="function", params
=[True, False])
45 def fs_local(request
):
46 fs
= FsLocal(lock
=request
.param
)
47 fs
.fs_connect({"path": valid_path()})
51 def fs_connect_exception_message(path
):
52 return "storage exception Invalid configuration param at '[storage]': path '{}' does not exist".format(
57 def file_open_file_not_found_exception(storage
):
58 f
= storage
if isinstance(storage
, str) else "/".join(storage
)
59 return "storage exception File {} does not exist".format(f
)
62 def file_open_io_exception(storage
):
63 f
= storage
if isinstance(storage
, str) else "/".join(storage
)
64 return "storage exception File {} cannot be opened".format(f
)
67 def dir_ls_not_a_directory_exception(storage
):
68 f
= storage
if isinstance(storage
, str) else "/".join(storage
)
69 return "storage exception File {} does not exist".format(f
)
72 def dir_ls_io_exception(storage
):
73 f
= storage
if isinstance(storage
, str) else "/".join(storage
)
74 return "storage exception File {} cannot be opened".format(f
)
77 def file_delete_exception_message(storage
):
78 return "storage exception File {} does not exist".format(storage
)
81 def test_constructor_without_logger():
83 assert fs
.logger
== logging
.getLogger("fs")
84 assert fs
.path
is None
87 def test_constructor_with_logger():
88 logger_name
= "fs_local"
89 fs
= FsLocal(logger_name
=logger_name
)
90 assert fs
.logger
== logging
.getLogger(logger_name
)
91 assert fs
.path
is None
94 def test_get_params(fs_local
):
95 params
= fs_local
.get_params()
96 assert len(params
) == 2
98 assert "path" in params
99 assert params
["fs"] == "local"
100 assert params
["path"] == valid_path()
103 @pytest.mark
.parametrize(
104 "config, exp_logger, exp_path",
106 ({"logger_name": "fs_local", "path": valid_path()}, "fs_local", valid_path()),
108 {"logger_name": "fs_local", "path": valid_path()[:-1]},
112 ({"path": valid_path()}, "fs", valid_path()),
113 ({"path": valid_path()[:-1]}, "fs", valid_path()),
116 def test_fs_connect_with_valid_config(config
, exp_logger
, exp_path
):
118 fs
.fs_connect(config
)
119 assert fs
.logger
== logging
.getLogger(exp_logger
)
120 assert fs
.path
== exp_path
123 @pytest.mark
.parametrize(
124 "config, exp_exception_message",
127 {"logger_name": "fs_local", "path": invalid_path()},
128 fs_connect_exception_message(invalid_path()),
131 {"logger_name": "fs_local", "path": invalid_path()[:-1]},
132 fs_connect_exception_message(invalid_path()[:-1]),
134 ({"path": invalid_path()}, fs_connect_exception_message(invalid_path())),
136 {"path": invalid_path()[:-1]},
137 fs_connect_exception_message(invalid_path()[:-1]),
141 def test_fs_connect_with_invalid_path(config
, exp_exception_message
):
143 with pytest
.raises(FsException
) as excinfo
:
144 fs
.fs_connect(config
)
145 assert str(excinfo
.value
) == exp_exception_message
148 def test_fs_disconnect(fs_local
):
149 fs_local
.fs_disconnect()
152 def test_mkdir_with_valid_path(fs_local
):
153 folder_name
= str(uuid
.uuid4())
154 folder_path
= valid_path() + folder_name
155 fs_local
.mkdir(folder_name
)
156 assert os
.path
.exists(folder_path
)
158 fs_local
.mkdir(folder_name
)
159 assert os
.path
.exists(folder_path
)
160 os
.rmdir(folder_path
)
163 def test_mkdir_with_exception(fs_local
):
164 folder_name
= str(uuid
.uuid4())
165 with pytest
.raises(FsException
) as excinfo
:
166 fs_local
.mkdir(folder_name
+ "/" + folder_name
)
167 assert excinfo
.value
.http_code
== http
.HTTPStatus
.INTERNAL_SERVER_ERROR
170 @pytest.mark
.parametrize(
171 "storage, mode, expected",
173 (str(uuid
.uuid4()), "file", False),
174 ([str(uuid
.uuid4())], "file", False),
175 (str(uuid
.uuid4()), "dir", False),
176 ([str(uuid
.uuid4())], "dir", False),
179 def test_file_exists_returns_false(fs_local
, storage
, mode
, expected
):
180 assert fs_local
.file_exists(storage
, mode
) == expected
183 @pytest.mark
.parametrize(
184 "storage, mode, expected",
186 (str(uuid
.uuid4()), "file", True),
187 ([str(uuid
.uuid4())], "file", True),
188 (str(uuid
.uuid4()), "dir", True),
189 ([str(uuid
.uuid4())], "dir", True),
192 def test_file_exists_returns_true(fs_local
, storage
, mode
, expected
):
194 valid_path() + storage
195 if isinstance(storage
, str)
196 else valid_path() + storage
[0]
202 assert fs_local
.file_exists(storage
, mode
) == expected
209 @pytest.mark
.parametrize(
212 (str(uuid
.uuid4()), "file"),
213 ([str(uuid
.uuid4())], "file"),
214 (str(uuid
.uuid4()), "dir"),
215 ([str(uuid
.uuid4())], "dir"),
218 def test_file_size(fs_local
, storage
, mode
):
220 valid_path() + storage
221 if isinstance(storage
, str)
222 else valid_path() + storage
[0]
228 size
= os
.path
.getsize(path
)
229 assert fs_local
.file_size(storage
) == size
236 @pytest.mark
.parametrize(
239 (["foo", "bar", "foobar"], str(uuid
.uuid4())),
240 (["foo", "bar", "foobar"], [str(uuid
.uuid4())]),
243 def test_file_extract(fs_local
, files
, path
):
245 os
.mknod(valid_path() + f
)
246 tar_path
= valid_path() + str(uuid
.uuid4()) + ".tar"
247 with tarfile
.open(tar_path
, "w") as tar
:
249 tar
.add(valid_path() + f
, arcname
=f
)
250 with tarfile
.open(tar_path
, "r") as tar
:
251 fs_local
.file_extract(tar
, path
)
252 extracted_path
= valid_path() + (path
if isinstance(path
, str) else "/".join(path
))
253 ls_dir
= os
.listdir(extracted_path
)
254 assert len(ls_dir
) == len(files
)
259 os
.remove(valid_path() + f
)
260 shutil
.rmtree(extracted_path
)
263 @pytest.mark
.parametrize(
266 (str(uuid
.uuid4()), "r"),
267 (str(uuid
.uuid4()), "w"),
268 (str(uuid
.uuid4()), "a"),
269 (str(uuid
.uuid4()), "rb"),
270 (str(uuid
.uuid4()), "wb"),
271 (str(uuid
.uuid4()), "ab"),
272 ([str(uuid
.uuid4())], "r"),
273 ([str(uuid
.uuid4())], "w"),
274 ([str(uuid
.uuid4())], "a"),
275 ([str(uuid
.uuid4())], "rb"),
276 ([str(uuid
.uuid4())], "wb"),
277 ([str(uuid
.uuid4())], "ab"),
280 def test_file_open(fs_local
, storage
, mode
):
282 valid_path() + storage
283 if isinstance(storage
, str)
284 else valid_path() + storage
[0]
287 file_obj
= fs_local
.file_open(storage
, mode
)
288 assert isinstance(file_obj
, io
.IOBase
)
289 assert file_obj
.closed
is False
293 @pytest.mark
.parametrize(
296 (str(uuid
.uuid4()), "r"),
297 (str(uuid
.uuid4()), "rb"),
298 ([str(uuid
.uuid4())], "r"),
299 ([str(uuid
.uuid4())], "rb"),
302 def test_file_open_file_not_found_exception(fs_local
, storage
, mode
):
303 with pytest
.raises(FsException
) as excinfo
:
304 fs_local
.file_open(storage
, mode
)
305 assert str(excinfo
.value
) == file_open_file_not_found_exception(storage
)
306 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
309 @pytest.mark
.parametrize(
312 (str(uuid
.uuid4()), "r"),
313 (str(uuid
.uuid4()), "w"),
314 (str(uuid
.uuid4()), "a"),
315 (str(uuid
.uuid4()), "rb"),
316 (str(uuid
.uuid4()), "wb"),
317 (str(uuid
.uuid4()), "ab"),
318 ([str(uuid
.uuid4())], "r"),
319 ([str(uuid
.uuid4())], "w"),
320 ([str(uuid
.uuid4())], "a"),
321 ([str(uuid
.uuid4())], "rb"),
322 ([str(uuid
.uuid4())], "wb"),
323 ([str(uuid
.uuid4())], "ab"),
326 def test_file_open_io_error(fs_local
, storage
, mode
):
328 valid_path() + storage
329 if isinstance(storage
, str)
330 else valid_path() + storage
[0]
334 with pytest
.raises(FsException
) as excinfo
:
335 fs_local
.file_open(storage
, mode
)
336 assert str(excinfo
.value
) == file_open_io_exception(storage
)
337 assert excinfo
.value
.http_code
== http
.HTTPStatus
.BAD_REQUEST
341 @pytest.mark
.parametrize(
342 "storage, with_files",
344 (str(uuid
.uuid4()), True),
345 (str(uuid
.uuid4()), False),
346 ([str(uuid
.uuid4())], True),
347 ([str(uuid
.uuid4())], False),
350 def test_dir_ls(fs_local
, storage
, with_files
):
352 valid_path() + storage
353 if isinstance(storage
, str)
354 else valid_path() + storage
[0]
357 if with_files
is True:
358 file_name
= str(uuid
.uuid4())
359 file_path
= path
+ "/" + file_name
361 result
= fs_local
.dir_ls(storage
)
363 if with_files
is True:
364 assert len(result
) == 1
365 assert result
[0] == file_name
367 assert len(result
) == 0
371 @pytest.mark
.parametrize("storage", [(str(uuid
.uuid4())), ([str(uuid
.uuid4())])])
372 def test_dir_ls_with_not_a_directory_error(fs_local
, storage
):
374 valid_path() + storage
375 if isinstance(storage
, str)
376 else valid_path() + storage
[0]
379 with pytest
.raises(FsException
) as excinfo
:
380 fs_local
.dir_ls(storage
)
381 assert str(excinfo
.value
) == dir_ls_not_a_directory_exception(storage
)
382 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
386 @pytest.mark
.parametrize("storage", [(str(uuid
.uuid4())), ([str(uuid
.uuid4())])])
387 def test_dir_ls_with_io_error(fs_local
, storage
):
389 valid_path() + storage
390 if isinstance(storage
, str)
391 else valid_path() + storage
[0]
395 with pytest
.raises(FsException
) as excinfo
:
396 fs_local
.dir_ls(storage
)
397 assert str(excinfo
.value
) == dir_ls_io_exception(storage
)
398 assert excinfo
.value
.http_code
== http
.HTTPStatus
.BAD_REQUEST
402 @pytest.mark
.parametrize(
403 "storage, with_files, ignore_non_exist",
405 (str(uuid
.uuid4()), True, True),
406 (str(uuid
.uuid4()), False, True),
407 (str(uuid
.uuid4()), True, False),
408 (str(uuid
.uuid4()), False, False),
409 ([str(uuid
.uuid4())], True, True),
410 ([str(uuid
.uuid4())], False, True),
411 ([str(uuid
.uuid4())], True, False),
412 ([str(uuid
.uuid4())], False, False),
415 def test_file_delete_with_dir(fs_local
, storage
, with_files
, ignore_non_exist
):
417 valid_path() + storage
418 if isinstance(storage
, str)
419 else valid_path() + storage
[0]
422 if with_files
is True:
423 file_path
= path
+ "/" + str(uuid
.uuid4())
425 fs_local
.file_delete(storage
, ignore_non_exist
)
426 assert os
.path
.exists(path
) is False
429 @pytest.mark
.parametrize("storage", [(str(uuid
.uuid4())), ([str(uuid
.uuid4())])])
430 def test_file_delete_expect_exception(fs_local
, storage
):
431 with pytest
.raises(FsException
) as excinfo
:
432 fs_local
.file_delete(storage
)
433 assert str(excinfo
.value
) == file_delete_exception_message(storage
)
434 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
437 @pytest.mark
.parametrize("storage", [(str(uuid
.uuid4())), ([str(uuid
.uuid4())])])
438 def test_file_delete_no_exception(fs_local
, storage
):
440 valid_path() + storage
441 if isinstance(storage
, str)
442 else valid_path() + storage
[0]
444 fs_local
.file_delete(storage
, ignore_non_exist
=True)
445 assert os
.path
.exists(path
) is False