diff options
Diffstat (limited to 'tests')
| -rwxr-xr-x | tests/presence/mockps.py | 139 | ||||
| -rwxr-xr-x | tests/presence/test-ps-bindings.py | 297 |
2 files changed, 389 insertions, 47 deletions
diff --git a/tests/presence/mockps.py b/tests/presence/mockps.py index e166f5c..011c381 100755 --- a/tests/presence/mockps.py +++ b/tests/presence/mockps.py @@ -18,9 +18,15 @@ 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): - dbus.DBusException.__init__(self) + def __init__(self, msg=None): + dbus.DBusException.__init__(self, msg) self._dbus_error_name = _PRESENCE_INTERFACE + '.NotFound' @@ -28,17 +34,42 @@ _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): + 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 @@ -67,14 +98,18 @@ class TestActivity(dbus.service.Object): def GetType(self): return self._type - @dbus.service.method(_ACTIVITY_INTERFACE, - async_callbacks=('async_cb', 'async_err_cb')) - def Join(self, async_cb, async_err_cb): - pass + @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): - return [] + 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): @@ -106,6 +141,26 @@ class TestBuddy(dbus.service.Object): 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() + @dbus.service.signal(_BUDDY_INTERFACE, signature="ay") def IconChanged(self, icon_data): pass @@ -178,11 +233,6 @@ class TestOwner(TestBuddy): 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.""" @@ -261,10 +311,22 @@ class TestPresenceService(dbus.service.Object): 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", async_callbacks=('async_cb', 'async_err_cb')) - def ShareActivity(self, actid, atype, name, properties, async_cb, async_err_cb): - pass + 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): @@ -285,21 +347,54 @@ class TestPresenceService(dbus.service.Object): 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="ssss") - def AddActivity(self, actid, name, color, atype): - objid = self._get_next_object_id() - act = TestActivity(self._bus_name, objid, actid, name, color, atype) - self._activities[actid] = act - self.ActivityAppeared(act._object_path) + @dbus.service.method(_PRESENCE_TEST_INTERFACE, in_signature="oo") + 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="oo") + 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] diff --git a/tests/presence/test-ps-bindings.py b/tests/presence/test-ps-bindings.py index 15ee0e2..d855101 100755 --- a/tests/presence/test-ps-bindings.py +++ b/tests/presence/test-ps-bindings.py @@ -47,12 +47,22 @@ def start_ps(): return pid def stop_ps(pid): - # EVIL HACK: get a new presence service object every time + # 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): @@ -68,11 +78,15 @@ class GenericTestCase(unittest.TestCase): user_data["err"] = str(err) gtk.main_quit() + def cleanup(self, user_data): + if user_data.has_key("sources"): + for source in user_data["sources"]: + gobject.source_remove(source) class BuddyTests(GenericTestCase): def _testOwner_helper(self, user_data): try: - ps = presenceservice.get_instance(False) + ps = get_ps() except RuntimeError, err: self._handle_error(err, user_data) return False @@ -115,10 +129,12 @@ class BuddyTests(GenericTestCase): gtk.main_quit() def _testBuddyAppeared_helper(self, user_data): - ps = presenceservice.get_instance(False) - ps.connect('buddy-appeared', self._testBuddyAppeared_helper_cb, user_data) + ps = get_ps() + sid = ps.connect('buddy-appeared', self._testBuddyAppeared_helper_cb, user_data) + user_data["sources"].append(sid) # Wait 5 seconds max for signal to be emitted - gobject.timeout_add(5000, self._testBuddyAppeared_helper_timeout, user_data) + sid = gobject.timeout_add(5000, self._testBuddyAppeared_helper_timeout, user_data) + user_data["sources"].append(sid) busobj = dbus.SessionBus().get_object(mockps._PRESENCE_SERVICE, mockps._PRESENCE_PATH) @@ -137,12 +153,13 @@ class BuddyTests(GenericTestCase): return False def testBuddyAppeared(self): - ps = presenceservice.get_instance(False) + ps = get_ps() assert ps, "Couldn't get presence service" - user_data = {"success": False, "err": "", "buddy": None} + user_data = {"success": False, "err": "", "buddy": None, "sources": []} gobject.idle_add(self._testBuddyAppeared_helper, user_data) gtk.main() + self.cleanup(user_data) assert user_data["success"] == True, user_data["err"] assert user_data["buddy"], "Buddy was not received" @@ -184,10 +201,13 @@ class BuddyTests(GenericTestCase): self._handle_error(err, user_data) return False - ps = presenceservice.get_instance(False) - ps.connect('buddy-disappeared', self._testBuddyDisappeared_helper_cb, user_data) + ps = get_ps() + sid = ps.connect('buddy-disappeared', self._testBuddyDisappeared_helper_cb, user_data) + user_data["sources"].append(sid) + # Wait 5 seconds max for signal to be emitted - gobject.timeout_add(5000, self._testBuddyDisappeared_helper_timeout, user_data) + sid = gobject.timeout_add(5000, self._testBuddyDisappeared_helper_timeout, user_data) + user_data["sources"].append(sid) # Delete the fake buddy try: @@ -199,12 +219,13 @@ class BuddyTests(GenericTestCase): return False def testBuddyDisappeared(self): - ps = presenceservice.get_instance(False) + ps = get_ps() assert ps, "Couldn't get presence service" - user_data = {"success": False, "err": "", "buddy": None} + user_data = {"success": False, "err": "", "buddy": None, "sources": []} gobject.idle_add(self._testBuddyDisappeared_helper, user_data) gtk.main() + self.cleanup(user_data) assert user_data["success"] == True, user_data["err"] assert user_data["buddy"], "Buddy was not received" @@ -220,11 +241,33 @@ class BuddyTests(GenericTestCase): 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, user_data): self._handle_error("Timeout waiting for activity-appeared signal", user_data) @@ -236,10 +279,13 @@ class ActivityTests(GenericTestCase): gtk.main_quit() def _testActivityAppeared_helper(self, user_data): - ps = presenceservice.get_instance(False) - ps.connect('activity-appeared', self._testActivityAppeared_helper_cb, user_data) + ps = get_ps() + sid = ps.connect('activity-appeared', self._testActivityAppeared_helper_cb, user_data) + user_data["sources"].append(sid) + # Wait 5 seconds max for signal to be emitted - gobject.timeout_add(5000, self._testActivityAppeared_helper_timeout, user_data) + sid = gobject.timeout_add(5000, self._testActivityAppeared_helper_timeout, user_data) + user_data["sources"].append(sid) busobj = dbus.SessionBus().get_object(mockps._PRESENCE_SERVICE, mockps._PRESENCE_PATH) @@ -250,7 +296,7 @@ class ActivityTests(GenericTestCase): return False try: - testps.AddActivity(self._AA_ID, self._AA_NAME, self._AA_COLOR, self._AA_TYPE) + testps.AddActivity(self._AA_ID, self._AA_NAME, self._AA_COLOR, self._AA_TYPE, {}) except dbus.exceptions.DBusException, err: self._handle_error(err, user_data) return False @@ -258,12 +304,13 @@ class ActivityTests(GenericTestCase): return False def testActivityAppeared(self): - ps = presenceservice.get_instance(False) + ps = get_ps() assert ps, "Couldn't get presence service" - user_data = {"success": False, "err": "", "activity": None} + user_data = {"success": False, "err": "", "activity": None, "sources": []} gobject.idle_add(self._testActivityAppeared_helper, user_data) gtk.main() + self.cleanup(user_data) assert user_data["success"] == True, user_data["err"] assert user_data["activity"], "Activity was not received" @@ -303,15 +350,18 @@ class ActivityTests(GenericTestCase): # Add a fake activity try: - testps.AddActivity(self._AA_ID, self._AA_NAME, self._AA_COLOR, self._AA_TYPE) + testps.AddActivity(self._AA_ID, self._AA_NAME, self._AA_COLOR, self._AA_TYPE, {}) except dbus.exceptions.DBusException, err: self._handle_error(err, user_data) return False - ps = presenceservice.get_instance(False) - ps.connect('activity-disappeared', self._testActivityDisappeared_helper_cb, user_data) + ps = get_ps() + sid = ps.connect('activity-disappeared', self._testActivityDisappeared_helper_cb, user_data) + user_data["sources"].append(sid) + # Wait 5 seconds max for signal to be emitted - gobject.timeout_add(5000, self._testActivityDisappeared_helper_timeout, user_data) + sid = gobject.timeout_add(5000, self._testActivityDisappeared_helper_timeout, user_data) + user_data["sources"].append(sid) # Delete the fake activity try: @@ -323,12 +373,13 @@ class ActivityTests(GenericTestCase): return False def testActivityDisappeared(self): - ps = presenceservice.get_instance(False) + ps = get_ps() assert ps, "Couldn't get presence service" - user_data = {"success": False, "err": "", "activity": None} + user_data = {"success": False, "err": "", "activity": None, "sources": []} gobject.idle_add(self._testActivityDisappeared_helper, user_data) gtk.main() + self.cleanup(user_data) assert user_data["success"] == True, user_data["err"] assert user_data["activity"], "Activity was not received" @@ -340,12 +391,208 @@ class ActivityTests(GenericTestCase): assert act.props.type == self._AA_TYPE, "Type doesn't match expected" assert act.props.joined == False, "Joined doesn't match expected" + def _testActivityShare_helper_is_done(self, user_data): + if user_data["got-act-appeared"] and user_data["got-joined-activity"]: + user_data["success"] = True + gtk.main_quit() + + def _testActivityShare_helper_timeout(self, user_data): + self._handle_error("Timeout waiting for activity share", user_data) + return False + + def _testActivityShare_helper_joined_activity_cb(self, buddy, activity, user_data): + user_data["joined-activity-buddy"] = buddy + user_data["joined-activity-activity"] = activity + user_data["got-joined-activity"] = True + self._testActivityShare_helper_is_done(user_data) + + def _testActivityShare_helper_cb(self, ps, activity, user_data): + user_data["activity"] = activity + user_data["got-act-appeared"] = True + self._testActivityShare_helper_is_done(user_data) + + def _testActivityShare_helper(self, user_data): + ps = get_ps() + mockact = MockSugarActivity(self._AA_ID, self._AA_NAME, self._AA_TYPE) + + sid = ps.connect('activity-appeared', self._testActivityShare_helper_cb, user_data) + user_data["sources"].append(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, user_data) + user_data["sources"].append(sid) + except RuntimeError, err: + self._handle_error(err, user_data) + return False + + # Wait 5 seconds max for signal to be emitted + sid = gobject.timeout_add(5000, self._testActivityShare_helper_timeout, user_data) + user_data["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" + + user_data = {"success": False, + "err": "", + "sources": [], + "activity": None, + "got-act-appeared": False, + "joined-activity-buddy": None, + "joined-activity-activity": None, + "got-joined-activity": False + } + gobject.idle_add(self._testActivityShare_helper, user_data) + gtk.main() + self.cleanup(user_data) + + assert user_data["success"] == True, user_data["err"] + assert user_data["activity"], "Shared activity was not received" + + act = user_data["activity"] + assert act.props.id == self._AA_ID, "ID doesn't match expected" + assert act.props.name == self._AA_NAME, "Name doesn't match expected" + # Shared activities from local machine take the owner's color + assert act.props.color == mockps._OWNER_COLOR, "Color doesn't match expected" + assert act.props.type == self._AA_TYPE, "Type doesn't match expected" + assert act.props.joined == False, "Joined doesn't match expected" + + buddies = act.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 user_data["joined-activity-activity"] == act, "Activity mismatch" + assert user_data["joined-activity-buddy"] == owner, "Owner mismatch" + + def _testActivityJoin_helper_is_done(self, user_data): + if user_data["got-act-appeared"] and \ + user_data["got-joined-activity"] and \ + user_data["got-buddy-joined"]: + user_data["success"] = True + gtk.main_quit() + + def _testActivityJoin_helper_timeout(self, user_data): + self._handle_error("Timeout waiting for activity share", user_data) + return False + + def _testActivityJoin_helper_buddy_joined_cb(self, activity, buddy, user_data): + user_data["buddy-joined-buddy"] = buddy + user_data["buddy-joined-activity"] = activity + user_data["got-buddy-joined"] = True + self._testActivityJoin_helper_is_done(user_data) + + def _testActivityJoin_helper_joined_activity_cb(self, buddy, activity, user_data): + user_data["joined-activity-buddy"] = buddy + user_data["joined-activity-activity"] = activity + user_data["got-joined-activity"] = True + self._testActivityJoin_helper_is_done(user_data) + + def _testActivityJoin_helper_cb(self, ps, activity, user_data): + user_data["activity"] = activity + user_data["got-act-appeared"] = True + + # Hook up to the join signals + sid = activity.connect("buddy-joined", self._testActivityJoin_helper_buddy_joined_cb, user_data) + user_data["sources"].append(sid) + + ps = get_ps() + owner = ps.get_owner() + sid = owner.connect("joined-activity", self._testActivityJoin_helper_joined_activity_cb, user_data) + user_data["sources"].append(sid) + + # Join the activity + activity.join() + + def _testActivityJoin_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 + + ps = get_ps() + sid = ps.connect('activity-appeared', self._testActivityJoin_helper_cb, user_data) + user_data["sources"].append(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, user_data) + return False + + # Wait 5 seconds max for signal to be emitted + sid = gobject.timeout_add(5000, self._testActivityJoin_helper_timeout, user_data) + user_data["sources"].append(sid) + + return False + + def testActivityJoin(self): + ps = get_ps() + assert ps, "Couldn't get presence service" + + user_data = {"success": False, + "err": "", + "sources": [], + "activity": None, + "got-act-appeared": False, + "joined-activity-buddy": None, + "joined-activity-activity": None, + "got-joined-activity": False, + "buddy-joined-buddy": None, + "buddy-joined-activity": None, + "got-buddy-joined": False + } + gobject.idle_add(self._testActivityJoin_helper, user_data) + gtk.main() + self.cleanup(user_data) + + assert user_data["success"] == True, "Test unsuccessful" + assert user_data["activity"], "Shared activity was not received" + + act = user_data["activity"] + assert act.props.id == self._AA_ID, "ID doesn't match expected" + assert act.props.name == self._AA_NAME, "Name doesn't match expected" + + buddies = act.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 user_data["joined-activity-activity"] == act, "Activity mismatch" + assert user_data["joined-activity-buddy"] == owner, "Owner mismatch" + assert user_data["buddy-joined-activity"] == act, "Activity mismatch" + assert user_data["buddy-joined-buddy"] == owner, "Owner mismatch" + def addToSuite(suite): suite.addTest(ActivityTests("testActivityAppeared")) suite.addTest(ActivityTests("testActivityDisappeared")) + suite.addTest(ActivityTests("testActivityShare")) + suite.addTest(ActivityTests("testActivityJoin")) addToSuite = staticmethod(addToSuite) def main(): + import logging + logging.basicConfig(level=logging.DEBUG) + suite = unittest.TestSuite() BuddyTests.addToSuite(suite) ActivityTests.addToSuite(suite) |