value = msgpack.packb(value, use_bin_type=True)
self._sdl.set(ns, {key: value})
+ def set_if(self, ns, key, old_value, new_value, usemsgpack=True):
+ """
+ Conditionally modify the value of a key if the current value in data storage matches the
+ user's last known value.
+
+ Parameters
+ ----------
+ ns: string
+ SDL namespace
+ key: string
+ SDL key
+ old_value:
+ Lask known object or byte array. See the `usemsgpack` parameter.
+ new_value:
+ Object or byte array to be written. See the `usemsgpack` parameter.
+ usemsgpack: boolean (optional, default is True)
+ Determines whether the value is serialized using msgpack before storing.
+ If usemsgpack is True, the msgpack function `packb` is invoked
+ on the value to yield a byte array that is then sent to SDL.
+ Stated differently, if usemsgpack is True, the value can be anything
+ that is serializable by msgpack.
+ If usemsgpack is False, the value must be bytes.
+
+ Returns
+ -------
+ bool
+ True for successful modification, false if the user's last known data did not
+ match the current value in data storage.
+ """
+ if usemsgpack:
+ old_value = msgpack.packb(old_value, use_bin_type=True)
+ new_value = msgpack.packb(new_value, use_bin_type=True)
+ return self._sdl.set_if(ns, key, old_value, new_value)
+
+ def set_if_not_exists(self, ns, key, value, usemsgpack=True):
+ """
+ Write data to SDL storage if key does not exist.
+
+ Parameters
+ ----------
+ ns: string
+ SDL namespace
+ key: string
+ SDL key
+ value:
+ Object or byte array to store. See the `usemsgpack` parameter.
+ usemsgpack: boolean (optional, default is True)
+ Determines whether the value is serialized using msgpack before storing.
+ If usemsgpack is True, the msgpack function `packb` is invoked
+ on the value to yield a byte array that is then sent to SDL.
+ Stated differently, if usemsgpack is True, the value can be anything
+ that is serializable by msgpack.
+ If usemsgpack is False, the value must be bytes.
+
+ Returns
+ -------
+ bool
+ True for successful modification, false if the user's last known data did not
+ match the current value in data storage.
+ """
+ if usemsgpack:
+ value = msgpack.packb(value, use_bin_type=True)
+ return self._sdl.set_if_not_exists(ns, key, value)
+
def get(self, ns, key, usemsgpack=True):
"""
Gets the value for the specified namespace and key,
result = msgpack.unpackb(result, raw=False)
return result
+ def find_keys(self, ns, prefix):
+ """
+ Find all keys matching search pattern under the namespace.
+
+ Parameters
+ ----------
+ ns: string
+ SDL namespace
+ prefix: string
+ Key search pattern
+
+ Returns
+ -------
+ keys: list
+ A list of found keys.
+ """
+ return self._sdl.find_keys(ns, f"{prefix}*")
+
def find_and_get(self, ns, prefix, usemsgpack=True):
"""
Gets all key-value pairs in the specified namespace
"""
# note: SDL "*" usage is inconsistent with real python regex, where it would be ".*"
- ret_dict = self._sdl.find_and_get(ns, "{0}*".format(prefix))
+ ret_dict = self._sdl.find_and_get(ns, f"{prefix}*")
if usemsgpack:
ret_dict = {k: msgpack.unpackb(v, raw=False) for k, v in ret_dict.items()}
return ret_dict
"""
self._sdl.remove(ns, {key})
+ def delete_if(self, ns, key, value, usemsgpack=True):
+ """
+ Conditionally remove data from SDL storage if the current data value matches the user's
+ last known value.
+
+ Parameters
+ ----------
+ ns: string
+ SDL namespace
+ key: string
+ SDL key
+ value:
+ Object or byte array to store. See the `usemsgpack` parameter.
+ usemsgpack: boolean (optional, default is True)
+ Determines whether the value is serialized using msgpack before storing.
+ If usemsgpack is True, the msgpack function `packb` is invoked
+ on the value to yield a byte array that is then sent to SDL.
+ Stated differently, if usemsgpack is True, the value can be anything
+ that is serializable by msgpack.
+ If usemsgpack is False, the value must be bytes.
+
+ Returns
+ -------
+ bool
+ True if successful removal, false if the user's last known data did not match the
+ current value in data storage.
+ """
+ if usemsgpack:
+ value = msgpack.packb(value, use_bin_type=True)
+ return self._sdl.remove_if(ns, key, value)
+
+ def add_member(self, ns, group, member, usemsgpack=True):
+ """
+ Add new members to a SDL group under the namespace.
+
+ Parameters
+ ----------
+ ns: string
+ SDL namespace
+ group: string
+ group name
+ member:
+ member to be added
+ usemsgpack: boolean (optional, default is True)
+ Determines whether the member is serialized using msgpack before storing.
+ If usemsgpack is True, the msgpack function `packb` is invoked
+ on the member to yield a byte array that is then sent to SDL.
+ Stated differently, if usemsgpack is True, the member can be anything
+ that is serializable by msgpack.
+ If usemsgpack is False, the member must be bytes.
+ """
+ if usemsgpack:
+ member = msgpack.packb(member, use_bin_type=True)
+ self._sdl.add_member(ns, group, {member})
+
+ def remove_member(self, ns, group, member, usemsgpack=True):
+ """
+ Remove members from a SDL group.
+
+ Parameters
+ ----------
+ ns: string
+ SDL namespace
+ group: string
+ group name
+ member:
+ member to be removed
+ usemsgpack: boolean (optional, default is True)
+ Determines whether the member is serialized using msgpack before storing.
+ If usemsgpack is True, the msgpack function `packb` is invoked
+ on the member to yield a byte array that is then sent to SDL.
+ Stated differently, if usemsgpack is True, the member can be anything
+ that is serializable by msgpack.
+ If usemsgpack is False, the member must be bytes.
+ """
+ if usemsgpack:
+ member = msgpack.packb(member, use_bin_type=True)
+ self._sdl.remove_member(ns, group, {member})
+
+ def remove_group(self, ns, group):
+ """
+ Remove a SDL group along with its members.
+
+ Parameters
+ ----------
+ ns: string
+ SDL namespace
+ group: string
+ group name to remove
+ usemsgpack: boolean (optional, default is True)
+ Determines whether the member is serialized using msgpack before storing.
+ If usemsgpack is True, the msgpack function `packb` is invoked
+ on the member to yield a byte array that is then sent to SDL.
+ Stated differently, if usemsgpack is True, the member can be anything
+ that is serializable by msgpack.
+ If usemsgpack is False, the member must be bytes.
+ """
+ self._sdl.remove_group(ns, group)
+
+ def get_members(self, ns, group, usemsgpack=True):
+ """
+ Get all the members of a SDL group.
+
+ Parameters
+ ----------
+ ns: string
+ SDL namespace
+ group: string
+ group name to retrive
+ usemsgpack: boolean (optional, default is True)
+ Determines whether the member is serialized using msgpack before storing.
+ If usemsgpack is True, the msgpack function `packb` is invoked
+ on the member to yield a byte array that is then sent to SDL.
+ Stated differently, if usemsgpack is True, the member can be anything
+ that is serializable by msgpack.
+ If usemsgpack is False, the member must be bytes.
+
+ Returns
+ -------
+ Set[str] or Set[bytes]
+ A set of the members of the group.
+ None
+ """
+ ret_set = self._sdl.get_members(ns, group)
+ if usemsgpack:
+ ret_set = {msgpack.unpackb(m, raw=False) for m in ret_set}
+ return ret_set
+
+ def is_member(self, ns, group, member, usemsgpack=True):
+ """
+ Validate if a given member is in the SDL group.
+
+ Parameters
+ ----------
+ ns: string
+ SDL namespace
+ group: string
+ group name
+ member:
+ member to validate
+ usemsgpack: boolean (optional, default is True)
+ Determines whether the member is serialized using msgpack before storing.
+ If usemsgpack is True, the msgpack function `packb` is invoked
+ on the member to yield a byte array that is then sent to SDL.
+ Stated differently, if usemsgpack is True, the member can be anything
+ that is serializable by msgpack.
+ If usemsgpack is False, the member must be bytes.
+
+ Returns
+ -------
+ bool
+ True if member was in the group, false otherwise.
+ """
+ if usemsgpack:
+ member = msgpack.packb(member, use_bin_type=True)
+ return self._sdl.is_member(ns, group, member)
+
+ def group_size(self, ns, group):
+ """
+ Return the number of members in a group.
+ If the group does not exist, value 0 is returned.
+
+ Parameters
+ ----------
+ ns: string
+ SDL namespace
+ group: string
+ group name to retrive size
+ usemsgpack: boolean (optional, default is True)
+ Determines whether the member is serialized using msgpack before storing.
+ If usemsgpack is True, the msgpack function `packb` is invoked
+ on the member to yield a byte array that is then sent to SDL.
+ Stated differently, if usemsgpack is True, the member can be anything
+ that is serializable by msgpack.
+ If usemsgpack is False, the member must be bytes.
+
+ Returns
+ -------
+ int
+ Number of members in a group.
+ """
+ return self._sdl.group_size(ns, group)
+
+ def set_and_publish(self, ns, channel, event, key, value, usemsgpack=True):
+ """
+ Publish event to channel after writing data.
+
+ Parameters
+ ----------
+ ns: string
+ SDL namespace
+ channel: string
+ channel to publish event
+ event: string
+ published message
+ key: string
+ SDL key
+ value:
+ Object or byte array to store. See the `usemsgpack` parameter.
+ usemsgpack: boolean (optional, default is True)
+ Determines whether the value is serialized using msgpack before storing.
+ If usemsgpack is True, the msgpack function `packb` is invoked
+ on the value to yield a byte array that is then sent to SDL.
+ Stated differently, if usemsgpack is True, the value can be anything
+ that is serializable by msgpack.
+ If usemsgpack is False, the value must be bytes.
+ """
+ if usemsgpack:
+ value = msgpack.packb(value, use_bin_type=True)
+ self._sdl.set_and_publish(ns, {channel: event}, {key: value})
+
+ def set_if_and_publish(self, ns, channel, event, key, old_value, new_value, usemsgpack=True):
+ """
+ Publish event to channel after conditionally modifying the value of a key if the
+ current value in data storage matches the user's last known value.
+
+ Parameters
+ ----------
+ ns: string
+ SDL namespace
+ channel: string
+ channel to publish event
+ event: string
+ published message
+ key: string
+ SDL key
+ old_value:
+ Lask known object or byte array. See the `usemsgpack` parameter.
+ new_value:
+ Object or byte array to be written. See the `usemsgpack` parameter.
+ usemsgpack: boolean (optional, default is True)
+ Determines whether the old_value & new_value is serialized using msgpack before storing.
+ If usemsgpack is True, the msgpack function `packb` is invoked
+ on the old_value & new_value to yield a byte array that is then sent to SDL.
+ Stated differently, if usemsgpack is True, the old_value & new_value can be anything
+ that is serializable by msgpack.
+ If usemsgpack is False, the old_value & new_value must be bytes.
+
+ Returns
+ -------
+ bool
+ True for successful modification, false if the user's last known data did not
+ match the current value in data storage.
+ """
+ if usemsgpack:
+ old_value = msgpack.packb(old_value, use_bin_type=True)
+ new_value = msgpack.packb(new_value, use_bin_type=True)
+ return self._sdl.set_if_and_publish(ns, {channel: event}, key, old_value, new_value)
+
+ def set_if_not_exists_and_publish(self, ns, channel, event, key, value, usemsgpack=True):
+ """
+ Publish event to channel after writing data to SDL storage if key does not exist.
+
+ Parameters
+ ----------
+ ns: string
+ SDL namespace
+ channel: string
+ channel to publish event
+ event: string
+ published message
+ key: string
+ SDL key
+ value:
+ Object or byte array to store. See the `usemsgpack` parameter.
+ usemsgpack: boolean (optional, default is True)
+ Determines whether the value is serialized using msgpack before storing.
+ If usemsgpack is True, the msgpack function `packb` is invoked
+ on the value to yield a byte array that is then sent to SDL.
+ Stated differently, if usemsgpack is True, the value can be anything
+ that is serializable by msgpack.
+ If usemsgpack is False, the value must be bytes.
+
+ Returns
+ -------
+ bool
+ True if key didn't exist yet and set operation was executed, false if key already
+ existed and thus its value was left untouched.
+ """
+ if usemsgpack:
+ value = msgpack.packb(value, use_bin_type=True)
+ return self._sdl.set_if_not_exists_and_publish(ns, {channel: event}, key, value)
+
+ def remove_and_publish(self, ns, channel, event, key):
+ """
+ Publish event to channel after removing data.
+
+ Parameters
+ ----------
+ ns: string
+ SDL namespace
+ channel: string
+ channel to publish event
+ event: string
+ published message
+ key: string
+ SDL key
+ """
+ self._sdl.remove_and_publish(ns, {channel: event}, {key})
+
+ def remove_if_and_publish(self, ns, channel, event, key, value, usemsgpack=True):
+ """
+ Publish event to channel after removing key and its data from database if the
+ current data value is expected one.
+
+ Parameters
+ ----------
+ ns: string
+ SDL namespace
+ channel: string
+ channel to publish event
+ event: string
+ published message
+ key: string
+ SDL key
+ value:
+ Object or byte array to store. See the `usemsgpack` parameter.
+ usemsgpack: boolean (optional, default is True)
+ Determines whether the value is serialized using msgpack before storing.
+ If usemsgpack is True, the msgpack function `packb` is invoked
+ on the value to yield a byte array that is then sent to SDL.
+ Stated differently, if usemsgpack is True, the value can be anything
+ that is serializable by msgpack.
+ If usemsgpack is False, the value must be bytes.
+
+ Returns
+ -------
+ bool
+ True if successful removal, false if the user's last known data did not match the
+ current value in data storage.
+ """
+ if usemsgpack:
+ value = msgpack.packb(value, use_bin_type=True)
+ return self._sdl.remove_if_and_publish(ns, {channel: event}, key, value)
+
+ def remove_all_and_publish(self, ns, channel, event):
+ """
+ Publish event to channel after removing all keys under the namespace.
+
+ Parameters
+ ----------
+ ns: string
+ SDL namespace
+ channel: string
+ channel to publish event
+ event: string
+ published message
+ """
+ self._sdl.remove_all_and_publish(ns, {channel: event})
+
+ def subscribe_channel(self, ns, cb, channel):
+ """
+ Subscribes the client to the specified channels.
+
+ Parameters
+ ----------
+ ns: string
+ SDL namespace
+ cb:
+ A function that is called when an event on channel is received.
+ channel: string
+ channel to subscribe
+ """
+ self._sdl.subscribe_channel(ns, cb, {channel})
+
+ def unsubscribe_channel(self, ns, channel):
+ """
+ unsubscribe_channel removes subscription from one or several channels.
+
+ Parameters
+ ----------
+ ns: string
+ SDL namespace
+ channel: string
+ channel to unsubscribe
+ """
+ self._sdl.unsubscribe_channel(ns, {channel})
+
+ def start_event_listener(self):
+ """
+ start_event_listener creates an event loop in a separate thread for handling
+ events from subscriptions. The registered callback function will be called
+ when an event is received.
+ """
+ self._sdl.start_event_listener()
+
+ def handle_events(self):
+ """
+ handle_events is a non-blocking function that returns a tuple containing channel
+ name and message received from an event. The registered callback function will
+ still be called when an event is received.
+
+ This function is called if SDL user decides to handle notifications in its own
+ event loop. Calling this function after start_event_listener raises an exception.
+ If there are no notifications, these returns None.
+
+ Returns
+ -------
+ Tuple:
+ (channel: str, message: str)
+ """
+ return self._sdl.handle_events()
+
def healthcheck(self):
"""
Checks if the sdl connection is healthy.
"""
tests data functions
"""
-
+import time
from ricxappframe.xapp_sdl import SDLWrapper
assert sdl.find_and_get(NS, "as.df") == {}
assert sdl.find_and_get(NS, "") == {}
+
+
+def test_sdl_set_get():
+ """
+ test set, get realted sdl methods
+ """
+ sdl = SDLWrapper(use_fake_sdl=True)
+
+ # set_if
+ sdl.set(NS, "gs.df1", "old")
+ assert sdl.get(NS, "gs.df1") == "old"
+
+ sdl.set_if(NS, "gs.df1", "young", "new")
+ assert sdl.get(NS, "gs.df1") == "old"
+
+ sdl.set_if(NS, "gs.df1", "old", "new")
+ assert sdl.get(NS, "gs.df1") == "new"
+
+ # set_if_not_exists
+ sdl.set(NS, "gs.df2", "old")
+ assert sdl.get(NS, "gs.df2") == "old"
+
+ sdl.set_if_not_exists(NS, "gs.df2", "new")
+ assert sdl.get(NS, "gs.df2") == "old"
+
+ sdl.set_if_not_exists(NS, "gs.df3", "new")
+ assert sdl.get(NS, "gs.df3") == "new"
+
+ # find_keys
+ assert sdl.find_keys(NS, "gs") == ["gs.df1", "gs.df2", "gs.df3"]
+ assert sdl.find_keys(NS, "gs.df1") == ["gs.df1"]
+ assert sdl.find_keys(NS, "gs.df2") == ["gs.df2"]
+ assert sdl.find_keys(NS, "gs.df3") == ["gs.df3"]
+
+ # delete_if
+ sdl.set(NS, "gs.df4", "delete_this")
+
+ assert sdl.delete_if(NS, "gs.df4", "delete") is False
+ assert sdl.delete_if(NS, "gs.df4", "delete_this") is True
+ assert sdl.get(NS, "gs.df4") is None
+
+
+def test_sdl_member():
+ """
+ test member related sdl methods
+ """
+ # add_member, remove_member, get_members
+ sdl = SDLWrapper(use_fake_sdl=True)
+
+ sdl.add_member(NS, "group1", "member1")
+ assert sdl.is_member(NS, "group1", "member1") is True
+
+ sdl.remove_member(NS, "group1", "not_member")
+ assert sdl.is_member(NS, "group1", "member1") is True
+
+ sdl.remove_member(NS, "group1", "member1")
+ assert sdl.is_member(NS, "group1", "member1") is False
+
+ # remove_group, group_size
+ sdl.add_member(NS, "group2", "member1")
+ sdl.add_member(NS, "group2", "member2")
+ assert sdl.group_size(NS, "group2") == 2
+ sdl.remove_group(NS, "group2")
+ assert sdl.group_size(NS, "group2") == 0
+
+ # get_members
+ sdl.add_member(NS, "group3", "member1")
+ sdl.add_member(NS, "group3", "member2")
+ members = sdl.get_members(NS, "group3")
+ assert "member1" in members
+ assert "member2" in members
+
+
+def test_sdl_set_and_publish_with_handle_events():
+ """
+ test set_and_publish* related sdl methods
+ """
+ CH = "channel"
+ EVENT = "event"
+ CALLED = None
+
+ def cb(channel, event):
+ nonlocal CH
+ nonlocal EVENT
+ nonlocal CALLED
+ # test is cb called
+ CALLED = True
+ assert channel == CH
+ assert event == EVENT
+
+ sdl = SDLWrapper(use_fake_sdl=True)
+ sdl.subscribe_channel(NS, cb, "channel")
+
+ # set_and_publish
+ CALLED = False
+ sdl.set_and_publish(NS, "channel", "event", "nt.df1", "old")
+ sdl.handle_events()
+ assert sdl.get(NS, "nt.df1") == "old"
+ assert CALLED is True
+
+ # set_if_and_publish fail
+ CALLED = False
+ sdl.set_if_and_publish(NS, "channel", "event", "nt.df1", "young", "new")
+ sdl.handle_events()
+ assert sdl.get(NS, "nt.df1") == "old"
+ assert CALLED is False
+ # set_if_and_publish success
+ sdl.set_if_and_publish(NS, "channel", "event", "nt.df1", "old", "new")
+ sdl.handle_events()
+ assert sdl.get(NS, "nt.df1") == "new"
+ assert CALLED is True
+
+ # set_if_not_exists_and_publish fail
+ CALLED = False
+ sdl.set_if_not_exists_and_publish(NS, "channel", "event", "nt.df1", "latest")
+ sdl.handle_events()
+ assert sdl.get(NS, "nt.df1") == "new"
+ assert CALLED is False
+ # set_if_not_exists_and_publish success
+ sdl.set_if_not_exists_and_publish(
+ NS, "channel", "event", "nt.df2", "latest")
+ sdl.handle_events()
+ assert sdl.get(NS, "nt.df2") == "latest"
+ assert CALLED is True
+
+ sdl.unsubscribe_channel(NS, "channel")
+
+
+def test_sdl_remove_and_publish_with_start_event_listener():
+ """
+ test remove_and_publish* related sdl methods
+ """
+ CH = "channel"
+ EVENT = "event"
+ CALLED = None
+
+ def cb(channel, event):
+ nonlocal CH
+ nonlocal EVENT
+ nonlocal CALLED
+ CALLED = True
+ assert channel == CH
+ assert event == EVENT
+
+ sdl = SDLWrapper(use_fake_sdl=True)
+ sdl.subscribe_channel(NS, cb, "channel")
+ sdl.start_event_listener()
+
+ # remove_and_publish success
+ CALLED = False
+ sdl.set(NS, "nt.df1", "old")
+ sdl.remove_and_publish(NS, "channel", "event", "nt.df1")
+ time.sleep(0.3)
+ assert sdl.get(NS, "nt.df1") is None
+ assert CALLED is True
+
+ # remove_if_and_publish
+ CALLED = False
+ sdl.set(NS, "nt.df1", "old")
+ # fail
+ sdl.remove_if_and_publish(NS, "channel", "event", "nt.df1", "new")
+ time.sleep(0.3)
+ assert sdl.get(NS, "nt.df1") == "old"
+ assert CALLED is False
+ # success
+ sdl.remove_if_and_publish(NS, "channel", "event", "nt.df1", "old")
+ time.sleep(0.3)
+ assert sdl.get(NS, "nt.df1") is None
+ assert CALLED is True
+
+ # remove_all_and_publish
+ CALLED = False
+ sdl.set(NS, "nt.df1", "data1")
+ sdl.set(NS, "nt.df2", "data2")
+ sdl.set(NS, "nt.df3", "data3")
+ sdl.remove_all_and_publish(NS, "channel", "event")
+ time.sleep(0.3)
+ assert sdl.get(NS, "nt.df1") is None
+ assert sdl.get(NS, "nt.df2") is None
+ assert sdl.get(NS, "nt.df3") is None
+ assert sdl.find_keys(NS, "*") == []
+ assert CALLED is True
+
+ sdl.unsubscribe_channel(NS, "channel")
Code Review / ric-plt / xapp-frame-py.git / commitdiff