diff options
Diffstat (limited to 'tests')
| -rwxr-xr-x | tests/presence/test-ps-bindings.py | 198 |
1 files changed, 170 insertions, 28 deletions
diff --git a/tests/presence/test-ps-bindings.py b/tests/presence/test-ps-bindings.py index 5363391..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" @@ -258,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) @@ -280,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" @@ -330,10 +355,13 @@ class ActivityTests(GenericTestCase): 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: @@ -345,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" @@ -383,31 +412,35 @@ class ActivityTests(GenericTestCase): self._testActivityShare_helper_is_done(user_data) def _testActivityShare_helper(self, user_data): - ps = presenceservice.get_instance(False) + ps = get_ps() mockact = MockSugarActivity(self._AA_ID, self._AA_NAME, self._AA_TYPE) - ps.connect('activity-appeared', self._testActivityShare_helper_cb, user_data) + 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() - owner.connect("joined-activity", self._testActivityShare_helper_joined_activity_cb, user_data) + 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 - gobject.timeout_add(5000, self._testActivityShare_helper_timeout, user_data) + 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 = presenceservice.get_instance(False) + 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, @@ -416,6 +449,7 @@ class ActivityTests(GenericTestCase): } 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" @@ -441,10 +475,118 @@ class ActivityTests(GenericTestCase): 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(): |