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, NotConnected)
31 def sync_storage_fixture(request):
32 request.cls.ns = 'some-ns'
34 request.cls.keys = {'a', 'b'}
35 request.cls.dm = {'b': b'2', 'a': b'1'}
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.group = 'some-group'
41 request.cls.groupmembers = set([b'm1', b'm2'])
42 request.cls.groupmember = b'm1'
43 request.cls.lock_name = 'some-lock-name'
44 request.cls.lock_int_expiration = 10
45 request.cls.lock_float_expiration = 1.1
46 request.cls.channels = {'abs', 'cbn'}
47 request.cls.channels_and_events = {'abs': 'cbn'}
49 with patch('ricsdl.backend.get_backend_instance') as mock_db_backend:
50 storage = SyncStorage()
51 request.cls.mock_db_backend = mock_db_backend.return_value
52 request.cls.storage = storage
56 @pytest.mark.usefixtures('sync_storage_fixture')
57 class TestSyncStorage:
58 def test_is_active_function_success(self):
59 self.mock_db_backend.is_connected.return_value = True
60 ret = self.storage.is_active()
61 self.mock_db_backend.is_connected.assert_called_once()
64 def test_is_active_function_can_catch_backend_exception_and_return_false(self):
65 self.mock_db_backend.is_connected.side_effect = NotConnected
66 ret = self.storage.is_active()
67 self.mock_db_backend.is_connected.assert_called_once()
70 def test_set_function_success(self):
71 self.storage.set(self.ns, self.dm)
72 self.mock_db_backend.set.assert_called_once_with(self.ns, self.dm)
74 def test_set_function_can_raise_exception_for_wrong_argument(self):
75 with pytest.raises(SdlTypeError):
76 self.storage.set(123, {'a': b'v1'})
77 with pytest.raises(SdlTypeError):
78 self.storage.set('ns', [1, 2])
79 with pytest.raises(SdlTypeError):
80 self.storage.set('ns', {0xbad: b'v1'})
81 with pytest.raises(SdlTypeError):
82 self.storage.set('ns', {'a': 0xbad})
84 def test_set_if_function_success(self):
85 self.mock_db_backend.set_if.return_value = True
86 ret = self.storage.set_if(self.ns, self.key, self.old_data, self.new_data)
87 self.mock_db_backend.set_if.assert_called_once_with(self.ns, self.key, self.old_data,
91 def test_set_if_function_can_return_false_if_same_data_already_exists(self):
92 self.mock_db_backend.set_if.return_value = False
93 ret = self.storage.set_if(self.ns, self.key, self.old_data, self.new_data)
94 self.mock_db_backend.set_if.assert_called_once_with(self.ns, self.key, self.old_data,
98 def test_set_if_function_can_raise_exception_for_wrong_argument(self):
99 with pytest.raises(SdlTypeError):
100 self.storage.set_if(0xbad, 'key', b'v1', b'v2')
101 with pytest.raises(SdlTypeError):
102 self.storage.set_if('ns', 0xbad, b'v1', b'v2')
103 with pytest.raises(SdlTypeError):
104 self.storage.set_if('ns', 'key', 0xbad, b'v2')
105 with pytest.raises(SdlTypeError):
106 self.storage.set_if('ns', 'key', b'v1', 0xbad)
108 def test_set_if_not_exists_function_success(self):
109 self.mock_db_backend.set_if_not_exists.return_value = True
110 ret = self.storage.set_if_not_exists(self.ns, self.key, self.new_data)
111 self.mock_db_backend.set_if_not_exists.assert_called_once_with(self.ns, self.key,
115 def test_set_if_not_exists_function_can_return_false_if_key_already_exists(self):
116 self.mock_db_backend.set_if_not_exists.return_value = False
117 ret = self.storage.set_if_not_exists(self.ns, self.key, self.new_data)
118 self.mock_db_backend.set_if_not_exists.assert_called_once_with(self.ns, self.key,
122 def test_set_if_not_exists_function_can_raise_exception_for_wrong_argument(self):
123 with pytest.raises(SdlTypeError):
124 self.storage.set_if_not_exists(0xbad, 'key', b'v1')
125 with pytest.raises(SdlTypeError):
126 self.storage.set_if_not_exists('ns', 0xbad, b'v1')
127 with pytest.raises(SdlTypeError):
128 self.storage.set_if_not_exists('ns', 'key', 0xbad)
130 def test_get_function_success(self):
131 self.mock_db_backend.get.return_value = self.dm
132 ret = self.storage.get(self.ns, self.keys)
133 self.mock_db_backend.get.assert_called_once()
134 call_args, _ = self.mock_db_backend.get.call_args
135 assert call_args[0] == self.ns
136 assert len(call_args[1]) == len(self.keys)
137 assert all(k in call_args[1] for k in self.keys)
138 assert ret == self.dm
139 # Validate that SDL returns a dictionary with keys in alphabetical order
140 assert sorted(self.dm)[0] == list(ret.keys())[0]
142 def test_get_function_can_return_empty_dict_when_no_key_values_exist(self):
143 self.mock_db_backend.get.return_value = dict()
144 ret = self.storage.get(self.ns, self.keys)
145 self.mock_db_backend.get.assert_called_once()
146 call_args, _ = self.mock_db_backend.get.call_args
147 assert call_args[0] == self.ns
148 assert len(call_args[1]) == len(self.keys)
149 assert all(k in call_args[1] for k in self.keys)
152 def test_get_function_can_raise_exception_for_wrong_argument(self):
153 with pytest.raises(SdlTypeError):
154 self.storage.get(0xbad, self.key)
155 with pytest.raises(SdlTypeError):
156 self.storage.get(self.ns, 0xbad)
158 def test_find_keys_function_success(self):
159 self.mock_db_backend.find_keys.return_value = self.matchedkeys
160 ret = self.storage.find_keys(self.ns, self.keyprefix)
161 self.mock_db_backend.find_keys.assert_called_once_with(self.ns, self.keyprefix)
162 assert ret == self.matchedkeys
164 def test_find_keys_function_can_return_empty_list_when_no_keys_exist(self):
165 self.mock_db_backend.find_keys.return_value = list()
166 ret = self.storage.find_keys(self.ns, self.keyprefix)
167 self.mock_db_backend.find_keys.assert_called_once_with(self.ns, self.keyprefix)
170 def test_find_keys_function_can_raise_exception_for_wrong_argument(self):
171 with pytest.raises(SdlTypeError):
172 self.storage.find_keys(0xbad, self.keyprefix)
173 with pytest.raises(SdlTypeError):
174 self.storage.find_keys(self.ns, 0xbad)
176 def test_find_and_get_function_success(self):
177 self.mock_db_backend.find_and_get.return_value = self.dm
178 ret = self.storage.find_and_get(self.ns, self.keyprefix)
179 self.mock_db_backend.find_and_get.assert_called_once_with(self.ns, self.keyprefix)
180 assert ret == self.dm
181 # Validate that SDL returns a dictionary with keys in alphabetical order
182 assert sorted(self.dm)[0] == list(ret.keys())[0]
184 def test_find_and_get_function_can_return_empty_dict_when_no_keys_exist(self):
185 self.mock_db_backend.find_and_get.return_value = dict()
186 ret = self.storage.find_and_get(self.ns, self.keyprefix)
187 self.mock_db_backend.find_and_get.assert_called_once_with(self.ns, self.keyprefix)
190 def test_find_and_get_function_can_raise_exception_for_wrong_argument(self):
191 with pytest.raises(SdlTypeError):
192 self.storage.find_and_get(0xbad, self.keyprefix)
193 with pytest.raises(SdlTypeError):
194 self.storage.find_and_get(self.ns, 0xbad)
196 def test_remove_function_success(self):
197 self.storage.remove(self.ns, self.keys)
198 self.mock_db_backend.remove.assert_called_once()
199 call_args, _ = self.mock_db_backend.remove.call_args
200 assert call_args[0] == self.ns
201 assert isinstance(call_args[1], list)
202 assert len(call_args[1]) == len(self.keys)
203 assert all(k in call_args[1] for k in self.keys)
205 def test_remove_function_can_raise_exception_for_wrong_argument(self):
206 with pytest.raises(SdlTypeError):
207 self.storage.remove(0xbad, self.keys)
208 with pytest.raises(SdlTypeError):
209 self.storage.remove(self.ns, 0xbad)
211 def test_remove_if_function_success(self):
212 self.mock_db_backend.remove_if.return_value = True
213 ret = self.storage.remove_if(self.ns, self.key, self.new_data)
214 self.mock_db_backend.remove_if.assert_called_once_with(self.ns, self.key, self.new_data)
217 def test_remove_if_function_can_return_false_if_data_does_not_match(self):
218 self.mock_db_backend.remove_if.return_value = False
219 ret = self.storage.remove_if(self.ns, self.key, self.old_data)
220 self.mock_db_backend.remove_if.assert_called_once_with(self.ns, self.key, self.old_data)
223 def test_remove_if_function_can_raise_exception_for_wrong_argument(self):
224 with pytest.raises(SdlTypeError):
225 self.storage.remove_if(0xbad, self.keys, self.old_data)
226 with pytest.raises(SdlTypeError):
227 self.storage.remove_if(self.ns, 0xbad, self.old_data)
228 with pytest.raises(SdlTypeError):
229 self.storage.remove_if(self.ns, self.keys, 0xbad)
231 def test_remove_all_function_success(self):
232 self.mock_db_backend.find_keys.return_value = ['a1']
233 self.storage.remove_all(self.ns)
234 self.mock_db_backend.find_keys.assert_called_once_with(self.ns, '*')
235 self.mock_db_backend.remove.assert_called_once_with(self.ns,
236 self.mock_db_backend.find_keys.return_value)
238 def test_remove_all_function_can_raise_exception_for_wrong_argument(self):
239 with pytest.raises(SdlTypeError):
240 self.storage.remove_all(0xbad)
242 def test_add_member_function_success(self):
243 self.storage.add_member(self.ns, self.group, self.groupmembers)
244 self.mock_db_backend.add_member.assert_called_once_with(self.ns,
245 self.group, self.groupmembers)
247 def test_add_member_function_can_raise_exception_for_wrong_argument(self):
248 with pytest.raises(SdlTypeError):
249 self.storage.add_member(0xbad, self.group, self.groupmembers)
250 with pytest.raises(SdlTypeError):
251 self.storage.add_member(self.ns, 0xbad, self.groupmembers)
252 with pytest.raises(SdlTypeError):
253 self.storage.add_member(self.ns, self.group, 0xbad)
255 def test_remove_member_function_success(self):
256 self.storage.remove_member(self.ns, self.group, self.groupmembers)
257 self.mock_db_backend.remove_member.assert_called_once_with(self.ns, self.group,
260 def test_remove_member_function_can_raise_exception_for_wrong_argument(self):
261 with pytest.raises(SdlTypeError):
262 self.storage.remove_member(0xbad, self.group, self.groupmembers)
263 with pytest.raises(SdlTypeError):
264 self.storage.remove_member(self.ns, 0xbad, self.groupmembers)
265 with pytest.raises(SdlTypeError):
266 self.storage.remove_member(self.ns, self.group, 0xbad)
268 def test_remove_group_function_success(self):
269 self.storage.remove_group(self.ns, self.group)
270 self.mock_db_backend.remove_group.assert_called_once_with(self.ns, self.group)
272 def test_remove_group_function_can_raise_exception_for_wrong_argument(self):
273 with pytest.raises(SdlTypeError):
274 self.storage.remove_group(0xbad, self.group)
275 with pytest.raises(SdlTypeError):
276 self.storage.remove_group(self.ns, 0xbad)
278 def test_get_members_function_success(self):
279 self.mock_db_backend.get_members.return_value = self.groupmembers
280 ret = self.storage.get_members(self.ns, self.group)
281 self.mock_db_backend.get_members.assert_called_once_with(self.ns, self.group)
282 assert ret == self.groupmembers
284 def test_get_members_function_can_raise_exception_for_wrong_argument(self):
285 with pytest.raises(SdlTypeError):
286 self.storage.get_members(0xbad, self.group)
287 with pytest.raises(SdlTypeError):
288 self.storage.get_members(self.ns, 0xbad)
290 def test_is_member_function_success(self):
291 self.mock_db_backend.is_member.return_value = True
292 ret = self.storage.is_member(self.ns, self.group, self.groupmember)
293 self.mock_db_backend.is_member.assert_called_once_with(self.ns, self.group,
297 def test_is_member_function_can_raise_exception_for_wrong_argument(self):
298 with pytest.raises(SdlTypeError):
299 self.storage.is_member(0xbad, self.group, self.groupmember)
300 with pytest.raises(SdlTypeError):
301 self.storage.is_member(self.ns, 0xbad, self.groupmember)
302 with pytest.raises(SdlTypeError):
303 self.storage.is_member(self.ns, self.group, 0xbad)
305 def test_group_size_function_success(self):
306 self.mock_db_backend.group_size.return_value = 100
307 ret = self.storage.group_size(self.ns, self.group)
308 self.mock_db_backend.group_size.assert_called_once_with(self.ns, self.group)
311 def test_group_size_function_can_raise_exception_for_wrong_argument(self):
312 with pytest.raises(SdlTypeError):
313 self.storage.group_size(0xbad, self.group)
314 with pytest.raises(SdlTypeError):
315 self.storage.group_size(self.ns, 0xbad)
317 def test_set_and_publish_function_success(self):
318 self.storage.set_and_publish(self.ns, self.channels_and_events, self.dm)
319 self.mock_db_backend.set_and_publish.assert_called_once_with(self.ns,
320 self.channels_and_events,
323 def test_set_and_publish_can_raise_exception_for_wrong_argument(self):
324 with pytest.raises(SdlTypeError):
325 self.storage.set_and_publish(123, self.channels_and_events, {'a': b'v1'})
326 with pytest.raises(SdlTypeError):
327 self.storage.set_and_publish('ns', self.channels_and_events, [1, 2])
328 with pytest.raises(SdlTypeError):
329 self.storage.set_and_publish('ns', self.channels_and_events, {0xbad: b'v1'})
330 with pytest.raises(SdlTypeError):
331 self.storage.set_and_publish('ns', self.channels_and_events, {'a': 0xbad})
333 def test_set_if_and_publish_success(self):
334 self.mock_db_backend.set_if_and_publish.return_value = True
335 ret = self.storage.set_if_and_publish(self.ns, self.channels_and_events, self.key,
336 self.old_data, self.new_data)
337 self.mock_db_backend.set_if_and_publish.assert_called_once_with(
338 self.ns, self.channels_and_events, self.key, self.old_data, self.new_data)
341 def test_set_if_and_publish_can_return_false_if_same_data_already_exists(self):
342 self.mock_db_backend.set_if_and_publish.return_value = False
343 ret = self.storage.set_if_and_publish(self.ns, self.channels_and_events, self.key,
344 self.old_data, self.new_data)
345 self.mock_db_backend.set_if_and_publish.assert_called_once_with(
346 self.ns, self.channels_and_events, self.key, self.old_data, self.new_data)
349 def test_set_if_and_publish_can_raise_exception_for_wrong_argument(self):
350 with pytest.raises(SdlTypeError):
351 self.storage.set_if_and_publish(0xbad, self.channels_and_events, 'key', b'v1', b'v2')
352 with pytest.raises(SdlTypeError):
353 self.storage.set_if_and_publish('ns', self.channels_and_events, 0xbad, b'v1', b'v2')
354 with pytest.raises(SdlTypeError):
355 self.storage.set_if_and_publish('ns', self.channels_and_events, 'key', 0xbad, b'v2')
356 with pytest.raises(SdlTypeError):
357 self.storage.set_if_and_publish('ns', self.channels_and_events, 'key', b'v1', 0xbad)
359 def test_set_if_not_exists_and_publish_success(self):
360 self.mock_db_backend.set_if_not_exists_and_publish.return_value = True
361 ret = self.storage.set_if_not_exists_and_publish(self.ns, self.channels_and_events,
362 self.key, self.new_data)
363 self.mock_db_backend.set_if_not_exists_and_publish.assert_called_once_with(
364 self.ns, self.channels_and_events, self.key, self.new_data)
367 def test_set_if_not_exists_and_publish_function_can_return_false_if_key_already_exists(self):
368 self.mock_db_backend.set_if_not_exists_and_publish.return_value = False
369 ret = self.storage.set_if_not_exists_and_publish(self.ns, self.channels_and_events,
370 self.key, self.new_data)
371 self.mock_db_backend.set_if_not_exists_and_publish.assert_called_once_with(
372 self.ns, self.channels_and_events, self.key, self.new_data)
375 def test_set_if_not_exists_and_publish_can_raise_exception_for_wrong_argument(self):
376 with pytest.raises(SdlTypeError):
377 self.storage.set_if_not_exists_and_publish(0xbad, self.channels_and_events, 'key',
379 with pytest.raises(SdlTypeError):
380 self.storage.set_if_not_exists_and_publish('ns', self.channels_and_events, 0xbad, b'v1')
381 with pytest.raises(SdlTypeError):
382 self.storage.set_if_not_exists_and_publish('ns', self.channels_and_events, 'key', 0xbad)
384 def test_remove_and_publish_function_success(self):
385 self.storage.remove_and_publish(self.ns, self.channels_and_events, self.keys)
386 self.mock_db_backend.remove_and_publish.assert_called_once_with(
387 self.ns, self.channels_and_events, list(self.keys))
389 def test_remove_and_publish_can_raise_exception_for_wrong_argument(self):
390 with pytest.raises(SdlTypeError):
391 self.storage.remove_and_publish(0xbad, self.channels_and_events, self.keys)
392 with pytest.raises(SdlTypeError):
393 self.storage.remove(self.ns, self.channels_and_events, 0xbad)
395 def test_remove_if_and_publish_success(self):
396 self.mock_db_backend.remove_if_and_publish.return_value = True
397 ret = self.storage.remove_if_and_publish(self.ns, self.channels_and_events, self.key,
399 self.mock_db_backend.remove_if_and_publish.assert_called_once_with(
400 self.ns, self.channels_and_events, self.key, self.new_data)
403 def test_remove_if_remove_and_publish_can_return_false_if_data_does_not_match(self):
404 self.mock_db_backend.remove_if_and_publish.return_value = False
405 ret = self.storage.remove_if_and_publish(self.ns, self.channels_and_events, self.key,
407 self.mock_db_backend.remove_if_and_publish.assert_called_once_with(
408 self.ns, self.channels_and_events, self.key, self.old_data)
411 def test_remove_if_remove_and_publish_can_raise_exception_for_wrong_argument(self):
412 with pytest.raises(SdlTypeError):
413 self.storage.remove_if_and_publish(0xbad, self.channels_and_events, self.keys,
415 with pytest.raises(SdlTypeError):
416 self.storage.remove_if_and_publish(self.ns, self.channels_and_events, 0xbad,
418 with pytest.raises(SdlTypeError):
419 self.storage.remove_if_and_publish(self.ns, self.channels_and_events, self.keys, 0xbad)
421 def test_remove_all_and_publish_success(self):
422 self.storage.remove_all_and_publish(self.ns, self.channels_and_events)
423 self.mock_db_backend.remove_all_and_publish.assert_called_once_with(
424 self.ns, self.channels_and_events)
426 def test_remove_all_and_publish_can_raise_exception_for_wrong_argument(self):
427 with pytest.raises(SdlTypeError):
428 self.storage.remove_all_and_publish(0xbad, self.channels_and_events)
430 def test_subscribe_function_success(self):
431 def cb(channel, message):
433 self.storage.subscribe_channel(self.ns, cb, self.channels)
434 self.mock_db_backend.subscribe_channel.assert_called_once_with(
435 self.ns, cb, list(self.channels))
437 def test_subscribe_can_raise_exception_for_wrong_argument(self):
438 def cb3(channel, message, extra):
442 with pytest.raises(SdlTypeError):
443 self.storage.subscribe_channel(self.ns, cb3, self.channels)
444 with pytest.raises(SdlTypeError):
445 self.storage.subscribe_channel(self.ns, cb1, self.channels)
447 def test_unsubscribe_function_success(self):
448 self.storage.unsubscribe_channel(self.ns, self.channels)
449 self.mock_db_backend.unsubscribe_channel.assert_called_once_with(
450 self.ns, list(self.channels))
452 def test_start_event_listener_success(self):
453 self.storage.start_event_listener()
454 self.mock_db_backend.start_event_listener.assert_called()
456 def test_handle_events_success(self):
457 self.storage.handle_events()
458 self.mock_db_backend.handle_events.assert_called()
460 @patch('ricsdl.syncstorage.SyncLock')
461 def test_get_lock_resource_function_success_when_expiration_time_is_integer(self, mock_db_lock):
462 ret = self.storage.get_lock_resource(self.ns, self.lock_name, self.lock_int_expiration)
463 mock_db_lock.assert_called_once_with(self.ns, self.lock_name, self.lock_int_expiration,
465 assert ret == mock_db_lock.return_value
467 @patch('ricsdl.syncstorage.SyncLock')
468 def test_get_lock_resource_function_success_when_expiration_time_is_float_number(self,
470 ret = self.storage.get_lock_resource(self.ns, self.lock_name, self.lock_float_expiration)
471 mock_db_lock.assert_called_once_with(self.ns, self.lock_name, self.lock_float_expiration,
473 assert ret == mock_db_lock.return_value
475 def test_get_lock_resource_function_can_raise_exception_for_wrong_argument(self):
476 with pytest.raises(SdlTypeError):
477 self.storage.get_lock_resource(0xbad, self.lock_name, self.lock_int_expiration)
478 with pytest.raises(SdlTypeError):
479 self.storage.get_lock_resource(self.ns, 0xbad, self.lock_int_expiration)
480 with pytest.raises(SdlTypeError):
481 self.storage.get_lock_resource(self.ns, self.lock_name, 'bad')
483 def test_get_backend_function_success(self):
484 ret = self.storage.get_backend()
485 assert ret == self.mock_db_backend
487 def test_storage_object_string_representation(self):
488 str_out = str(self.storage)
489 assert str_out is not None
493 def lock_fixture(request):
494 request.cls.ns = 'some-ns'
495 request.cls.lockname = 'some-lock-name'
496 request.cls.expiration = 10
497 request.cls.retry_interval = 0.1
498 request.cls.retry_timeout = 1
500 with patch('ricsdl.backend.get_backend_lock_instance') as mock_db_backend_lock:
501 lock = SyncLock('test-ns', 'test-lock-name', request.cls.expiration, Mock())
502 request.cls.mock_db_backend_lock = mock_db_backend_lock.return_value
503 request.cls.lock = lock
507 @pytest.mark.usefixtures('lock_fixture')
509 def test_acquire_function_success_when_timeout_and_interval_are_integers(self):
510 self.lock.acquire(self.retry_interval, self.retry_timeout)
511 self.mock_db_backend_lock.acquire.assert_called_once_with(self.retry_interval,
514 def test_acquire_function_success_when_timeout_and_interval_are_float_numbers(self):
515 self.lock.acquire(float(self.retry_interval), float(self.retry_timeout))
516 self.mock_db_backend_lock.acquire.assert_called_once_with(float(self.retry_interval),
517 float(self.retry_timeout))
519 def test_acquire_function_can_raise_exception_for_wrong_argument(self):
520 with pytest.raises(SdlTypeError):
521 self.lock.acquire('bad', self.retry_timeout)
522 with pytest.raises(SdlTypeError):
523 self.lock.acquire(self.retry_interval, 'bad')
525 def test_release_function_success(self):
527 self.mock_db_backend_lock.release.assert_called_once()
529 def test_refresh_function_success(self):
531 self.mock_db_backend_lock.refresh.assert_called_once()
533 def test_get_validity_time_function_success(self):
534 self.mock_db_backend_lock.get_validity_time.return_value = self.expiration
535 ret = self.lock.get_validity_time()
536 self.mock_db_backend_lock.get_validity_time.assert_called_once()
537 assert ret == self.expiration
539 def test_get_validity_time_function_success_when_returned_time_is_float(self):
540 self.mock_db_backend_lock.get_validity_time.return_value = float(self.expiration)
541 ret = self.lock.get_validity_time()
542 self.mock_db_backend_lock.get_validity_time.assert_called_once()
543 assert ret == float(self.expiration)
545 def test_lock_object_string_representation(self):
546 str_out = str(self.lock)
547 assert str_out is not None
550 def test_function_arg_validator():
551 @func_arg_checker(SdlTypeError, 0, a=str, b=(int, float), c=set, d=(dict, type(None)))
552 def _my_func(a='abc', b=1, c={'x', 'y'}, d={'x': b'1'}):
554 with pytest.raises(SdlTypeError, match=r"Wrong argument type: 'a'=<class 'NoneType'>. "
555 r"Must be: <class 'str'>"):
558 with pytest.raises(SdlTypeError, match=r"Wrong argument type: 'b'=<class 'str'>. "):
559 _my_func('abc', 'wrong type')
561 with pytest.raises(SdlTypeError, match=r"Wrong argument type: 'c'=<class 'str'>. "
562 r"Must be: <class 'set'>"):
563 _my_func('abc', 1.0, 'wrong type')
565 with pytest.raises(SdlTypeError, match=r"Wrong argument type: 'd'=<class 'str'>. "):
566 _my_func('abc', 1.0, {'x', 'y'}, 'wrong type')
569 def test_function_kwarg_validator():
570 @func_arg_checker(SdlTypeError, 0, a=str, b=(int, float), c=set, d=(dict, type(None)))
571 def _my_func(a='abc', b=1, c={'x', 'y'}, d={'x': b'1'}):
573 with pytest.raises(SdlTypeError, match=r"Wrong argument type: 'a'=<class 'NoneType'>. "
574 r"Must be: <class 'str'>"):
577 with pytest.raises(SdlTypeError, match=r"Wrong argument type: 'b'=<class 'str'>. "):
578 _my_func(b='wrong type')
580 with pytest.raises(SdlTypeError, match=r"Wrong argument type: 'c'=<class 'str'>. "
581 r"Must be: <class 'set'>"):
582 _my_func(c='wrong type')
584 with pytest.raises(SdlTypeError, match=r"Wrong argument type: 'd'=<class 'str'>. "):
585 _my_func(d='wrong type')