from ricsdl.syncstorage import SyncStorage
from ricsdl.syncstorage import SyncLock
from ricsdl.syncstorage import func_arg_checker
-from ricsdl.exceptions import SdlTypeError
+from ricsdl.exceptions import (SdlTypeError, NotConnected)
@pytest.fixture()
request.cls.ns = 'some-ns'
request.cls.key = 'a'
request.cls.keys = {'a', 'b'}
- request.cls.dm = {'a': b'1', 'b': b'2'}
+ request.cls.dm = {'b': b'2', 'a': b'1'}
request.cls.old_data = b'1'
request.cls.new_data = b'3'
request.cls.keyprefix = 'x'
request.cls.matchedkeys = ['x1', 'x2', 'x3', 'x4', 'x5']
- request.cls.is_atomic = True
request.cls.group = 'some-group'
request.cls.groupmembers = set([b'm1', b'm2'])
request.cls.groupmember = b'm1'
request.cls.lock_name = 'some-lock-name'
request.cls.lock_int_expiration = 10
request.cls.lock_float_expiration = 1.1
+ request.cls.channels = {'abs', 'cbn'}
+ request.cls.channels_and_events = {'abs': 'cbn'}
with patch('ricsdl.backend.get_backend_instance') as mock_db_backend:
storage = SyncStorage()
@pytest.mark.usefixtures('sync_storage_fixture')
class TestSyncStorage:
+ def test_is_active_function_success(self):
+ self.mock_db_backend.is_connected.return_value = True
+ ret = self.storage.is_active()
+ self.mock_db_backend.is_connected.assert_called_once()
+ assert ret is True
+
+ def test_is_active_function_can_catch_backend_exception_and_return_false(self):
+ self.mock_db_backend.is_connected.side_effect = NotConnected
+ ret = self.storage.is_active()
+ self.mock_db_backend.is_connected.assert_called_once()
+ assert ret is False
+
def test_set_function_success(self):
self.storage.set(self.ns, self.dm)
self.mock_db_backend.set.assert_called_once_with(self.ns, self.dm)
self.storage.set(123, {'a': b'v1'})
with pytest.raises(SdlTypeError):
self.storage.set('ns', [1, 2])
+ with pytest.raises(SdlTypeError):
+ self.storage.set('ns', {0xbad: b'v1'})
+ with pytest.raises(SdlTypeError):
+ self.storage.set('ns', {'a': 0xbad})
def test_set_if_function_success(self):
self.mock_db_backend.set_if.return_value = True
assert len(call_args[1]) == len(self.keys)
assert all(k in call_args[1] for k in self.keys)
assert ret == self.dm
+ # Validate that SDL returns a dictionary with keys in alphabetical order
+ assert sorted(self.dm)[0] == list(ret.keys())[0]
def test_get_function_can_return_empty_dict_when_no_key_values_exist(self):
self.mock_db_backend.get.return_value = dict()
def test_find_and_get_function_success(self):
self.mock_db_backend.find_and_get.return_value = self.dm
- ret = self.storage.find_and_get(self.ns, self.keyprefix, self.is_atomic)
- self.mock_db_backend.find_and_get.assert_called_once_with(self.ns, self.keyprefix,
- self.is_atomic)
+ ret = self.storage.find_and_get(self.ns, self.keyprefix)
+ self.mock_db_backend.find_and_get.assert_called_once_with(self.ns, self.keyprefix)
assert ret == self.dm
+ # Validate that SDL returns a dictionary with keys in alphabetical order
+ assert sorted(self.dm)[0] == list(ret.keys())[0]
def test_find_and_get_function_can_return_empty_dict_when_no_keys_exist(self):
self.mock_db_backend.find_and_get.return_value = dict()
- ret = self.storage.find_and_get(self.ns, self.keyprefix, self.is_atomic)
- self.mock_db_backend.find_and_get.assert_called_once_with(self.ns, self.keyprefix,
- self.is_atomic)
+ ret = self.storage.find_and_get(self.ns, self.keyprefix)
+ self.mock_db_backend.find_and_get.assert_called_once_with(self.ns, self.keyprefix)
assert ret == dict()
def test_find_and_get_function_can_raise_exception_for_wrong_argument(self):
with pytest.raises(SdlTypeError):
- self.storage.find_and_get(0xbad, self.keyprefix, self.is_atomic)
- with pytest.raises(SdlTypeError):
- self.storage.find_and_get(self.ns, 0xbad, self.is_atomic)
+ self.storage.find_and_get(0xbad, self.keyprefix)
with pytest.raises(SdlTypeError):
- self.storage.find_and_get(self.ns, self.keyprefix, 0xbad)
+ self.storage.find_and_get(self.ns, 0xbad)
def test_remove_function_success(self):
self.storage.remove(self.ns, self.keys)
def test_remove_all_function_success(self):
self.mock_db_backend.find_keys.return_value = ['a1']
self.storage.remove_all(self.ns)
- self.mock_db_backend.find_keys.assert_called_once_with(self.ns, '')
+ self.mock_db_backend.find_keys.assert_called_once_with(self.ns, '*')
self.mock_db_backend.remove.assert_called_once_with(self.ns,
self.mock_db_backend.find_keys.return_value)
with pytest.raises(SdlTypeError):
self.storage.group_size(self.ns, 0xbad)
+ def test_set_and_publish_function_success(self):
+ self.storage.set_and_publish(self.ns, self.channels_and_events, self.dm)
+ self.mock_db_backend.set_and_publish.assert_called_once_with(self.ns,
+ self.channels_and_events,
+ self.dm)
+
+ def test_set_and_publish_can_raise_exception_for_wrong_argument(self):
+ with pytest.raises(SdlTypeError):
+ self.storage.set_and_publish(123, self.channels_and_events, {'a': b'v1'})
+ with pytest.raises(SdlTypeError):
+ self.storage.set_and_publish('ns', self.channels_and_events, [1, 2])
+ with pytest.raises(SdlTypeError):
+ self.storage.set_and_publish('ns', self.channels_and_events, {0xbad: b'v1'})
+ with pytest.raises(SdlTypeError):
+ self.storage.set_and_publish('ns', self.channels_and_events, {'a': 0xbad})
+
+ def test_set_if_and_publish_success(self):
+ self.mock_db_backend.set_if_and_publish.return_value = True
+ ret = self.storage.set_if_and_publish(self.ns, self.channels_and_events, self.key,
+ self.old_data, self.new_data)
+ self.mock_db_backend.set_if_and_publish.assert_called_once_with(
+ self.ns, self.channels_and_events, self.key, self.old_data, self.new_data)
+ assert ret is True
+
+ def test_set_if_and_publish_can_return_false_if_same_data_already_exists(self):
+ self.mock_db_backend.set_if_and_publish.return_value = False
+ ret = self.storage.set_if_and_publish(self.ns, self.channels_and_events, self.key,
+ self.old_data, self.new_data)
+ self.mock_db_backend.set_if_and_publish.assert_called_once_with(
+ self.ns, self.channels_and_events, self.key, self.old_data, self.new_data)
+ assert ret is False
+
+ def test_set_if_and_publish_can_raise_exception_for_wrong_argument(self):
+ with pytest.raises(SdlTypeError):
+ self.storage.set_if_and_publish(0xbad, self.channels_and_events, 'key', b'v1', b'v2')
+ with pytest.raises(SdlTypeError):
+ self.storage.set_if_and_publish('ns', self.channels_and_events, 0xbad, b'v1', b'v2')
+ with pytest.raises(SdlTypeError):
+ self.storage.set_if_and_publish('ns', self.channels_and_events, 'key', 0xbad, b'v2')
+ with pytest.raises(SdlTypeError):
+ self.storage.set_if_and_publish('ns', self.channels_and_events, 'key', b'v1', 0xbad)
+
+ def test_set_if_not_exists_and_publish_success(self):
+ self.mock_db_backend.set_if_not_exists_and_publish.return_value = True
+ ret = self.storage.set_if_not_exists_and_publish(self.ns, self.channels_and_events,
+ self.key, self.new_data)
+ self.mock_db_backend.set_if_not_exists_and_publish.assert_called_once_with(
+ self.ns, self.channels_and_events, self.key, self.new_data)
+ assert ret is True
+
+ def test_set_if_not_exists_and_publish_function_can_return_false_if_key_already_exists(self):
+ self.mock_db_backend.set_if_not_exists_and_publish.return_value = False
+ ret = self.storage.set_if_not_exists_and_publish(self.ns, self.channels_and_events,
+ self.key, self.new_data)
+ self.mock_db_backend.set_if_not_exists_and_publish.assert_called_once_with(
+ self.ns, self.channels_and_events, self.key, self.new_data)
+ assert ret is False
+
+ def test_set_if_not_exists_and_publish_can_raise_exception_for_wrong_argument(self):
+ with pytest.raises(SdlTypeError):
+ self.storage.set_if_not_exists_and_publish(0xbad, self.channels_and_events, 'key',
+ b'v1')
+ with pytest.raises(SdlTypeError):
+ self.storage.set_if_not_exists_and_publish('ns', self.channels_and_events, 0xbad, b'v1')
+ with pytest.raises(SdlTypeError):
+ self.storage.set_if_not_exists_and_publish('ns', self.channels_and_events, 'key', 0xbad)
+
+ def test_remove_and_publish_function_success(self):
+ self.storage.remove_and_publish(self.ns, self.channels_and_events, self.keys)
+ self.mock_db_backend.remove_and_publish.assert_called_once_with(
+ self.ns, self.channels_and_events, list(self.keys))
+
+ def test_remove_and_publish_can_raise_exception_for_wrong_argument(self):
+ with pytest.raises(SdlTypeError):
+ self.storage.remove_and_publish(0xbad, self.channels_and_events, self.keys)
+ with pytest.raises(SdlTypeError):
+ self.storage.remove(self.ns, self.channels_and_events, 0xbad)
+
+ def test_remove_if_and_publish_success(self):
+ self.mock_db_backend.remove_if_and_publish.return_value = True
+ ret = self.storage.remove_if_and_publish(self.ns, self.channels_and_events, self.key,
+ self.new_data)
+ self.mock_db_backend.remove_if_and_publish.assert_called_once_with(
+ self.ns, self.channels_and_events, self.key, self.new_data)
+ assert ret is True
+
+ def test_remove_if_remove_and_publish_can_return_false_if_data_does_not_match(self):
+ self.mock_db_backend.remove_if_and_publish.return_value = False
+ ret = self.storage.remove_if_and_publish(self.ns, self.channels_and_events, self.key,
+ self.old_data)
+ self.mock_db_backend.remove_if_and_publish.assert_called_once_with(
+ self.ns, self.channels_and_events, self.key, self.old_data)
+ assert ret is False
+
+ def test_remove_if_remove_and_publish_can_raise_exception_for_wrong_argument(self):
+ with pytest.raises(SdlTypeError):
+ self.storage.remove_if_and_publish(0xbad, self.channels_and_events, self.keys,
+ self.old_data)
+ with pytest.raises(SdlTypeError):
+ self.storage.remove_if_and_publish(self.ns, self.channels_and_events, 0xbad,
+ self.old_data)
+ with pytest.raises(SdlTypeError):
+ self.storage.remove_if_and_publish(self.ns, self.channels_and_events, self.keys, 0xbad)
+
+ def test_remove_all_and_publish_success(self):
+ self.storage.remove_all_and_publish(self.ns, self.channels_and_events)
+ self.mock_db_backend.remove_all_and_publish.assert_called_once_with(
+ self.ns, self.channels_and_events)
+
+ def test_remove_all_and_publish_can_raise_exception_for_wrong_argument(self):
+ with pytest.raises(SdlTypeError):
+ self.storage.remove_all_and_publish(0xbad, self.channels_and_events)
+
+ def test_subscribe_function_success(self):
+ def cb(channel, message):
+ pass
+ self.storage.subscribe_channel(self.ns, cb, self.channels)
+ self.mock_db_backend.subscribe_channel.assert_called_once_with(
+ self.ns, cb, list(self.channels))
+
+ def test_subscribe_can_raise_exception_for_wrong_argument(self):
+ def cb3(channel, message, extra):
+ pass
+ def cb1(channel):
+ pass
+ with pytest.raises(SdlTypeError):
+ self.storage.subscribe_channel(self.ns, cb3, self.channels)
+ with pytest.raises(SdlTypeError):
+ self.storage.subscribe_channel(self.ns, cb1, self.channels)
+
+ def test_unsubscribe_function_success(self):
+ self.storage.unsubscribe_channel(self.ns, self.channels)
+ self.mock_db_backend.unsubscribe_channel.assert_called_once_with(
+ self.ns, list(self.channels))
+
+ def test_start_event_listener_success(self):
+ self.storage.start_event_listener()
+ self.mock_db_backend.start_event_listener.assert_called()
+
+ def test_handle_events_success(self):
+ self.storage.handle_events()
+ self.mock_db_backend.handle_events.assert_called()
+
@patch('ricsdl.syncstorage.SyncLock')
def test_get_lock_resource_function_success_when_expiration_time_is_integer(self, mock_db_lock):
ret = self.storage.get_lock_resource(self.ns, self.lock_name, self.lock_int_expiration)
Code Review / ric-plt / sdlpy.git / blobdiff