#!/usr/bin/env python # Copyright (C) 2007, Red Hat, Inc. # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation; either version 2 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program; if not, write to the Free Software # Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA import gobject import dbus, dbus.service, dbus.glib _PRESENCE_SERVICE = "org.laptop.Sugar.Presence" _PRESENCE_INTERFACE = "org.laptop.Sugar.Presence" _PRESENCE_TEST_INTERFACE = "org.laptop.Sugar.Presence._Test" _PRESENCE_PATH = "/org/laptop/Sugar/Presence" class NotFoundError(dbus.DBusException): def __init__(self, msg=None): dbus.DBusException.__init__(self, msg) self._dbus_error_name = _PRESENCE_INTERFACE + '.NotFound' _ACTIVITY_PATH = "/org/laptop/Sugar/Presence/Activities/" _ACTIVITY_INTERFACE = "org.laptop.Sugar.Presence.Activity" class TestActivity(dbus.service.Object): def __init__(self, bus_name, object_id, parent, actid, name, color, atype, properties): self._parent = parent self._actid = actid self._aname = name self._color = color self._type = atype self._properties = {} for (key, value) in properties.items(): self._properties[str(key)] = str(value) self._buddies = {} self._object_id = object_id self._object_path = _ACTIVITY_PATH + str(self._object_id) dbus.service.Object.__init__(self, bus_name, self._object_path) def add_buddy(self, buddy): if self._buddies.has_key(buddy._key): raise NotFoundError("Buddy already in activity") self._buddies[buddy._key] = buddy self.BuddyJoined(buddy._object_path) def remove_buddy(self, buddy): if not self._buddies.has_key(buddy._key): raise NotFoundError("Buddy not in activity") self.BuddyLeft(buddy._object_path) del self._buddies[buddy._key] def disappear(self): # remove all buddies from activity for buddy in self.get_buddies(): self.BuddyLeft(buddy._object_path) self._buddies = {} def get_buddies(self): return self._buddies.values() @dbus.service.signal(_ACTIVITY_INTERFACE, signature="o") def BuddyJoined(self, buddy_path): pass @dbus.service.signal(_ACTIVITY_INTERFACE, signature="o") def BuddyLeft(self, buddy_path): pass @dbus.service.signal(_ACTIVITY_INTERFACE, signature="o") def NewChannel(self, channel_path): pass @dbus.service.method(_ACTIVITY_INTERFACE, out_signature="s") def GetId(self): return self._actid @dbus.service.method(_ACTIVITY_INTERFACE, out_signature="s") def GetName(self): return self._aname @dbus.service.method(_ACTIVITY_INTERFACE, out_signature="s") def GetColor(self): return self._color @dbus.service.method(_ACTIVITY_INTERFACE, out_signature="s") def GetType(self): return self._type @dbus.service.method(_ACTIVITY_INTERFACE) def Join(self): owner = self._parent._owner self.add_buddy(owner) owner.add_activity(self) @dbus.service.method(_ACTIVITY_INTERFACE, out_signature="ao") def GetJoinedBuddies(self): ret = [] for buddy in self._buddies.values(): ret.append(dbus.ObjectPath(buddy._object_path)) return ret @dbus.service.method(_ACTIVITY_INTERFACE, out_signature="soao") def GetChannels(self): return None _BUDDY_PATH = "/org/laptop/Sugar/Presence/Buddies/" _BUDDY_INTERFACE = "org.laptop.Sugar.Presence.Buddy" _OWNER_INTERFACE = "org.laptop.Sugar.Presence.Buddy.Owner" _PROP_NICK = "nick" _PROP_KEY = "key" _PROP_ICON = "icon" _PROP_CURACT = "current-activity" _PROP_COLOR = "color" _PROP_OWNER = "owner" class TestBuddy(dbus.service.Object): def __init__(self, bus_name, object_id, pubkey, nick, color): self._key = pubkey self._nick = nick self._color = color self._owner = False self._curact = None self._icon = "" self._activities = {} self._object_id = object_id self._object_path = _BUDDY_PATH + str(self._object_id) dbus.service.Object.__init__(self, bus_name, self._object_path) def add_activity(self, activity): if self._activities.has_key(activity._actid): raise NotFoundError("Buddy already in activity") self._activities[activity._actid] = activity self.JoinedActivity(activity._object_path) def remove_activity(self, activity): if not self._activities.has_key(activity._actid): raise NotFoundError("Buddy not in activity") self.LeftActivity(activity._object_path) del self._activities[activity._actid] def leave_activities(self): for activity in self.get_activities(): self.LeftActivity(activity._object_path) self._activities = {} def get_activities(self): return self._activities.values() def set_current_activity(self, actid): self._curact = actid self.PropertyChanged({_PROP_CURACT: actid}) @dbus.service.signal(_BUDDY_INTERFACE, signature="ay") def IconChanged(self, icon_data): pass @dbus.service.signal(_BUDDY_INTERFACE, signature="o") def JoinedActivity(self, activity_path): pass @dbus.service.signal(_BUDDY_INTERFACE, signature="o") def LeftActivity(self, activity_path): pass @dbus.service.signal(_BUDDY_INTERFACE, signature="a{sv}") def PropertyChanged(self, updated): pass # dbus methods @dbus.service.method(_BUDDY_INTERFACE, in_signature="", out_signature="ay") def GetIcon(self): return dbus.ByteArray(self._icon) @dbus.service.method(_BUDDY_INTERFACE, in_signature="", out_signature="ao") def GetJoinedActivities(self): acts = [] for activity in self._activities.values(): acts.append(dbus.ObjectPath(activity._object_path)) return acts @dbus.service.method(_BUDDY_INTERFACE, in_signature="", out_signature="a{sv}") def GetProperties(self): props = {} props[_PROP_NICK] = self._nick props[_PROP_OWNER] = self._owner props[_PROP_KEY] = self._key props[_PROP_COLOR] = self._color if self._curact: props[_PROP_CURACT] = self._curact else: props[_PROP_CURACT] = "" return props _OWNER_PUBKEY = "AAAAB3NzaC1kc3MAAACBAKEVDFJW9D9GK20QFYRKbhV7kpjnhKkkzudn34ij" \ "Ixje+x1ZXTIU6J1GFmJYrHq9uBRi72lOVAosGUop+HHZFRyTeYLxItmKfIoD" \ "S2rwyL9cGRoDsD4yjECMqa2I+pGxriw4OmHeu5vmBkk+5bXBdkLf0EfseuPC" \ "lT7FE+Fj4C6FAAAAFQCygOIpXXybKlVTcEfprOQp3Uud0QAAAIBjyjQhOWHq" \ "FdJlALmnriQR+Zi1i4N/UMjWihF245RXJuUU6DyYbx4QxznxRnYKx/ZvsD0O" \ "9+ihzmQd6eFwU/jQ6sxiL7DSlCJ3axgG9Yvbf7ELeXGo4/Z9keOVdei0sXz4" \ "VBvJC0c0laELsnU0spFC62qQKxNemTbXDGksauj19gAAAIEAmcvY8VX47pRP" \ "k7MjrDzZlPvvNQgHMNZSwHGIsF7EMGVDCYpbQTyR+cmtJBBFVyxtNbK7TWTZ" \ "K8uH1tm9GyMcViUdIT4xCirA0JanE597KdlBz39l/623wF4jvbnnHOZ/pIT9" \ "tPd1pCYJf+L7OEKCBUAyQhcq159X8A1toM48Soc=" _OWNER_PRIVKEY = "MIIBuwIBAAKBgQChFQxSVvQ/RittEBWESm4Ve5KY54SpJM7nZ9+IoyMY3vs" \ "dWV0yFOidRhZiWKx6vbgUYu9pTlQKLBlKKfhx2RUck3mC8SLZinyKA0tq8M" \ "i/XBkaA7A+MoxAjKmtiPqRsa4sODph3rub5gZJPuW1wXZC39BH7HrjwpU+x" \ "RPhY+AuhQIVALKA4ildfJsqVVNwR+ms5CndS53RAoGAY8o0ITlh6hXSZQC5" \ "p64kEfmYtYuDf1DI1ooRduOUVyblFOg8mG8eEMc58UZ2Csf2b7A9Dvfooc5" \ "kHenhcFP40OrMYi+w0pQid2sYBvWL23+xC3lxqOP2fZHjlXXotLF8+FQbyQ" \ "tHNJWhC7J1NLKRQutqkCsTXpk21wxpLGro9fYCgYEAmcvY8VX47pRPk7Mjr" \ "DzZlPvvNQgHMNZSwHGIsF7EMGVDCYpbQTyR+cmtJBBFVyxtNbK7TWTZK8uH" \ "1tm9GyMcViUdIT4xCirA0JanE597KdlBz39l/623wF4jvbnnHOZ/pIT9tPd" \ "1pCYJf+L7OEKCBUAyQhcq159X8A1toM48SocCFAvkZYCYtLhSDEPrlf0jLD" \ "jrMz+i" _OWNER_NICK = "TestOwner" _OWNER_COLOR = "#75C228,#308C30" class TestOwner(TestBuddy): def __init__(self, bus_name, object_id): TestBuddy.__init__(self, bus_name, object_id, _OWNER_PUBKEY, _OWNER_NICK, _OWNER_COLOR) self._owner = True class TestPresenceService(dbus.service.Object): """A test D-Bus PresenceService used to exercise the Sugar PS bindings.""" def __init__(self): self._next_object_id = 0 self._activities = {} self._buddies = {} self._bus_name = dbus.service.BusName(_PRESENCE_SERVICE, bus=dbus.SessionBus()) objid = self._get_next_object_id() self._owner = TestOwner(self._bus_name, objid) dbus.service.Object.__init__(self, self._bus_name, _PRESENCE_PATH) def _get_next_object_id(self): """Increment and return the object ID counter.""" self._next_object_id = self._next_object_id + 1 return self._next_object_id @dbus.service.signal(_PRESENCE_INTERFACE, signature="o") def ActivityAppeared(self, activity): pass @dbus.service.signal(_PRESENCE_INTERFACE, signature="o") def ActivityDisappeared(self, activity): pass @dbus.service.signal(_PRESENCE_INTERFACE, signature="o") def BuddyAppeared(self, buddy): pass @dbus.service.signal(_PRESENCE_INTERFACE, signature="o") def BuddyDisappeared(self, buddy): pass @dbus.service.signal(_PRESENCE_INTERFACE, signature="o") def ActivityInvitation(self, activity): pass @dbus.service.signal(_PRESENCE_INTERFACE, signature="soo") def PrivateInvitation(self, bus_name, connection, channel): pass @dbus.service.method(_PRESENCE_INTERFACE, out_signature="ao") def GetActivities(self): ret = [] for act in self._activities.values(): ret.append(dbus.ObjectPath(act._object_path)) return ret @dbus.service.method(_PRESENCE_INTERFACE, in_signature="s", out_signature="o") def GetActivityById(self, actid): if self._activities.has_key(actid): return dbus.ObjectPath(self._activities[actid]._object_path) raise NotFoundError("The activity was not found.") @dbus.service.method(_PRESENCE_INTERFACE, out_signature="ao") def GetBuddies(self): ret = [] for buddy in self._buddies.values(): ret.append(buddy._object_path) return ret @dbus.service.method(_PRESENCE_INTERFACE, in_signature="ay", out_signature="o") def GetBuddyByPublicKey(self, key): key = ''.join([chr(item) for item in key]) if self._buddies.has_key(key): return self._buddies[key]._object_path raise NotFoundError("The buddy was not found.") @dbus.service.method(_PRESENCE_INTERFACE, out_signature="o") def GetOwner(self): if not self._owner: raise NotFoundError("The owner was not found.") return dbus.ObjectPath(self._owner._object_path) def _internal_share_activity(self, actid, atype, name, properties, color=None): objid = self._get_next_object_id() if not color: color = self._owner._color act = TestActivity(self._bus_name, objid, self, actid, name, color, atype, properties) self._activities[actid] = act self.ActivityAppeared(act._object_path) return act @dbus.service.method(_PRESENCE_INTERFACE, in_signature="sssa{sv}", out_signature="o") def ShareActivity(self, actid, atype, name, properties): act = self._internal_share_activity(actid, atype, name, properties) act.add_buddy(self._owner) self._owner.add_activity(act) return act._object_path @dbus.service.method(_PRESENCE_INTERFACE, out_signature="so") def GetPreferredConnection(self): return "bar.baz.foo", "/bar/baz/foo" # Private methods used for testing @dbus.service.method(_PRESENCE_TEST_INTERFACE, in_signature="ayss") def AddBuddy(self, pubkey, nick, color): pubkey = ''.join([chr(item) for item in pubkey]) objid = self._get_next_object_id() buddy = TestBuddy(self._bus_name, objid, pubkey, nick, color) self._buddies[pubkey] = buddy self.BuddyAppeared(buddy._object_path) @dbus.service.method(_PRESENCE_TEST_INTERFACE, in_signature="ay") def RemoveBuddy(self, pubkey): pubkey = ''.join([chr(item) for item in pubkey]) if not self._buddies.has_key(pubkey): raise NotFoundError("Buddy not found") buddy = self._buddies[pubkey] activities = buddy.get_activities() # remove activity from the buddy buddy.leave_activities() # remove the buddy from all activities for act in activities: act.remove_buddy(buddy) self.BuddyDisappeared(buddy._object_path) del self._buddies[pubkey] @dbus.service.method(_PRESENCE_TEST_INTERFACE, in_signature="ays") def AddBuddyToActivity(self, pubkey, actid): pubkey = ''.join([chr(item) for item in pubkey]) if not self._buddies.has_key(pubkey): raise NotFoundError("Buddy unknown") if not self._activities.has_key(actid): raise NotFoundError("Activity unknown") buddy = self._buddies[pubkey] activity = self._activities[actid] activity.add_buddy(buddy) buddy.add_activity(activity) @dbus.service.method(_PRESENCE_TEST_INTERFACE, in_signature="ays") def RemoveBuddyFromActivity(self, pubkey, actid): pubkey = ''.join([chr(item) for item in pubkey]) if not self._buddies.has_key(pubkey): raise NotFoundError("Buddy unknown") if not self._activities.has_key(actid): raise NotFoundError("Activity unknown") buddy = self._buddies[pubkey] activity = self._activities[actid] buddy.remove_activity(activity) activity.remove_buddy(buddy) @dbus.service.method(_PRESENCE_TEST_INTERFACE, in_signature="ssssa{sv}") def AddActivity(self, actid, name, color, atype, properties): self._internal_share_activity(actid, atype, name, properties, color=color) @dbus.service.method(_PRESENCE_TEST_INTERFACE, in_signature="s") def RemoveActivity(self, actid): if not self._activities.has_key(actid): raise NotFoundError("Activity not found") act = self._activities[actid] # remove activity from all buddies for buddy in act.get_buddies(): buddy.remove_activity(act) act.disappear() self.ActivityDisappeared(act._object_path) del self._activities[actid] @dbus.service.method(_PRESENCE_TEST_INTERFACE, in_signature="ays") def SetBuddyCurrentActivity(self, pubkey, actid): pubkey = ''.join([chr(item) for item in pubkey]) if not self._buddies.has_key(pubkey): raise NotFoundError("Buddy unknown") buddy = self._buddies[pubkey] buddy.set_current_activity(actid) def main(): import logging logging.basicConfig(level=logging.DEBUG) loop = gobject.MainLoop() ps = TestPresenceService() loop.run() if __name__ == "__main__": main()