Kill the old tests. Trying to start a proper test suite.
Not sure about the usefulness of the presence service stuff but I guess it should be moved to the presence-service module anyway?
This commit is contained in:
parent
08cd44dbe2
commit
8403dd5987
@ -1,136 +0,0 @@
|
||||
#!/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, sys
|
||||
import dbus, dbus.glib
|
||||
import gobject
|
||||
|
||||
from sugar.presence import presenceservice
|
||||
from sugar.p2p import network
|
||||
|
||||
class ReadHTTPRequestHandler(network.ChunkedGlibHTTPRequestHandler):
|
||||
def translate_path(self, path):
|
||||
return self.server._filepath
|
||||
|
||||
class ReadHTTPServer(network.GlibTCPServer):
|
||||
def __init__(self, server_address, request_handler, filepath):
|
||||
self._filepath = filepath
|
||||
network.GlibTCPServer.__init__(self, server_address, request_handler);
|
||||
|
||||
class XMLRPCResponder(object):
|
||||
def __init__(self, have_file=False):
|
||||
self._have_file = have_file
|
||||
|
||||
def _set_have_file(self):
|
||||
self._have_file = True
|
||||
|
||||
def have_file(self):
|
||||
return self._have_file
|
||||
|
||||
|
||||
class MockReadActivity(gobject.GObject):
|
||||
__gproperties__ = {
|
||||
'title' : (str, None, None, None, gobject.PARAM_READABLE)
|
||||
}
|
||||
|
||||
def __init__(self, filepath):
|
||||
self._actid = "ef60b3af42f7b5aa558ef9269e2ed7998798afc0"
|
||||
self._name = "Test Read Activity"
|
||||
self._type = "org.laptop.sugar.ReadActivity"
|
||||
gobject.GObject.__init__(self)
|
||||
|
||||
self._ps_act = None
|
||||
self._filepath = os.path.abspath(filepath)
|
||||
self._file_server = ReadHTTPServer(("", 8867), ReadHTTPRequestHandler, self._filepath)
|
||||
|
||||
self._xmlrpc_server = network.GlibXMLRPCServer(("", 8868))
|
||||
responder = XMLRPCResponder(have_file=True)
|
||||
self._xmlrpc_server.register_instance(responder)
|
||||
|
||||
def _activity_appeared_cb(self, ps, activity):
|
||||
if activity.props.id != self._actid:
|
||||
return
|
||||
self._ps_act = activity
|
||||
|
||||
def share(self):
|
||||
ps = presenceservice.get_instance()
|
||||
ps.connect("activity-appeared", self._activity_appeared_cb)
|
||||
ps.share_activity(self)
|
||||
return False
|
||||
|
||||
def do_get_property(self, pspec):
|
||||
if pspec.name == "title":
|
||||
return self._name
|
||||
|
||||
def get_id(self):
|
||||
return self._actid
|
||||
|
||||
def get_service_name(self):
|
||||
return self._type
|
||||
|
||||
def start_ps():
|
||||
import commands
|
||||
(s, o) = commands.getstatusoutput("which sugar-presence-service")
|
||||
if s != 0:
|
||||
raise RuntimeError("Failed to find sugar presence service: %s" % o)
|
||||
argv = [o, "1"]
|
||||
(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")
|
||||
|
||||
print "Started presence service PID %d" % pid
|
||||
return pid
|
||||
|
||||
|
||||
def stop_ps(pid):
|
||||
if pid >= 0:
|
||||
os.kill(pid, 15)
|
||||
print "Stopped presence service PID %d" % pid
|
||||
|
||||
def main():
|
||||
if len(sys.argv) != 2:
|
||||
raise RuntimeError("Must specify a PDF to share.")
|
||||
path = os.path.abspath(sys.argv[1])
|
||||
if not os.path.exists(path):
|
||||
raise RuntimeError("File %s doesn't exist." % path)
|
||||
mact = MockReadActivity(path)
|
||||
pid = start_ps()
|
||||
loop = gobject.MainLoop()
|
||||
gobject.timeout_add(2000, mact.share)
|
||||
try:
|
||||
loop.run()
|
||||
except KeyboardInterrupt:
|
||||
pass
|
||||
stop_ps(pid)
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
@ -1,422 +0,0 @@
|
||||
#!/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()
|
@ -1,723 +0,0 @@
|
||||
#!/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"]
|
||||
(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; close the
|
||||
# connection to completely clear all signal matches too
|
||||
presenceservice._ps._bus.close()
|
||||
del presenceservice._ps
|
||||
presenceservice._ps = None
|
||||
if pid >= 0:
|
||||
os.kill(pid, 15)
|
||||
|
||||
def get_ps():
|
||||
ps = presenceservice.get_instance(False)
|
||||
# HACK
|
||||
# Set exit on disconnect to False so we don't get aborted when
|
||||
# explicitly closing the bus connection in stop_ps()
|
||||
ps._bus.set_exit_on_disconnect(False)
|
||||
return ps
|
||||
|
||||
|
||||
class GenericTestCase(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self._pspid = start_ps()
|
||||
self._success = False
|
||||
self._err = ""
|
||||
self._signals = []
|
||||
self._sources = []
|
||||
|
||||
def tearDown(self):
|
||||
# Remove all signal handlers
|
||||
for (obj, sid) in self._signals:
|
||||
obj.disconnect(sid)
|
||||
for source in self._sources:
|
||||
gobject.source_remove(source)
|
||||
|
||||
if self._pspid > 0:
|
||||
stop_ps(self._pspid)
|
||||
self._pspid = -1
|
||||
|
||||
def _handle_success(self):
|
||||
self._success = True
|
||||
gtk.main_quit()
|
||||
|
||||
def _handle_error(self, err):
|
||||
self._success = False
|
||||
self._err = str(err)
|
||||
gtk.main_quit()
|
||||
|
||||
class BuddyTests(GenericTestCase):
|
||||
def _testOwner_helper(self):
|
||||
try:
|
||||
ps = get_ps()
|
||||
except RuntimeError, err:
|
||||
self._handle_error(err)
|
||||
return False
|
||||
|
||||
try:
|
||||
owner = ps.get_owner()
|
||||
except RuntimeError, err:
|
||||
self._handle_error(err)
|
||||
return False
|
||||
|
||||
self._owner = owner
|
||||
self._handle_success()
|
||||
return False
|
||||
|
||||
def testOwner(self):
|
||||
gobject.idle_add(self._testOwner_helper)
|
||||
gtk.main()
|
||||
|
||||
assert self._success == True, "Test unsuccessful."
|
||||
assert self._owner, "Owner could not be found."
|
||||
|
||||
assert self._owner.props.key == mockps._OWNER_PUBKEY, "Owner public key doesn't match expected"
|
||||
assert self._owner.props.nick == mockps._OWNER_NICK, "Owner nickname doesn't match expected"
|
||||
assert self._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):
|
||||
self._handle_error("Timeout waiting for buddy-appeared signal")
|
||||
return False
|
||||
|
||||
def _testBuddyAppeared_helper_cb(self, ps, buddy):
|
||||
self._buddy = buddy
|
||||
self._handle_success()
|
||||
|
||||
def _testBuddyAppeared_helper(self):
|
||||
ps = get_ps()
|
||||
sid = ps.connect('buddy-appeared', self._testBuddyAppeared_helper_cb)
|
||||
self._signals.append((ps, sid))
|
||||
# Wait 5 seconds max for signal to be emitted
|
||||
sid = gobject.timeout_add(5000, self._testBuddyAppeared_helper_timeout)
|
||||
self._sources.append(sid)
|
||||
|
||||
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)
|
||||
return False
|
||||
|
||||
try:
|
||||
testps.AddBuddy(self._BA_PUBKEY, self._BA_NICK, self._BA_COLOR)
|
||||
except dbus.exceptions.DBusException, err:
|
||||
self._handle_error(err)
|
||||
return False
|
||||
|
||||
return False
|
||||
|
||||
def testBuddyAppeared(self):
|
||||
ps = get_ps()
|
||||
assert ps, "Couldn't get presence service"
|
||||
|
||||
self._buddy = None
|
||||
gobject.idle_add(self._testBuddyAppeared_helper)
|
||||
gtk.main()
|
||||
|
||||
assert self._success == True, "Test unsuccessful."
|
||||
assert self._buddy, "Buddy was not received"
|
||||
|
||||
assert self._buddy.props.key == self._BA_PUBKEY, "Public key doesn't match expected"
|
||||
assert self._buddy.props.nick == self._BA_NICK, "Nickname doesn't match expected"
|
||||
assert self._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):
|
||||
self._handle_error("Timeout waiting for buddy-disappeared signal")
|
||||
return False
|
||||
|
||||
def _testBuddyDisappeared_helper_cb(self, ps, buddy):
|
||||
self._buddy = buddy
|
||||
self._handle_success()
|
||||
|
||||
def _testBuddyDisappeared_helper(self):
|
||||
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)
|
||||
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)
|
||||
return False
|
||||
|
||||
ps = get_ps()
|
||||
sid = ps.connect('buddy-disappeared', self._testBuddyDisappeared_helper_cb)
|
||||
self._signals.append((ps, sid))
|
||||
|
||||
# Wait 5 seconds max for signal to be emitted
|
||||
sid = gobject.timeout_add(5000, self._testBuddyDisappeared_helper_timeout)
|
||||
self._sources.append(sid)
|
||||
|
||||
# Delete the fake buddy
|
||||
try:
|
||||
testps.RemoveBuddy(self._BA_PUBKEY)
|
||||
except dbus.exceptions.DBusException, err:
|
||||
self._handle_error(err)
|
||||
return False
|
||||
|
||||
return False
|
||||
|
||||
def testBuddyDisappeared(self):
|
||||
ps = get_ps()
|
||||
assert ps, "Couldn't get presence service"
|
||||
|
||||
self._buddy = None
|
||||
gobject.idle_add(self._testBuddyDisappeared_helper)
|
||||
gtk.main()
|
||||
|
||||
assert self._success == True, "Test unsuccessful."
|
||||
assert self._buddy, "Buddy was not received"
|
||||
|
||||
assert self._buddy.props.key == self._BA_PUBKEY, "Public key doesn't match expected"
|
||||
assert self._buddy.props.nick == self._BA_NICK, "Nickname doesn't match expected"
|
||||
assert self._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)
|
||||
|
||||
class MockSugarActivity(gobject.GObject):
|
||||
__gproperties__ = {
|
||||
'title' : (str, None, None, None, gobject.PARAM_READABLE)
|
||||
}
|
||||
|
||||
def __init__(self, actid, name, atype):
|
||||
self._actid = actid
|
||||
self._name = name
|
||||
self._type = atype
|
||||
gobject.GObject.__init__(self)
|
||||
|
||||
def do_get_property(self, pspec):
|
||||
if pspec.name == "title":
|
||||
return self._name
|
||||
|
||||
def get_id(self):
|
||||
return self._actid
|
||||
|
||||
def get_service_name(self):
|
||||
return self._type
|
||||
|
||||
class ActivityTests(GenericTestCase):
|
||||
_AA_ID = "d622b99b9f365d712296094b9f6110521e6c9cba"
|
||||
_AA_NAME = "Test Activity"
|
||||
_AA_TYPE = "org.laptop.Sugar.Foobar"
|
||||
_AA_COLOR = "#adfe20,#bf781a"
|
||||
_AA_PROPS = {"foo": "asdfadf", "bar":"5323aggdas"}
|
||||
|
||||
def _testActivityAppeared_helper_timeout(self):
|
||||
self._handle_error("Timeout waiting for activity-appeared signal")
|
||||
return False
|
||||
|
||||
def _testActivityAppeared_helper_cb(self, ps, activity):
|
||||
self._activity = activity
|
||||
self._handle_success()
|
||||
|
||||
def _testActivityAppeared_helper(self):
|
||||
ps = get_ps()
|
||||
sid = ps.connect('activity-appeared', self._testActivityAppeared_helper_cb)
|
||||
self._signals.append((ps, sid))
|
||||
|
||||
# Wait 5 seconds max for signal to be emitted
|
||||
sid = gobject.timeout_add(5000, self._testActivityAppeared_helper_timeout)
|
||||
self._sources.append(sid)
|
||||
|
||||
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)
|
||||
return False
|
||||
|
||||
try:
|
||||
testps.AddActivity(self._AA_ID, self._AA_NAME, self._AA_COLOR, self._AA_TYPE, {})
|
||||
except dbus.exceptions.DBusException, err:
|
||||
self._handle_error(err)
|
||||
return False
|
||||
|
||||
return False
|
||||
|
||||
def testActivityAppeared(self):
|
||||
ps = get_ps()
|
||||
assert ps, "Couldn't get presence service"
|
||||
|
||||
self._activity = None
|
||||
gobject.idle_add(self._testActivityAppeared_helper)
|
||||
gtk.main()
|
||||
|
||||
assert self._success == True, "Test unsuccessful"
|
||||
assert self._activity, "Activity was not received"
|
||||
|
||||
assert self._activity.props.id == self._AA_ID, "ID doesn't match expected"
|
||||
assert self._activity.props.name == self._AA_NAME, "Name doesn't match expected"
|
||||
assert self._activity.props.color == self._AA_COLOR, "Color doesn't match expected"
|
||||
assert self._activity.props.type == self._AA_TYPE, "Type doesn't match expected"
|
||||
assert self._activity.props.joined == False, "Joined doesn't match expected"
|
||||
|
||||
# Try to get activity by activity ID
|
||||
act2 = ps.get_activity(self._AA_ID)
|
||||
assert act2.props.id == self._AA_ID, "ID doesn't match expected"
|
||||
assert act2.props.name == self._AA_NAME, "Name doesn't match expected"
|
||||
assert act2.props.color == self._AA_COLOR, "Color doesn't match expected"
|
||||
assert act2.props.type == self._AA_TYPE, "Type doesn't match expected"
|
||||
assert act2.props.joined == False, "Joined doesn't match expected"
|
||||
|
||||
def _testActivityDisappeared_helper_timeout(self):
|
||||
self._handle_error("Timeout waiting for activity-disappeared signal")
|
||||
return False
|
||||
|
||||
def _testActivityDisappeared_helper_cb(self, ps, activity):
|
||||
self._activity = activity
|
||||
self._handle_success()
|
||||
|
||||
def _testActivityDisappeared_helper(self):
|
||||
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)
|
||||
return False
|
||||
|
||||
# Add a fake activity
|
||||
try:
|
||||
testps.AddActivity(self._AA_ID, self._AA_NAME, self._AA_COLOR, self._AA_TYPE, {})
|
||||
except dbus.exceptions.DBusException, err:
|
||||
self._handle_error(err)
|
||||
return False
|
||||
|
||||
ps = get_ps()
|
||||
sid = ps.connect('activity-disappeared', self._testActivityDisappeared_helper_cb)
|
||||
self._signals.append((ps, sid))
|
||||
|
||||
# Wait 5 seconds max for signal to be emitted
|
||||
sid = gobject.timeout_add(5000, self._testActivityDisappeared_helper_timeout)
|
||||
self._sources.append(sid)
|
||||
|
||||
# Delete the fake activity
|
||||
try:
|
||||
testps.RemoveActivity(self._AA_ID)
|
||||
except dbus.exceptions.DBusException, err:
|
||||
self._handle_error(err)
|
||||
return False
|
||||
|
||||
return False
|
||||
|
||||
def testActivityDisappeared(self):
|
||||
ps = get_ps()
|
||||
assert ps, "Couldn't get presence service"
|
||||
|
||||
self._activity = None
|
||||
gobject.idle_add(self._testActivityDisappeared_helper)
|
||||
gtk.main()
|
||||
|
||||
assert self._success == True, "Test unsuccessful"
|
||||
assert self._activity, "Activity was not received"
|
||||
|
||||
assert self._activity.props.id == self._AA_ID, "ID doesn't match expected"
|
||||
assert self._activity.props.name == self._AA_NAME, "Name doesn't match expected"
|
||||
assert self._activity.props.color == self._AA_COLOR, "Color doesn't match expected"
|
||||
assert self._activity.props.type == self._AA_TYPE, "Type doesn't match expected"
|
||||
assert self._activity.props.joined == False, "Joined doesn't match expected"
|
||||
|
||||
def _testActivityShare_helper_is_done(self):
|
||||
if self._got_act_appeared and self._got_joined_activity:
|
||||
self._handle_success()
|
||||
|
||||
def _testActivityShare_helper_timeout(self):
|
||||
self._handle_error("Timeout waiting for activity share")
|
||||
return False
|
||||
|
||||
def _testActivityShare_helper_joined_activity_cb(self, buddy, activity):
|
||||
self._joined_activity_buddy = buddy
|
||||
self._joined_activity_activity = activity
|
||||
self._got_joined_activity = True
|
||||
self._testActivityShare_helper_is_done()
|
||||
|
||||
def _testActivityShare_helper_cb(self, ps, activity):
|
||||
self._activity = activity
|
||||
self._got_act_appeared = True
|
||||
self._testActivityShare_helper_is_done()
|
||||
|
||||
def _testActivityShare_helper(self):
|
||||
ps = get_ps()
|
||||
mockact = MockSugarActivity(self._AA_ID, self._AA_NAME, self._AA_TYPE)
|
||||
|
||||
sid = ps.connect('activity-appeared', self._testActivityShare_helper_cb)
|
||||
self._signals.append((ps, sid))
|
||||
try:
|
||||
# Hook up to the owner's joined-activity signal
|
||||
owner = ps.get_owner()
|
||||
sid = owner.connect("joined-activity", self._testActivityShare_helper_joined_activity_cb)
|
||||
self._signals.append((owner, sid))
|
||||
except RuntimeError, err:
|
||||
self._handle_error(err)
|
||||
return False
|
||||
|
||||
# Wait 5 seconds max for signal to be emitted
|
||||
sid = gobject.timeout_add(5000, self._testActivityShare_helper_timeout)
|
||||
self._sources.append(sid)
|
||||
|
||||
ps.share_activity(mockact, self._AA_PROPS)
|
||||
|
||||
return False
|
||||
|
||||
def testActivityShare(self):
|
||||
ps = get_ps()
|
||||
assert ps, "Couldn't get presence service"
|
||||
|
||||
self._activity = None
|
||||
self._got_act_appeared = False
|
||||
self._joined_activity_buddy = None
|
||||
self._joined_activity_activity = None
|
||||
self._got_joined_activity = False
|
||||
gobject.idle_add(self._testActivityShare_helper)
|
||||
gtk.main()
|
||||
|
||||
assert self._success == True, "Test unsuccessful."
|
||||
assert self._activity, "Shared activity was not received"
|
||||
|
||||
assert self._activity.props.id == self._AA_ID, "ID doesn't match expected"
|
||||
assert self._activity.props.name == self._AA_NAME, "Name doesn't match expected"
|
||||
# Shared activities from local machine take the owner's color
|
||||
assert self._activity.props.color == mockps._OWNER_COLOR, "Color doesn't match expected"
|
||||
assert self._activity.props.type == self._AA_TYPE, "Type doesn't match expected"
|
||||
assert self._activity.props.joined == False, "Joined doesn't match expected"
|
||||
|
||||
buddies = self._activity.get_joined_buddies()
|
||||
assert len(buddies) == 1, "No buddies in activity"
|
||||
owner = buddies[0]
|
||||
assert owner.props.key == mockps._OWNER_PUBKEY, "Buddy key doesn't match expected"
|
||||
assert owner.props.nick == mockps._OWNER_NICK, "Buddy nick doesn't match expected"
|
||||
assert owner.props.color == mockps._OWNER_COLOR, "Buddy color doesn't match expected"
|
||||
|
||||
real_owner = ps.get_owner()
|
||||
assert real_owner == owner, "Owner mismatch"
|
||||
|
||||
assert self._joined_activity_activity == self._activity, "Activity mismatch"
|
||||
assert self._joined_activity_buddy == owner, "Owner mismatch"
|
||||
|
||||
def _testActivityJoin_helper_is_done(self):
|
||||
if self._got_act_appeared and self._got_joined_activity and \
|
||||
self._got_buddy_joined:
|
||||
self._handle_success()
|
||||
|
||||
def _testActivityJoin_helper_timeout(self):
|
||||
self._handle_error("Timeout waiting for activity share")
|
||||
return False
|
||||
|
||||
def _testActivityJoin_helper_buddy_joined_cb(self, activity, buddy):
|
||||
self._buddy_joined_buddy = buddy
|
||||
self._buddy_joined_activity = activity
|
||||
self._got_buddy_joined = True
|
||||
self._testActivityJoin_helper_is_done()
|
||||
|
||||
def _testActivityJoin_helper_joined_activity_cb(self, buddy, activity):
|
||||
self._joined_activity_buddy = buddy
|
||||
self._joined_activity_activity = activity
|
||||
self._got_joined_activity = True
|
||||
self._testActivityJoin_helper_is_done()
|
||||
|
||||
def _testActivityJoin_helper_cb(self, ps, activity):
|
||||
self._activity = activity
|
||||
self._got_act_appeared = True
|
||||
|
||||
# Hook up to the join signals
|
||||
sid = activity.connect("buddy-joined", self._testActivityJoin_helper_buddy_joined_cb)
|
||||
self._signals.append((activity, sid))
|
||||
|
||||
ps = get_ps()
|
||||
owner = ps.get_owner()
|
||||
sid = owner.connect("joined-activity", self._testActivityJoin_helper_joined_activity_cb)
|
||||
self._signals.append((owner, sid))
|
||||
|
||||
# Join the activity
|
||||
activity.join()
|
||||
|
||||
def _testActivityJoin_helper(self):
|
||||
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)
|
||||
return False
|
||||
|
||||
ps = get_ps()
|
||||
sid = ps.connect('activity-appeared', self._testActivityJoin_helper_cb)
|
||||
self._signals.append((ps, sid))
|
||||
|
||||
# Add a fake activity
|
||||
try:
|
||||
testps.AddActivity(self._AA_ID, self._AA_NAME, self._AA_COLOR, self._AA_TYPE, {})
|
||||
except dbus.exceptions.DBusException, err:
|
||||
self._handle_error(err)
|
||||
return False
|
||||
|
||||
# Wait 5 seconds max for signal to be emitted
|
||||
sid = gobject.timeout_add(5000, self._testActivityJoin_helper_timeout)
|
||||
self._sources.append(sid)
|
||||
|
||||
return False
|
||||
|
||||
def testActivityJoin(self):
|
||||
ps = get_ps()
|
||||
assert ps, "Couldn't get presence service"
|
||||
|
||||
self._activity = None
|
||||
self._got_act_appeared = False
|
||||
self._joined_activity_buddy = None
|
||||
self._joined_activity_activity = None
|
||||
self._got_joined_activity = False
|
||||
self._buddy_joined_buddy = None
|
||||
self._buddy_joined_activity = None
|
||||
self._got_buddy_joined = False
|
||||
gobject.idle_add(self._testActivityJoin_helper)
|
||||
gtk.main()
|
||||
|
||||
assert self._success == True, "Test unsuccessful"
|
||||
assert self._activity, "Shared activity was not received"
|
||||
|
||||
assert self._activity.props.id == self._AA_ID, "ID doesn't match expected"
|
||||
assert self._activity.props.name == self._AA_NAME, "Name doesn't match expected"
|
||||
|
||||
buddies = self._activity.get_joined_buddies()
|
||||
assert len(buddies) == 1, "No buddies in activity"
|
||||
owner = buddies[0]
|
||||
assert owner.props.key == mockps._OWNER_PUBKEY, "Buddy key doesn't match expected"
|
||||
assert owner.props.nick == mockps._OWNER_NICK, "Buddy nick doesn't match expected"
|
||||
assert owner.props.color == mockps._OWNER_COLOR, "Buddy color doesn't match expected"
|
||||
|
||||
real_owner = ps.get_owner()
|
||||
assert real_owner == owner, "Owner mismatch"
|
||||
|
||||
assert self._joined_activity_activity == self._activity, "Activity mismatch"
|
||||
assert self._joined_activity_buddy == owner, "Owner mismatch"
|
||||
assert self._buddy_joined_activity == self._activity, "Activity mismatch"
|
||||
assert self._buddy_joined_buddy == owner, "Owner mismatch"
|
||||
|
||||
def _testCurrentActivity_helper_timeout(self):
|
||||
self._handle_error("Timeout waiting for current activity")
|
||||
return False
|
||||
|
||||
def _testCurrentActivity_set_current_activity(self, actid):
|
||||
busobj = dbus.SessionBus().get_object(mockps._PRESENCE_SERVICE,
|
||||
mockps._PRESENCE_PATH)
|
||||
try:
|
||||
testps = dbus.Interface(busobj, mockps._PRESENCE_TEST_INTERFACE)
|
||||
testps.SetBuddyCurrentActivity(self._buddy.props.key, actid)
|
||||
except dbus.exceptions.DBusException, err:
|
||||
self._handle_error(err)
|
||||
return
|
||||
|
||||
def _testCurrentActivity_buddy_property_changed_cb(self, buddy, proplist):
|
||||
if not self._start_monitor:
|
||||
return
|
||||
if not 'current-activity' in proplist:
|
||||
return
|
||||
buddy_curact = buddy.props.current_activity
|
||||
if buddy_curact.props.id == self._AA_ID:
|
||||
self._got_first_curact = True
|
||||
# set next current activity
|
||||
self._testCurrentActivity_set_current_activity(self._other_actid)
|
||||
elif buddy_curact.props.id == self._other_actid:
|
||||
self._got_other_curact = True
|
||||
|
||||
if self._got_first_curact and self._got_other_curact:
|
||||
self._handle_success()
|
||||
|
||||
def _testCurrentActivity_start_monitor_helper(self):
|
||||
if len(self._activities) != 2 or not self._buddy:
|
||||
return
|
||||
self._start_monitor = True
|
||||
# Set first current activity
|
||||
self._testCurrentActivity_set_current_activity(self._AA_ID)
|
||||
|
||||
def _testCurrentActivity_activity_helper_cb(self, ps, activity):
|
||||
if activity in self._activities:
|
||||
self._handle_error("Activity %s already known." % activity.props.id)
|
||||
self._activities.append(activity)
|
||||
self._testCurrentActivity_start_monitor_helper()
|
||||
|
||||
def _testCurrentActivity_buddy_helper_cb(self, ps, buddy):
|
||||
self._buddy = buddy
|
||||
sid = buddy.connect("property-changed", self._testCurrentActivity_buddy_property_changed_cb)
|
||||
self._signals.append((buddy, sid))
|
||||
self._testCurrentActivity_start_monitor_helper()
|
||||
|
||||
def _testCurrentActivity_helper(self):
|
||||
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)
|
||||
return False
|
||||
|
||||
ps = get_ps()
|
||||
sid = ps.connect('activity-appeared', self._testCurrentActivity_activity_helper_cb)
|
||||
self._signals.append((ps, sid))
|
||||
sid = ps.connect('buddy-appeared', self._testCurrentActivity_buddy_helper_cb)
|
||||
self._signals.append((ps, sid))
|
||||
|
||||
# Add a fake buddy
|
||||
try:
|
||||
testps.AddBuddy(BuddyTests._BA_PUBKEY, BuddyTests._BA_NICK, BuddyTests._BA_COLOR)
|
||||
except dbus.exceptions.DBusException, err:
|
||||
self._handle_error(err)
|
||||
return False
|
||||
|
||||
# Add first fake activity
|
||||
try:
|
||||
testps.AddActivity(self._AA_ID, self._AA_NAME, self._AA_COLOR, self._AA_TYPE, {})
|
||||
testps.AddBuddyToActivity(BuddyTests._BA_PUBKEY, self._AA_ID)
|
||||
except dbus.exceptions.DBusException, err:
|
||||
self._handle_error(err)
|
||||
return False
|
||||
|
||||
# Add second fake activity
|
||||
try:
|
||||
testps.AddActivity(self._other_actid, self._other_actname,
|
||||
self._other_actcolor, self._AA_TYPE, {})
|
||||
testps.AddBuddyToActivity(BuddyTests._BA_PUBKEY, self._other_actid)
|
||||
except dbus.exceptions.DBusException, err:
|
||||
self._handle_error(err)
|
||||
return False
|
||||
|
||||
# Wait 10 seconds max for everything to complete
|
||||
sid = gobject.timeout_add(10000, self._testCurrentActivity_helper_timeout)
|
||||
self._sources.append(sid)
|
||||
|
||||
return False
|
||||
|
||||
def testCurrentActivity(self):
|
||||
ps = get_ps()
|
||||
assert ps, "Couldn't get presence service"
|
||||
|
||||
self._other_actid = "ea8a94522c53a6741e141adece1711e4d9884678"
|
||||
self._other_actname = "Some random activity"
|
||||
self._other_actcolor = "#073838,#3A6E3A"
|
||||
self._activities = []
|
||||
self._got_first_curact = False
|
||||
self._got_other_curact = False
|
||||
self._start_monitor = False
|
||||
gobject.idle_add(self._testCurrentActivity_helper)
|
||||
gtk.main()
|
||||
|
||||
assert self._success == True, "Test unsuccessful"
|
||||
assert len(self._activities) == 2, "Shared activities were not received"
|
||||
assert self._got_first_curact == True, "Couldn't discover first activity"
|
||||
assert self._got_other_curact == True, "Couldn't discover second activity"
|
||||
assert self._start_monitor == True, "Couldn't discover both activities"
|
||||
|
||||
# check the buddy
|
||||
assert self._buddy.props.key == BuddyTests._BA_PUBKEY, "Buddy key doesn't match expected"
|
||||
assert self._buddy.props.nick == BuddyTests._BA_NICK, "Buddy nick doesn't match expected"
|
||||
assert self._buddy.props.color == BuddyTests._BA_COLOR, "Buddy color doesn't match expected"
|
||||
assert self._buddy.props.current_activity.props.id == self._other_actid, "Buddy current activity didn't match expected"
|
||||
|
||||
# check both activities
|
||||
found = 0
|
||||
for act in self._activities:
|
||||
if act.props.id == self._AA_ID:
|
||||
assert act.props.name == self._AA_NAME, "Name doesn't match expected"
|
||||
assert act.props.color == self._AA_COLOR, "Color doesn't match expected"
|
||||
buddies = act.get_joined_buddies()
|
||||
assert len(buddies) == 1, "Unexpected number of buddies in first activity"
|
||||
assert buddies[0] == self._buddy, "Unexpected buddy in first activity"
|
||||
found += 1
|
||||
elif act.props.id == self._other_actid:
|
||||
assert act.props.name == self._other_actname, "Name doesn't match expected"
|
||||
assert act.props.color == self._other_actcolor, "Color doesn't match expected"
|
||||
buddies = act.get_joined_buddies()
|
||||
assert len(buddies) == 1, "Unexpected number of buddies in first activity"
|
||||
assert buddies[0] == self._buddy, "Unexpected buddy in first activity"
|
||||
found += 1
|
||||
|
||||
assert found == 2, "Couldn't discover both activities"
|
||||
|
||||
def addToSuite(suite):
|
||||
suite.addTest(ActivityTests("testActivityAppeared"))
|
||||
suite.addTest(ActivityTests("testActivityDisappeared"))
|
||||
suite.addTest(ActivityTests("testActivityShare"))
|
||||
suite.addTest(ActivityTests("testActivityJoin"))
|
||||
suite.addTest(ActivityTests("testCurrentActivity"))
|
||||
addToSuite = staticmethod(addToSuite)
|
||||
|
||||
def main():
|
||||
import logging
|
||||
logging.basicConfig(level=logging.DEBUG)
|
||||
|
||||
suite = unittest.TestSuite()
|
||||
BuddyTests.addToSuite(suite)
|
||||
ActivityTests.addToSuite(suite)
|
||||
runner = unittest.TextTestRunner()
|
||||
runner.run(suite)
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
@ -1,30 +0,0 @@
|
||||
#!/usr/bin/env python
|
||||
|
||||
# Copyright (C) 2006, 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 sys
|
||||
|
||||
from sugar import objects
|
||||
|
||||
print 'MIME type for test.pdf (from extension):'
|
||||
print objects.mime.get_from_file_name('test.pdf')
|
||||
|
||||
print ''
|
||||
|
||||
if len(sys.argv) > 1:
|
||||
print 'MIME type for file %s:' % sys.argv[1]
|
||||
print objects.mime.get_for_file(sys.argv[1])
|
@ -1,42 +0,0 @@
|
||||
#!/usr/bin/env python
|
||||
|
||||
# Copyright (C) 2007, Eduardo Silva (edsiper@gmail.com)
|
||||
#
|
||||
# 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 pygtk
|
||||
pygtk.require('2.0')
|
||||
import gtk
|
||||
|
||||
from sugar.graphics.notebook import Notebook
|
||||
|
||||
window = gtk.Window()
|
||||
window.connect("destroy", lambda w: gtk.main_quit())
|
||||
window.set_size_request(800, 600)
|
||||
window.show_all()
|
||||
|
||||
nb = Notebook(can_close_tabs=True)
|
||||
|
||||
window.add(nb)
|
||||
|
||||
button1 = gtk.Button('Example 1')
|
||||
button2 = gtk.Button('Example 2')
|
||||
button3 = gtk.Button('Example 3')
|
||||
|
||||
nb.add_page('Testing label 1', button1)
|
||||
nb.add_page('Testing label 2', button2)
|
||||
nb.add_page('Testing label 3', button3)
|
||||
|
||||
gtk.main()
|
@ -1,66 +0,0 @@
|
||||
#!/usr/bin/env python
|
||||
|
||||
# Copyright (C) 2006, 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 sys
|
||||
|
||||
import gtk
|
||||
import hippo
|
||||
|
||||
from sugar.graphics.xocolor import XoColor
|
||||
from sugar.graphics.canvasicon import CanvasIcon
|
||||
from sugar import env
|
||||
|
||||
sys.path.append(env.get_shell_path())
|
||||
|
||||
from view.home.snowflakelayout import SnowflakeLayout
|
||||
|
||||
def add_snowflake(parent, size):
|
||||
box = hippo.CanvasBox()
|
||||
parent.append(box)
|
||||
|
||||
layout = SnowflakeLayout()
|
||||
box.set_layout(layout)
|
||||
|
||||
icon = CanvasIcon(scale=0.8, xo_color=XoColor(),
|
||||
icon_name='theme:xo')
|
||||
layout.add_center(icon)
|
||||
|
||||
for k in range(0, size):
|
||||
icon = CanvasIcon(scale=0.4, xo_color=XoColor(),
|
||||
icon_name='theme:xo')
|
||||
layout.add(icon)
|
||||
|
||||
window = gtk.Window()
|
||||
window.set_default_size(gtk.gdk.screen_width(), gtk.gdk.screen_height())
|
||||
window.connect("destroy", lambda w: gtk.main_quit())
|
||||
window.show()
|
||||
|
||||
canvas = hippo.Canvas()
|
||||
|
||||
root = hippo.CanvasBox(background_color=0xe2e2e2ff)
|
||||
canvas.set_root(root)
|
||||
|
||||
add_snowflake(root, 10)
|
||||
add_snowflake(root, 20)
|
||||
add_snowflake(root, 15)
|
||||
add_snowflake(root, 5)
|
||||
|
||||
canvas.show()
|
||||
window.add(canvas)
|
||||
|
||||
gtk.main()
|
@ -1,62 +0,0 @@
|
||||
#!/usr/bin/env python
|
||||
|
||||
# Copyright (C) 2006, 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 sys
|
||||
import random
|
||||
|
||||
import pygtk
|
||||
pygtk.require('2.0')
|
||||
import gobject
|
||||
|
||||
import gtk
|
||||
import hippo
|
||||
|
||||
from sugar.graphics.spreadlayout import SpreadLayout
|
||||
from sugar.graphics.xocolor import XoColor
|
||||
from sugar.graphics.canvasicon import CanvasIcon
|
||||
|
||||
def _create_icon():
|
||||
color = XoColor()
|
||||
|
||||
scale = 1.0 + random.random() * 1.5
|
||||
icon = CanvasIcon(scale=scale, xo_color=color,
|
||||
icon_name='theme:xo')
|
||||
icon.set_tooltip('Test')
|
||||
layout.add(icon)
|
||||
|
||||
return (len(box.get_children()) < 50)
|
||||
|
||||
window = gtk.Window()
|
||||
window.connect("destroy", lambda w: gtk.main_quit())
|
||||
window.show()
|
||||
|
||||
canvas = hippo.Canvas()
|
||||
|
||||
box = hippo.CanvasBox(background_color=0xe2e2e2ff)
|
||||
|
||||
layout = SpreadLayout()
|
||||
box.set_layout(layout)
|
||||
|
||||
canvas.set_root(box)
|
||||
|
||||
window.add(canvas)
|
||||
canvas.show()
|
||||
|
||||
gobject.timeout_add(200, _create_icon)
|
||||
|
||||
gtk.main()
|
113
tests/test-ui.py
113
tests/test-ui.py
@ -1,113 +0,0 @@
|
||||
#!/usr/bin/env python
|
||||
|
||||
# Copyright (C) 2006, 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 gtk
|
||||
|
||||
from sugar.graphics.window import Window
|
||||
from sugar.graphics.toolbutton import ToolButton
|
||||
from sugar.graphics.toolbox import Toolbox
|
||||
from sugar.graphics.palette import Palette
|
||||
|
||||
class EditToolbar(gtk.Toolbar):
|
||||
def __init__(self):
|
||||
gtk.Toolbar.__init__(self)
|
||||
|
||||
class TextToolbar(gtk.Toolbar):
|
||||
def __init__(self):
|
||||
gtk.Toolbar.__init__(self)
|
||||
|
||||
button = ToolButton('text-format-bold')
|
||||
self.insert(button, -1)
|
||||
button.show()
|
||||
|
||||
palette = Palette('Title')
|
||||
button.set_palette(palette)
|
||||
|
||||
menu_item = gtk.MenuItem('First menu item')
|
||||
palette.menu.append(menu_item)
|
||||
menu_item.show()
|
||||
|
||||
menu_item = gtk.MenuItem('Second menu item')
|
||||
palette.menu.append(menu_item)
|
||||
menu_item.show()
|
||||
|
||||
menu_item = gtk.MenuItem('Third menu item')
|
||||
palette.menu.append(menu_item)
|
||||
menu_item.show()
|
||||
|
||||
class ImageToolbar(gtk.Toolbar):
|
||||
def __init__(self):
|
||||
gtk.Toolbar.__init__(self)
|
||||
|
||||
class TableToolbar(gtk.Toolbar):
|
||||
def __init__(self):
|
||||
gtk.Toolbar.__init__(self)
|
||||
|
||||
class FormatToolbar(gtk.Toolbar):
|
||||
def __init__(self):
|
||||
gtk.Toolbar.__init__(self)
|
||||
|
||||
class ViewToolbar(gtk.Toolbar):
|
||||
def __init__(self):
|
||||
gtk.Toolbar.__init__(self)
|
||||
|
||||
window = Window()
|
||||
window.connect("destroy", lambda w: gtk.main_quit())
|
||||
|
||||
toolbox = Toolbox()
|
||||
window.set_toolbox(toolbox)
|
||||
toolbox.show()
|
||||
|
||||
edit_toolbar = EditToolbar()
|
||||
toolbox.add_toolbar('Edit', edit_toolbar)
|
||||
edit_toolbar.show()
|
||||
|
||||
text_toolbar = TextToolbar()
|
||||
toolbox.add_toolbar('Text', text_toolbar)
|
||||
text_toolbar.show()
|
||||
|
||||
image_toolbar = ImageToolbar()
|
||||
toolbox.add_toolbar('Image', image_toolbar)
|
||||
image_toolbar.show()
|
||||
|
||||
table_toolbar = TableToolbar()
|
||||
toolbox.add_toolbar('Table', table_toolbar)
|
||||
table_toolbar.show()
|
||||
|
||||
format_toolbar = FormatToolbar()
|
||||
toolbox.add_toolbar('Format', format_toolbar)
|
||||
format_toolbar.show()
|
||||
|
||||
view_toolbar = ViewToolbar()
|
||||
toolbox.add_toolbar('View', view_toolbar)
|
||||
view_toolbar.show()
|
||||
|
||||
toolbox.set_current_toolbar(1)
|
||||
|
||||
scrolled_window = gtk.ScrolledWindow()
|
||||
scrolled_window.set_policy(gtk.POLICY_NEVER, gtk.POLICY_ALWAYS)
|
||||
window.set_canvas(scrolled_window)
|
||||
scrolled_window.show()
|
||||
|
||||
text_view = gtk.TextView()
|
||||
scrolled_window.add(text_view)
|
||||
text_view.show()
|
||||
|
||||
window.show()
|
||||
|
||||
gtk.main()
|
Loading…
Reference in New Issue
Block a user