7416cfb4f0f50932027f8b23464a5a9499579ae1
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(path
)
55 def file_open_file_not_found_exception(storage
):
56 f
= storage
if isinstance(storage
, str) else '/'.join(storage
)
57 return "storage exception File {} does not exist".format(f
)
60 def file_open_io_exception(storage
):
61 f
= storage
if isinstance(storage
, str) else '/'.join(storage
)
62 return "storage exception File {} cannot be opened".format(f
)
65 def dir_ls_not_a_directory_exception(storage
):
66 f
= storage
if isinstance(storage
, str) else '/'.join(storage
)
67 return "storage exception File {} does not exist".format(f
)
70 def dir_ls_io_exception(storage
):
71 f
= storage
if isinstance(storage
, str) else '/'.join(storage
)
72 return "storage exception File {} cannot be opened".format(f
)
75 def file_delete_exception_message(storage
):
76 return "storage exception File {} does not exist".format(storage
)
79 def test_constructor_without_logger():
81 assert fs
.logger
== logging
.getLogger('fs')
82 assert fs
.path
is None
85 def test_constructor_with_logger():
86 logger_name
= 'fs_local'
87 fs
= FsLocal(logger_name
=logger_name
)
88 assert fs
.logger
== logging
.getLogger(logger_name
)
89 assert fs
.path
is None
92 def test_get_params(fs_local
):
93 params
= fs_local
.get_params()
94 assert len(params
) == 2
96 assert "path" in params
97 assert params
["fs"] == "local"
98 assert params
["path"] == valid_path()
101 @pytest.mark
.parametrize("config, exp_logger, exp_path", [
102 ({'logger_name': 'fs_local', 'path': valid_path()}, 'fs_local', valid_path()),
103 ({'logger_name': 'fs_local', 'path': valid_path()[:-1]}, 'fs_local', valid_path()),
104 ({'path': valid_path()}, 'fs', valid_path()),
105 ({'path': valid_path()[:-1]}, 'fs', valid_path())])
106 def test_fs_connect_with_valid_config(config
, exp_logger
, exp_path
):
108 fs
.fs_connect(config
)
109 assert fs
.logger
== logging
.getLogger(exp_logger
)
110 assert fs
.path
== exp_path
113 @pytest.mark
.parametrize("config, exp_exception_message", [
114 ({'logger_name': 'fs_local', 'path': invalid_path()}, fs_connect_exception_message(invalid_path())),
115 ({'logger_name': 'fs_local', 'path': invalid_path()[:-1]}, fs_connect_exception_message(invalid_path()[:-1])),
116 ({'path': invalid_path()}, fs_connect_exception_message(invalid_path())),
117 ({'path': invalid_path()[:-1]}, fs_connect_exception_message(invalid_path()[:-1]))])
118 def test_fs_connect_with_invalid_path(config
, exp_exception_message
):
120 with pytest
.raises(FsException
) as excinfo
:
121 fs
.fs_connect(config
)
122 assert str(excinfo
.value
) == exp_exception_message
125 def test_fs_disconnect(fs_local
):
126 fs_local
.fs_disconnect()
129 def test_mkdir_with_valid_path(fs_local
):
130 folder_name
= str(uuid
.uuid4())
131 folder_path
= valid_path() + folder_name
132 fs_local
.mkdir(folder_name
)
133 assert os
.path
.exists(folder_path
)
135 fs_local
.mkdir(folder_name
)
136 assert os
.path
.exists(folder_path
)
137 os
.rmdir(folder_path
)
140 def test_mkdir_with_exception(fs_local
):
141 folder_name
= str(uuid
.uuid4())
142 with pytest
.raises(FsException
) as excinfo
:
143 fs_local
.mkdir(folder_name
+ "/" + folder_name
)
144 assert excinfo
.value
.http_code
== http
.HTTPStatus
.INTERNAL_SERVER_ERROR
147 @pytest.mark
.parametrize("storage, mode, expected", [
148 (str(uuid
.uuid4()), 'file', False),
149 ([str(uuid
.uuid4())], 'file', False),
150 (str(uuid
.uuid4()), 'dir', False),
151 ([str(uuid
.uuid4())], 'dir', False)])
152 def test_file_exists_returns_false(fs_local
, storage
, mode
, expected
):
153 assert fs_local
.file_exists(storage
, mode
) == expected
156 @pytest.mark
.parametrize("storage, mode, expected", [
157 (str(uuid
.uuid4()), 'file', True),
158 ([str(uuid
.uuid4())], 'file', True),
159 (str(uuid
.uuid4()), 'dir', True),
160 ([str(uuid
.uuid4())], 'dir', True)])
161 def test_file_exists_returns_true(fs_local
, storage
, mode
, expected
):
162 path
= valid_path() + storage
if isinstance(storage
, str) else valid_path() + storage
[0]
167 assert fs_local
.file_exists(storage
, mode
) == expected
174 @pytest.mark
.parametrize("storage, mode", [
175 (str(uuid
.uuid4()), 'file'),
176 ([str(uuid
.uuid4())], 'file'),
177 (str(uuid
.uuid4()), 'dir'),
178 ([str(uuid
.uuid4())], 'dir')])
179 def test_file_size(fs_local
, storage
, mode
):
180 path
= valid_path() + storage
if isinstance(storage
, str) else valid_path() + storage
[0]
185 size
= os
.path
.getsize(path
)
186 assert fs_local
.file_size(storage
) == size
193 @pytest.mark
.parametrize("files, path", [
194 (['foo', 'bar', 'foobar'], str(uuid
.uuid4())),
195 (['foo', 'bar', 'foobar'], [str(uuid
.uuid4())])])
196 def test_file_extract(fs_local
, files
, path
):
198 os
.mknod(valid_path() + f
)
199 tar_path
= valid_path() + str(uuid
.uuid4()) + '.tar'
200 with tarfile
.open(tar_path
, 'w') as tar
:
202 tar
.add(valid_path() + f
, arcname
=f
)
203 with tarfile
.open(tar_path
, 'r') as tar
:
204 fs_local
.file_extract(tar
, path
)
205 extracted_path
= valid_path() + (path
if isinstance(path
, str) else '/'.join(path
))
206 ls_dir
= os
.listdir(extracted_path
)
207 assert len(ls_dir
) == len(files
)
212 os
.remove(valid_path() + f
)
213 shutil
.rmtree(extracted_path
)
216 @pytest.mark
.parametrize("storage, mode", [
217 (str(uuid
.uuid4()), 'r'),
218 (str(uuid
.uuid4()), 'w'),
219 (str(uuid
.uuid4()), 'a'),
220 (str(uuid
.uuid4()), 'rb'),
221 (str(uuid
.uuid4()), 'wb'),
222 (str(uuid
.uuid4()), 'ab'),
223 ([str(uuid
.uuid4())], 'r'),
224 ([str(uuid
.uuid4())], 'w'),
225 ([str(uuid
.uuid4())], 'a'),
226 ([str(uuid
.uuid4())], 'rb'),
227 ([str(uuid
.uuid4())], 'wb'),
228 ([str(uuid
.uuid4())], 'ab')])
229 def test_file_open(fs_local
, storage
, mode
):
230 path
= valid_path() + storage
if isinstance(storage
, str) else valid_path() + storage
[0]
232 file_obj
= fs_local
.file_open(storage
, mode
)
233 assert isinstance(file_obj
, io
.IOBase
)
234 assert file_obj
.closed
is False
238 @pytest.mark
.parametrize("storage, mode", [
239 (str(uuid
.uuid4()), 'r'),
240 (str(uuid
.uuid4()), 'rb'),
241 ([str(uuid
.uuid4())], 'r'),
242 ([str(uuid
.uuid4())], 'rb')])
243 def test_file_open_file_not_found_exception(fs_local
, storage
, mode
):
244 with pytest
.raises(FsException
) as excinfo
:
245 fs_local
.file_open(storage
, mode
)
246 assert str(excinfo
.value
) == file_open_file_not_found_exception(storage
)
247 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
250 @pytest.mark
.parametrize("storage, mode", [
251 (str(uuid
.uuid4()), 'r'),
252 (str(uuid
.uuid4()), 'w'),
253 (str(uuid
.uuid4()), 'a'),
254 (str(uuid
.uuid4()), 'rb'),
255 (str(uuid
.uuid4()), 'wb'),
256 (str(uuid
.uuid4()), 'ab'),
257 ([str(uuid
.uuid4())], 'r'),
258 ([str(uuid
.uuid4())], 'w'),
259 ([str(uuid
.uuid4())], 'a'),
260 ([str(uuid
.uuid4())], 'rb'),
261 ([str(uuid
.uuid4())], 'wb'),
262 ([str(uuid
.uuid4())], 'ab')])
263 def test_file_open_io_error(fs_local
, storage
, mode
):
264 path
= valid_path() + storage
if isinstance(storage
, str) else valid_path() + storage
[0]
267 with pytest
.raises(FsException
) as excinfo
:
268 fs_local
.file_open(storage
, mode
)
269 assert str(excinfo
.value
) == file_open_io_exception(storage
)
270 assert excinfo
.value
.http_code
== http
.HTTPStatus
.BAD_REQUEST
274 @pytest.mark
.parametrize("storage, with_files", [
275 (str(uuid
.uuid4()), True),
276 (str(uuid
.uuid4()), False),
277 ([str(uuid
.uuid4())], True),
278 ([str(uuid
.uuid4())], False)])
279 def test_dir_ls(fs_local
, storage
, with_files
):
280 path
= valid_path() + storage
if isinstance(storage
, str) else valid_path() + storage
[0]
282 if with_files
is True:
283 file_name
= str(uuid
.uuid4())
284 file_path
= path
+ '/' + file_name
286 result
= fs_local
.dir_ls(storage
)
288 if with_files
is True:
289 assert len(result
) == 1
290 assert result
[0] == file_name
292 assert len(result
) == 0
296 @pytest.mark
.parametrize("storage", [
298 ([str(uuid
.uuid4())])])
299 def test_dir_ls_with_not_a_directory_error(fs_local
, storage
):
300 path
= valid_path() + storage
if isinstance(storage
, str) else valid_path() + storage
[0]
302 with pytest
.raises(FsException
) as excinfo
:
303 fs_local
.dir_ls(storage
)
304 assert str(excinfo
.value
) == dir_ls_not_a_directory_exception(storage
)
305 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
309 @pytest.mark
.parametrize("storage", [
311 ([str(uuid
.uuid4())])])
312 def test_dir_ls_with_io_error(fs_local
, storage
):
313 path
= valid_path() + storage
if isinstance(storage
, str) else valid_path() + storage
[0]
316 with pytest
.raises(FsException
) as excinfo
:
317 fs_local
.dir_ls(storage
)
318 assert str(excinfo
.value
) == dir_ls_io_exception(storage
)
319 assert excinfo
.value
.http_code
== http
.HTTPStatus
.BAD_REQUEST
323 @pytest.mark
.parametrize("storage, with_files, ignore_non_exist", [
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 ([str(uuid
.uuid4())], True, True),
329 ([str(uuid
.uuid4())], False, True),
330 ([str(uuid
.uuid4())], True, False),
331 ([str(uuid
.uuid4())], False, False)])
332 def test_file_delete_with_dir(fs_local
, storage
, with_files
, ignore_non_exist
):
333 path
= valid_path() + storage
if isinstance(storage
, str) else valid_path() + storage
[0]
335 if with_files
is True:
336 file_path
= path
+ '/' + str(uuid
.uuid4())
338 fs_local
.file_delete(storage
, ignore_non_exist
)
339 assert os
.path
.exists(path
) is False
342 @pytest.mark
.parametrize("storage", [
344 ([str(uuid
.uuid4())])])
345 def test_file_delete_expect_exception(fs_local
, storage
):
346 with pytest
.raises(FsException
) as excinfo
:
347 fs_local
.file_delete(storage
)
348 assert str(excinfo
.value
) == file_delete_exception_message(storage
)
349 assert excinfo
.value
.http_code
== http
.HTTPStatus
.NOT_FOUND
352 @pytest.mark
.parametrize("storage", [
354 ([str(uuid
.uuid4())])])
355 def test_file_delete_no_exception(fs_local
, storage
):
356 path
= valid_path() + storage
if isinstance(storage
, str) else valid_path() + storage
[0]
357 fs_local
.file_delete(storage
, ignore_non_exist
=True)
358 assert os
.path
.exists(path
) is False