1 # Copyright (c) 2019 AT&T Intellectual Property.
2 # Copyright (c) 2018-2019 Nokia.
4 # Licensed under the Apache License, Version 2.0 (the "License");
5 # you may not use this file except in compliance with the License.
6 # You may obtain 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,
12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 # See the License for the specific language governing permissions and
14 # limitations under the License.
17 # This source code is part of the near-RT RIC (RAN Intelligent Controller)
18 # platform project (RICP).
22 from unittest.mock import patch, Mock
24 from ricsdl.syncstorage import SyncStorage
25 from ricsdl.syncstorage import SyncLock
26 from ricsdl.syncstorage import func_arg_checker
27 from ricsdl.exceptions import SdlTypeError
31 def sync_storage_fixture(request):
32 request.cls.ns = 'some-ns'
34 request.cls.keys = {'a', 'b'}
35 request.cls.dm = {'a': b'1', 'b': b'2'}
36 request.cls.old_data = b'1'
37 request.cls.new_data = b'3'
38 request.cls.keyprefix = 'x'
39 request.cls.matchedkeys = ['x1', 'x2', 'x3', 'x4', 'x5']
40 request.cls.is_atomic = True
41 request.cls.group = 'some-group'
42 request.cls.groupmembers = set([b'm1', b'm2'])
43 request.cls.groupmember = b'm1'
44 request.cls.lock_name = 'some-lock-name'
45 request.cls.lock_int_expiration = 10
46 request.cls.lock_float_expiration = 1.1
48 with patch('ricsdl.backend.get_backend_instance') as mock_db_backend:
49 storage = SyncStorage()
50 request.cls.mock_db_backend = mock_db_backend.return_value
51 request.cls.storage = storage
55 @pytest.mark.usefixtures('sync_storage_fixture')
56 class TestSyncStorage:
57 def test_set_function_success(self):
58 self.storage.set(self.ns, self.dm)
59 self.mock_db_backend.set.assert_called_once_with(self.ns, self.dm)
61 def test_set_function_can_raise_exception_for_wrong_argument(self):
62 with pytest.raises(SdlTypeError):
63 self.storage.set(123, {'a': b'v1'})
64 with pytest.raises(SdlTypeError):
65 self.storage.set('ns', [1, 2])
67 def test_set_if_function_success(self):
68 self.mock_db_backend.set_if.return_value = True
69 ret = self.storage.set_if(self.ns, self.key, self.old_data, self.new_data)
70 self.mock_db_backend.set_if.assert_called_once_with(self.ns, self.key, self.old_data,
74 def test_set_if_function_can_return_false_if_same_data_already_exists(self):
75 self.mock_db_backend.set_if.return_value = False
76 ret = self.storage.set_if(self.ns, self.key, self.old_data, self.new_data)
77 self.mock_db_backend.set_if.assert_called_once_with(self.ns, self.key, self.old_data,
81 def test_set_if_function_can_raise_exception_for_wrong_argument(self):
82 with pytest.raises(SdlTypeError):
83 self.storage.set_if(0xbad, 'key', b'v1', b'v2')
84 with pytest.raises(SdlTypeError):
85 self.storage.set_if('ns', 0xbad, b'v1', b'v2')
86 with pytest.raises(SdlTypeError):
87 self.storage.set_if('ns', 'key', 0xbad, b'v2')
88 with pytest.raises(SdlTypeError):
89 self.storage.set_if('ns', 'key', b'v1', 0xbad)
91 def test_set_if_not_exists_function_success(self):
92 self.mock_db_backend.set_if_not_exists.return_value = True
93 ret = self.storage.set_if_not_exists(self.ns, self.key, self.new_data)
94 self.mock_db_backend.set_if_not_exists.assert_called_once_with(self.ns, self.key,
98 def test_set_if_not_exists_function_can_return_false_if_key_already_exists(self):
99 self.mock_db_backend.set_if_not_exists.return_value = False
100 ret = self.storage.set_if_not_exists(self.ns, self.key, self.new_data)
101 self.mock_db_backend.set_if_not_exists.assert_called_once_with(self.ns, self.key,
105 def test_set_if_not_exists_function_can_raise_exception_for_wrong_argument(self):
106 with pytest.raises(SdlTypeError):
107 self.storage.set_if_not_exists(0xbad, 'key', b'v1')
108 with pytest.raises(SdlTypeError):
109 self.storage.set_if_not_exists('ns', 0xbad, b'v1')
110 with pytest.raises(SdlTypeError):
111 self.storage.set_if_not_exists('ns', 'key', 0xbad)
113 def test_get_function_success(self):
114 self.mock_db_backend.get.return_value = self.dm
115 ret = self.storage.get(self.ns, self.keys)
116 self.mock_db_backend.get.assert_called_once()
117 call_args, _ = self.mock_db_backend.get.call_args
118 assert call_args[0] == self.ns
119 assert len(call_args[1]) == len(self.keys)
120 assert all(k in call_args[1] for k in self.keys)
121 assert ret == self.dm
123 def test_get_function_can_return_empty_dict_when_no_key_values_exist(self):
124 self.mock_db_backend.get.return_value = dict()
125 ret = self.storage.get(self.ns, self.keys)
126 self.mock_db_backend.get.assert_called_once()
127 call_args, _ = self.mock_db_backend.get.call_args
128 assert call_args[0] == self.ns
129 assert len(call_args[1]) == len(self.keys)
130 assert all(k in call_args[1] for k in self.keys)
133 def test_get_function_can_raise_exception_for_wrong_argument(self):
134 with pytest.raises(SdlTypeError):
135 self.storage.get(0xbad, self.key)
136 with pytest.raises(SdlTypeError):
137 self.storage.get(self.ns, 0xbad)
139 def test_find_keys_function_success(self):
140 self.mock_db_backend.find_keys.return_value = self.matchedkeys
141 ret = self.storage.find_keys(self.ns, self.keyprefix)
142 self.mock_db_backend.find_keys.assert_called_once_with(self.ns, self.keyprefix)
143 assert ret == self.matchedkeys
145 def test_find_keys_function_can_return_empty_list_when_no_keys_exist(self):
146 self.mock_db_backend.find_keys.return_value = list()
147 ret = self.storage.find_keys(self.ns, self.keyprefix)
148 self.mock_db_backend.find_keys.assert_called_once_with(self.ns, self.keyprefix)
151 def test_find_keys_function_can_raise_exception_for_wrong_argument(self):
152 with pytest.raises(SdlTypeError):
153 self.storage.find_keys(0xbad, self.keyprefix)
154 with pytest.raises(SdlTypeError):
155 self.storage.find_keys(self.ns, 0xbad)
157 def test_find_and_get_function_success(self):
158 self.mock_db_backend.find_and_get.return_value = self.dm
159 ret = self.storage.find_and_get(self.ns, self.keyprefix, self.is_atomic)
160 self.mock_db_backend.find_and_get.assert_called_once_with(self.ns, self.keyprefix,
162 assert ret == self.dm
164 def test_find_and_get_function_can_return_empty_dict_when_no_keys_exist(self):
165 self.mock_db_backend.find_and_get.return_value = dict()
166 ret = self.storage.find_and_get(self.ns, self.keyprefix, self.is_atomic)
167 self.mock_db_backend.find_and_get.assert_called_once_with(self.ns, self.keyprefix,
171 def test_find_and_get_function_can_raise_exception_for_wrong_argument(self):
172 with pytest.raises(SdlTypeError):
173 self.storage.find_and_get(0xbad, self.keyprefix, self.is_atomic)
174 with pytest.raises(SdlTypeError):
175 self.storage.find_and_get(self.ns, 0xbad, self.is_atomic)
176 with pytest.raises(SdlTypeError):
177 self.storage.find_and_get(self.ns, self.keyprefix, 0xbad)
179 def test_remove_function_success(self):
180 self.storage.remove(self.ns, self.keys)
181 self.mock_db_backend.remove.assert_called_once()
182 call_args, _ = self.mock_db_backend.remove.call_args
183 assert call_args[0] == self.ns
184 assert isinstance(call_args[1], list)
185 assert len(call_args[1]) == len(self.keys)
186 assert all(k in call_args[1] for k in self.keys)
188 def test_remove_function_can_raise_exception_for_wrong_argument(self):
189 with pytest.raises(SdlTypeError):
190 self.storage.remove(0xbad, self.keys)
191 with pytest.raises(SdlTypeError):
192 self.storage.remove(self.ns, 0xbad)
194 def test_remove_if_function_success(self):
195 self.mock_db_backend.remove_if.return_value = True
196 ret = self.storage.remove_if(self.ns, self.key, self.new_data)
197 self.mock_db_backend.remove_if.assert_called_once_with(self.ns, self.key, self.new_data)
200 def test_remove_if_function_can_return_false_if_data_does_not_match(self):
201 self.mock_db_backend.remove_if.return_value = False
202 ret = self.storage.remove_if(self.ns, self.key, self.old_data)
203 self.mock_db_backend.remove_if.assert_called_once_with(self.ns, self.key, self.old_data)
206 def test_remove_if_function_can_raise_exception_for_wrong_argument(self):
207 with pytest.raises(SdlTypeError):
208 self.storage.remove_if(0xbad, self.keys, self.old_data)
209 with pytest.raises(SdlTypeError):
210 self.storage.remove_if(self.ns, 0xbad, self.old_data)
211 with pytest.raises(SdlTypeError):
212 self.storage.remove_if(self.ns, self.keys, 0xbad)
214 def test_remove_all_function_success(self):
215 self.mock_db_backend.find_keys.return_value = ['a1']
216 self.storage.remove_all(self.ns)
217 self.mock_db_backend.find_keys.assert_called_once_with(self.ns, '')
218 self.mock_db_backend.remove.assert_called_once_with(self.ns,
219 self.mock_db_backend.find_keys.return_value)
221 def test_remove_all_function_can_raise_exception_for_wrong_argument(self):
222 with pytest.raises(SdlTypeError):
223 self.storage.remove_all(0xbad)
225 def test_add_member_function_success(self):
226 self.storage.add_member(self.ns, self.group, self.groupmembers)
227 self.mock_db_backend.add_member.assert_called_once_with(self.ns,
228 self.group, self.groupmembers)
230 def test_add_member_function_can_raise_exception_for_wrong_argument(self):
231 with pytest.raises(SdlTypeError):
232 self.storage.add_member(0xbad, self.group, self.groupmembers)
233 with pytest.raises(SdlTypeError):
234 self.storage.add_member(self.ns, 0xbad, self.groupmembers)
235 with pytest.raises(SdlTypeError):
236 self.storage.add_member(self.ns, self.group, 0xbad)
238 def test_remove_member_function_success(self):
239 self.storage.remove_member(self.ns, self.group, self.groupmembers)
240 self.mock_db_backend.remove_member.assert_called_once_with(self.ns, self.group,
243 def test_remove_member_function_can_raise_exception_for_wrong_argument(self):
244 with pytest.raises(SdlTypeError):
245 self.storage.remove_member(0xbad, self.group, self.groupmembers)
246 with pytest.raises(SdlTypeError):
247 self.storage.remove_member(self.ns, 0xbad, self.groupmembers)
248 with pytest.raises(SdlTypeError):
249 self.storage.remove_member(self.ns, self.group, 0xbad)
251 def test_remove_group_function_success(self):
252 self.storage.remove_group(self.ns, self.group)
253 self.mock_db_backend.remove_group.assert_called_once_with(self.ns, self.group)
255 def test_remove_group_function_can_raise_exception_for_wrong_argument(self):
256 with pytest.raises(SdlTypeError):
257 self.storage.remove_group(0xbad, self.group)
258 with pytest.raises(SdlTypeError):
259 self.storage.remove_group(self.ns, 0xbad)
261 def test_get_members_function_success(self):
262 self.mock_db_backend.get_members.return_value = self.groupmembers
263 ret = self.storage.get_members(self.ns, self.group)
264 self.mock_db_backend.get_members.assert_called_once_with(self.ns, self.group)
265 assert ret == self.groupmembers
267 def test_get_members_function_can_raise_exception_for_wrong_argument(self):
268 with pytest.raises(SdlTypeError):
269 self.storage.get_members(0xbad, self.group)
270 with pytest.raises(SdlTypeError):
271 self.storage.get_members(self.ns, 0xbad)
273 def test_is_member_function_success(self):
274 self.mock_db_backend.is_member.return_value = True
275 ret = self.storage.is_member(self.ns, self.group, self.groupmember)
276 self.mock_db_backend.is_member.assert_called_once_with(self.ns, self.group,
280 def test_is_member_function_can_raise_exception_for_wrong_argument(self):
281 with pytest.raises(SdlTypeError):
282 self.storage.is_member(0xbad, self.group, self.groupmember)
283 with pytest.raises(SdlTypeError):
284 self.storage.is_member(self.ns, 0xbad, self.groupmember)
285 with pytest.raises(SdlTypeError):
286 self.storage.is_member(self.ns, self.group, 0xbad)
288 def test_group_size_function_success(self):
289 self.mock_db_backend.group_size.return_value = 100
290 ret = self.storage.group_size(self.ns, self.group)
291 self.mock_db_backend.group_size.assert_called_once_with(self.ns, self.group)
294 def test_group_size_function_can_raise_exception_for_wrong_argument(self):
295 with pytest.raises(SdlTypeError):
296 self.storage.group_size(0xbad, self.group)
297 with pytest.raises(SdlTypeError):
298 self.storage.group_size(self.ns, 0xbad)
300 @patch('ricsdl.syncstorage.SyncLock')
301 def test_get_lock_resource_function_success_when_expiration_time_is_integer(self, mock_db_lock):
302 ret = self.storage.get_lock_resource(self.ns, self.lock_name, self.lock_int_expiration)
303 mock_db_lock.assert_called_once_with(self.ns, self.lock_name, self.lock_int_expiration,
305 assert ret == mock_db_lock.return_value
307 @patch('ricsdl.syncstorage.SyncLock')
308 def test_get_lock_resource_function_success_when_expiration_time_is_float_number(self,
310 ret = self.storage.get_lock_resource(self.ns, self.lock_name, self.lock_float_expiration)
311 mock_db_lock.assert_called_once_with(self.ns, self.lock_name, self.lock_float_expiration,
313 assert ret == mock_db_lock.return_value
315 def test_get_lock_resource_function_can_raise_exception_for_wrong_argument(self):
316 with pytest.raises(SdlTypeError):
317 self.storage.get_lock_resource(0xbad, self.lock_name, self.lock_int_expiration)
318 with pytest.raises(SdlTypeError):
319 self.storage.get_lock_resource(self.ns, 0xbad, self.lock_int_expiration)
320 with pytest.raises(SdlTypeError):
321 self.storage.get_lock_resource(self.ns, self.lock_name, 'bad')
323 def test_get_backend_function_success(self):
324 ret = self.storage.get_backend()
325 assert ret == self.mock_db_backend
327 def test_storage_object_string_representation(self):
328 str_out = str(self.storage)
329 assert str_out is not None
333 def lock_fixture(request):
334 request.cls.ns = 'some-ns'
335 request.cls.lockname = 'some-lock-name'
336 request.cls.expiration = 10
337 request.cls.retry_interval = 0.1
338 request.cls.retry_timeout = 1
340 with patch('ricsdl.backend.get_backend_lock_instance') as mock_db_backend_lock:
341 lock = SyncLock('test-ns', 'test-lock-name', request.cls.expiration, Mock())
342 request.cls.mock_db_backend_lock = mock_db_backend_lock.return_value
343 request.cls.lock = lock
347 @pytest.mark.usefixtures('lock_fixture')
349 def test_acquire_function_success_when_timeout_and_interval_are_integers(self):
350 self.lock.acquire(self.retry_interval, self.retry_timeout)
351 self.mock_db_backend_lock.acquire.assert_called_once_with(self.retry_interval,
354 def test_acquire_function_success_when_timeout_and_interval_are_float_numbers(self):
355 self.lock.acquire(float(self.retry_interval), float(self.retry_timeout))
356 self.mock_db_backend_lock.acquire.assert_called_once_with(float(self.retry_interval),
357 float(self.retry_timeout))
359 def test_acquire_function_can_raise_exception_for_wrong_argument(self):
360 with pytest.raises(SdlTypeError):
361 self.lock.acquire('bad', self.retry_timeout)
362 with pytest.raises(SdlTypeError):
363 self.lock.acquire(self.retry_interval, 'bad')
365 def test_release_function_success(self):
367 self.mock_db_backend_lock.release.assert_called_once()
369 def test_refresh_function_success(self):
371 self.mock_db_backend_lock.refresh.assert_called_once()
373 def test_get_validity_time_function_success(self):
374 self.mock_db_backend_lock.get_validity_time.return_value = self.expiration
375 ret = self.lock.get_validity_time()
376 self.mock_db_backend_lock.get_validity_time.assert_called_once()
377 assert ret == self.expiration
379 def test_get_validity_time_function_success_when_returned_time_is_float(self):
380 self.mock_db_backend_lock.get_validity_time.return_value = float(self.expiration)
381 ret = self.lock.get_validity_time()
382 self.mock_db_backend_lock.get_validity_time.assert_called_once()
383 assert ret == float(self.expiration)
385 def test_lock_object_string_representation(self):
386 str_out = str(self.lock)
387 assert str_out is not None
390 def test_function_arg_validator():
391 @func_arg_checker(SdlTypeError, 0, a=str, b=(int, float), c=set, d=(dict, type(None)))
392 def _my_func(a='abc', b=1, c={'x', 'y'}, d={'x': b'1'}):
394 with pytest.raises(SdlTypeError, match=r"Wrong argument type: 'a'=<class 'NoneType'>. "
395 r"Must be: <class 'str'>"):
398 with pytest.raises(SdlTypeError, match=r"Wrong argument type: 'b'=<class 'str'>. "):
399 _my_func('abc', 'wrong type')
401 with pytest.raises(SdlTypeError, match=r"Wrong argument type: 'c'=<class 'str'>. "
402 r"Must be: <class 'set'>"):
403 _my_func('abc', 1.0, 'wrong type')
405 with pytest.raises(SdlTypeError, match=r"Wrong argument type: 'd'=<class 'str'>. "):
406 _my_func('abc', 1.0, {'x', 'y'}, 'wrong type')
409 def test_function_kwarg_validator():
410 @func_arg_checker(SdlTypeError, 0, a=str, b=(int, float), c=set, d=(dict, type(None)))
411 def _my_func(a='abc', b=1, c={'x', 'y'}, d={'x': b'1'}):
413 with pytest.raises(SdlTypeError, match=r"Wrong argument type: 'a'=<class 'NoneType'>. "
414 r"Must be: <class 'str'>"):
417 with pytest.raises(SdlTypeError, match=r"Wrong argument type: 'b'=<class 'str'>. "):
418 _my_func(b='wrong type')
420 with pytest.raises(SdlTypeError, match=r"Wrong argument type: 'c'=<class 'str'>. "
421 r"Must be: <class 'set'>"):
422 _my_func(c='wrong type')
424 with pytest.raises(SdlTypeError, match=r"Wrong argument type: 'd'=<class 'str'>. "):
425 _my_func(d='wrong type')