sugarlabs/sugar-toolkit-gtk3
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Merge branch 'master' of git+ssh://dev.laptop.org/git/sugar

Browse Source
This commit is contained in:
Marco Pesenti Gritti 2007-04-24 14:13:04 +02:00
commit 4e3b1c7ce2
3 changed files with 570 additions and 106 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

146
sugar/presence/presenceservice.py
View File

@ -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()
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()
_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"""
)
_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
)
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

298
tests/presence/mockps.py Executable file
View File

@ -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()

228
tests/presence/test-ps-bindings.py Executable file
View File

@ -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()