diff --git a/sugar/presence/presenceservice.py b/sugar/presence/presenceservice.py index 7592cdd4..c3ed5c70 100644 --- a/sugar/presence/presenceservice.py +++ b/sugar/presence/presenceservice.py @@ -29,71 +29,6 @@ import logging # see PEP: http://docs.python.org/whatsnew/pep-328.html import buddy, activity -class ObjectCache(object): - """Path to Activity/Buddy object cache - - On notification of a new object of either type the - PresenceService client stores the object's representation - in this object. - - XXX Why not just sub-class dict? We're only adding two - methods then and we would have all of the other - standard operations on dictionaries. - """ - def __init__(self): - """Initialise the cache""" - self._cache = {} - - def get(self, object_path): - """Retrieve specified object from the cache - - object_path -- full dbus path to the object - - returns a presence.buddy.Buddy or presence.activity.Activity - instance or None if the object_path is not yet cached. - - XXX could be written as return self._cache.get( object_path ) - """ - try: - return self._cache[object_path] - except KeyError: - return None - - def add(self, obj): - """Adds given presence object to the cache - - obj -- presence Buddy or Activity representation, the object's - object_path() method is used as the key for storage - - returns None - - XXX should raise an error on collisions, shouldn't it? or - return True/False to say whether the item was actually - added - """ - op = obj.object_path() - if not self._cache.has_key(op): - self._cache[op] = obj - - def remove(self, object_path): - """Remove the given presence object from the cache - - object_path -- full dbus path to the object - - returns None - - XXX does two checks instead of one with a try:except for the - keyerror, normal case of deleting existing penalised as - a result. - - try: - return self._cache.pop( key ) - except KeyError: - return None - """ - if self._cache.has_key(object_path): - del self._cache[object_path] - DBUS_SERVICE = "org.laptop.Sugar.Presence" DBUS_INTERFACE = "org.laptop.Sugar.Presence" @@ -131,26 +66,34 @@ class PresenceService(gobject.GObject): _PS_ACTIVITY_OP = DBUS_PATH + "/Activities/" - def __init__(self): + def __init__(self, allow_offline_iface=True): """Initialise the service and attempt to connect to events """ gobject.GObject.__init__(self) - self._objcache = ObjectCache() + self._objcache = {} + + # Get a connection to the session bus + self._bus = dbus.SessionBus() + self._bus.add_signal_receiver(self._name_owner_changed_cb, + signal_name="NameOwnerChanged", + dbus_interface="org.freedesktop.DBus") + # attempt to load the interface to the service... + self._allow_offline_iface = allow_offline_iface self._get_ps() - - _bus_ = None - def _get_bus( self ): - """Retrieve dbus session-bus or create new""" - if not self._bus_: - self._bus_ = dbus.SessionBus() - return self._bus_ - _bus = property( - _get_bus, None, None, - """DBUS SessionBus object for user-local communications""" - ) + + def _name_owner_changed_cb(self, name, old, new): + if name != DBUS_SERVICE: + return + if (old and len(old)) and (not new and not len(new)): + # PS went away, clear out PS dbus service wrapper + self._ps_ = None + elif (not old and not len(old)) and (new and len(new)): + # PS started up + self._get_ps() + _ps_ = None - def _get_ps( self ): + def _get_ps(self): """Retrieve dbus interface to PresenceService Also registers for updates from various dbus events on the @@ -175,7 +118,9 @@ class PresenceService(gobject.GObject): """Failure retrieving %r interface from the D-BUS service %r %r: %s""", DBUS_INTERFACE, DBUS_SERVICE, DBUS_PATH, err ) - return _OfflineInterface() + if self._allow_offline_iface: + return _OfflineInterface() + raise RuntimeError("Failed to connect to the presence service.") else: self._ps_ = ps ps.connect_to_signal('BuddyAppeared', self._buddy_appeared_cb) @@ -199,12 +144,16 @@ class PresenceService(gobject.GObject): Note that this method is called throughout the class whenever the representation of the object is required, it is not only - called when the object is first discovered. + called when the object is first discovered. The point is to only have + _one_ Python object for any D-Bus object represented by an object path, + effectively wrapping the D-Bus object in a single Python GObject. returns presence Buddy or Activity representation """ - obj = self._objcache.get(object_path) - if not obj: + obj = None + try: + obj = self._objcache[object_path] + except KeyError: if object_path.startswith(self._PS_BUDDY_OP): obj = buddy.Buddy(self._bus, self._new_object, self._del_object, object_path) @@ -213,12 +162,13 @@ class PresenceService(gobject.GObject): self._del_object, object_path) else: raise RuntimeError("Unknown object type") - self._objcache.add(obj) + self._objcache[object_path] = obj return obj def _del_object(self, object_path): - # FIXME - pass + """Fully remove an object from the object cache when it's no longer needed. + """ + del self._objcache[object_path] def _emit_buddy_appeared_signal(self, object_path): """Emit GObject event with presence.buddy.Buddy object""" @@ -248,13 +198,8 @@ class PresenceService(gobject.GObject): gobject.idle_add(self._emit_activity_invitation_signal, object_path) def _emit_private_invitation_signal(self, bus_name, connection, channel): - """Emit GObject event with bus_name, connection and channel - - XXX This seems to generate the wrong GObject event? It generates - 'service-disappeared' instead of private-invitation for some - reason. That event doesn't even seem to be registered? - """ - self.emit('service-disappeared', bus_name, connection, channel) + """Emit GObject event with bus_name, connection and channel""" + self.emit('private-invitation', bus_name, connection, channel) return False def _private_invitation_cb(self, bus_name, connection, channel): @@ -281,10 +226,8 @@ class PresenceService(gobject.GObject): gobject.idle_add(self._emit_activity_disappeared_signal, object_path) def get(self, object_path): - """Retrieve given object path as a Buddy/Activity object - - XXX This is basically just an alias for _new_object, i.e. it - just adds an extra function-call to the operation. + """Return the Buddy or Activity object corresponding to the given + D-Bus object path. """ return self._new_object(object_path) @@ -368,12 +311,7 @@ class PresenceService(gobject.GObject): return self._new_object(buddy_op) def get_owner(self): - """Retrieves "owner" as a Buddy - - XXX check that it really is a Buddy that's produced, what is - this the owner of? Shouldn't it be getting an activity - and then asking who the owner of that is? - """ + """Retrieves the laptop "owner" Buddy object.""" try: owner_op = self._ps.GetOwner() except dbus.exceptions.DBusException, err: @@ -381,7 +319,7 @@ class PresenceService(gobject.GObject): """Unable to retrieve local user/owner from presence service: %s""", err ) - return None + raise RuntimeError("Could not get owner object from presence service.") return self._new_object(owner_op) def _share_activity_cb(self, activity, op): @@ -499,10 +437,10 @@ class _MockPresenceService(gobject.GObject): return None _ps = None -def get_instance(): +def get_instance(allow_offline_iface=False): """Retrieve this process' view of the PresenceService""" global _ps if not _ps: - _ps = PresenceService() + _ps = PresenceService(allow_offline_iface) return _ps diff --git a/tests/presence/mockps.py b/tests/presence/mockps.py new file mode 100755 index 00000000..071f3d78 --- /dev/null +++ b/tests/presence/mockps.py @@ -0,0 +1,298 @@ +#!/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 + +class NotFoundError(dbus.DBusException): + def __init__(self): + dbus.DBusException.__init__(self) + 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, actid, name, color, atype): + self._actid = actid + self._name = name + self._color = color + self._type = atype + 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) + + @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, in_signature="", out_signature="s") + def GetId(self): + return self._actid + + @dbus.service.method(_ACTIVITY_INTERFACE, in_signature="", out_signature="s") + def GetColor(self): + return self._color + + @dbus.service.method(_ACTIVITY_INTERFACE, in_signature="", out_signature="s") + def GetType(self): + return self._type + + @dbus.service.method(_ACTIVITY_INTERFACE, in_signature="", out_signature="", + async_callbacks=('async_cb', 'async_err_cb')) + def Join(self, async_cb, async_err_cb): + pass + + @dbus.service.method(_ACTIVITY_INTERFACE, in_signature="", out_signature="ao") + def GetJoinedBuddies(self): + return [] + + @dbus.service.method(_ACTIVITY_INTERFACE, in_signature="", out_signature="soao") + def GetChannels(self): + return None + + @dbus.service.method(_ACTIVITY_INTERFACE, in_signature="", out_signature="s") + def GetName(self): + return self._name + + +_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) + + @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 key in self._activities.keys(): + acts.append(dbus.ObjectPath(key)) + 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 + + +_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 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) + + @dbus.service.method(_PRESENCE_INTERFACE, in_signature="sssa{sv}", + out_signature="o", async_callbacks=('async_cb', 'async_err_cb')) + def ShareActivity(self, actid, atype, name, properties, async_cb, async_err_cb): + pass + + @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 self._buddies.has_key(pubkey): + buddy = self._buddies[pubkey] + self.BuddyDisappeared(buddy._object_path) + del self._buddies[pubkey] + return + raise NotFoundError("Buddy not found") + +def main(): + loop = gobject.MainLoop() + ps = TestPresenceService() + loop.run() + +if __name__ == "__main__": + main() diff --git a/tests/presence/test-ps-bindings.py b/tests/presence/test-ps-bindings.py new file mode 100755 index 00000000..a0378a51 --- /dev/null +++ b/tests/presence/test-ps-bindings.py @@ -0,0 +1,228 @@ +#!/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 os, time +import dbus +import gobject, gtk +import unittest +from sugar.presence import presenceservice + +import mockps + +def start_ps(): + argv = ["mockps.py", "mockps.py"] + (pid, stdin, stdout, stderr) = gobject.spawn_async(argv, flags=gobject.SPAWN_LEAVE_DESCRIPTORS_OPEN) + + # Wait until it shows up on the bus + tries = 0 + bus = dbus.SessionBus() + while tries < 10: + time.sleep(0.5) + bus_object = bus.get_object('org.freedesktop.DBus', '/org/freedesktop/DBus') + try: + if bus_object.GetNameOwner(presenceservice.DBUS_SERVICE, dbus_interface='org.freedesktop.DBus'): + break + except dbus.exceptions.DBusException, err: + pass + tries += 1 + + if tries >= 5: + stop_ps(pid) + raise RuntimeError("Couldn't start the mock presence service") + + return pid + +def stop_ps(pid): + # EVIL HACK: get a new presence service object every time + del presenceservice._ps + presenceservice._ps = None + if pid >= 0: + os.kill(pid, 15) + + +class BuddyTests(unittest.TestCase): + def setUp(self): + self._pspid = start_ps() + + def tearDown(self): + if self._pspid > 0: + stop_ps(self._pspid) + self._pspid = -1 + + def _handle_error(self, err, user_data): + user_data["success"] = False + user_data["err"] = str(err) + gtk.main_quit() + + def _testOwner_helper(self, user_data): + try: + ps = presenceservice.get_instance(False) + except RuntimeError, err: + self._handle_error(err, user_data) + return False + + try: + owner = ps.get_owner() + except RuntimeError, err: + self._handle_error(err, user_data) + return False + + user_data["success"] = True + user_data["owner"] = owner + gtk.main_quit() + return False + + def testOwner(self): + user_data = {"success": False, "err": "", "owner": None} + gobject.idle_add(self._testOwner_helper, user_data) + gtk.main() + + assert user_data["success"] == True, user_data["err"] + assert user_data["owner"], "Owner could not be found." + + owner = user_data["owner"] + assert owner.props.key == mockps._OWNER_PUBKEY, "Owner public key doesn't match expected" + assert owner.props.nick == mockps._OWNER_NICK, "Owner nickname doesn't match expected" + assert owner.props.color == mockps._OWNER_COLOR, "Owner color doesn't match expected" + + _BA_PUBKEY = "akjadskjjfahfdahfdsahjfhfewaew3253232832832q098qewa98fdsafa98fa" + _BA_NICK = "BuddyAppearedTestBuddy" + _BA_COLOR = "#23adfb,#56bb11" + + def _testBuddyAppeared_helper_timeout(self, user_data): + self._handle_error("Timeout waiting for buddy-appeared signal", user_data) + return False + + def _testBuddyAppeared_helper_cb(self, ps, buddy, user_data): + user_data["buddy"] = buddy + user_data["success"] = True + gtk.main_quit() + + def _testBuddyAppeared_helper(self, user_data): + ps = presenceservice.get_instance(False) + ps.connect('buddy-appeared', self._testBuddyAppeared_helper_cb, user_data) + # Wait 5 seconds max for signal to be emitted + gobject.timeout_add(5000, self._testBuddyAppeared_helper_timeout, user_data) + + busobj = dbus.SessionBus().get_object(mockps._PRESENCE_SERVICE, + mockps._PRESENCE_PATH) + try: + testps = dbus.Interface(busobj, mockps._PRESENCE_TEST_INTERFACE) + except dbus.exceptions.DBusException, err: + self._handle_error(err, user_data) + return False + + try: + testps.AddBuddy(self._BA_PUBKEY, self._BA_NICK, self._BA_COLOR) + except dbus.exceptions.DBusException, err: + self._handle_error(err, user_data) + return False + + return False + + def testBuddyAppeared(self): + ps = presenceservice.get_instance(False) + assert ps, "Couldn't get presence service" + + user_data = {"success": False, "err": "", "buddy": None} + gobject.idle_add(self._testBuddyAppeared_helper, user_data) + gtk.main() + + assert user_data["success"] == True, user_data["err"] + assert user_data["buddy"], "Buddy was not received" + + buddy = user_data["buddy"] + assert buddy.props.key == self._BA_PUBKEY, "Public key doesn't match expected" + assert buddy.props.nick == self._BA_NICK, "Nickname doesn't match expected" + assert buddy.props.color == self._BA_COLOR, "Color doesn't match expected" + + # Try to get buddy by public key + buddy2 = ps.get_buddy(self._BA_PUBKEY) + assert buddy2, "Couldn't get buddy by public key" + assert buddy2.props.key == self._BA_PUBKEY, "Public key doesn't match expected" + assert buddy2.props.nick == self._BA_NICK, "Nickname doesn't match expected" + assert buddy2.props.color == self._BA_COLOR, "Color doesn't match expected" + + def _testBuddyDisappeared_helper_timeout(self, user_data): + self._handle_error("Timeout waiting for buddy-disappeared signal", user_data) + return False + + def _testBuddyDisappeared_helper_cb(self, ps, buddy, user_data): + user_data["buddy"] = buddy + user_data["success"] = True + gtk.main_quit() + + def _testBuddyDisappeared_helper(self, user_data): + busobj = dbus.SessionBus().get_object(mockps._PRESENCE_SERVICE, + mockps._PRESENCE_PATH) + try: + testps = dbus.Interface(busobj, mockps._PRESENCE_TEST_INTERFACE) + except dbus.exceptions.DBusException, err: + self._handle_error(err, user_data) + return False + + # Add a fake buddy + try: + testps.AddBuddy(self._BA_PUBKEY, self._BA_NICK, self._BA_COLOR) + except dbus.exceptions.DBusException, err: + self._handle_error(err, user_data) + return False + + ps = presenceservice.get_instance(False) + ps.connect('buddy-disappeared', self._testBuddyDisappeared_helper_cb, user_data) + # Wait 5 seconds max for signal to be emitted + gobject.timeout_add(5000, self._testBuddyDisappeared_helper_timeout, user_data) + + # Delete the fake buddy + try: + testps.RemoveBuddy(self._BA_PUBKEY) + except dbus.exceptions.DBusException, err: + self._handle_error(err, user_data) + return False + + return False + + def testBuddyDisappeared(self): + ps = presenceservice.get_instance(False) + assert ps, "Couldn't get presence service" + + user_data = {"success": False, "err": "", "buddy": None} + gobject.idle_add(self._testBuddyDisappeared_helper, user_data) + gtk.main() + + assert user_data["success"] == True, user_data["err"] + assert user_data["buddy"], "Buddy was not received" + + buddy = user_data["buddy"] + assert buddy.props.key == self._BA_PUBKEY, "Public key doesn't match expected" + assert buddy.props.nick == self._BA_NICK, "Nickname doesn't match expected" + assert buddy.props.color == self._BA_COLOR, "Color doesn't match expected" + + def addToSuite(suite): + suite.addTest(BuddyTests("testOwner")) + suite.addTest(BuddyTests("testBuddyAppeared")) + suite.addTest(BuddyTests("testBuddyDisappeared")) + addToSuite = staticmethod(addToSuite) + +def main(): + suite = unittest.TestSuite() + BuddyTests.addToSuite(suite) + runner = unittest.TextTestRunner() + runner.run(suite) + +if __name__ == "__main__": + main()