pitivi r1321 - in trunk: . pitivi tests

[edwardrv svn gnome org]

Author: edwardrv
Date: Thu Oct 16 14:19:55 2008
New Revision: 1321
URL: http://svn.gnome.org/viewvc/pitivi?rev=1321&view=rev

Log:
	* pitivi/Makefile.am:
	* pitivi/signalinterface.py:
	* tests/Makefile.am:
	* tests/test_signallable.py:
	New signallable interface

Added:
   trunk/pitivi/signalinterface.py
   trunk/tests/test_signallable.py
Modified:
   trunk/ChangeLog
   trunk/pitivi/Makefile.am
   trunk/tests/Makefile.am

Modified: trunk/pitivi/Makefile.am
==============================================================================
--- trunk/pitivi/Makefile.am	(original)
+++ trunk/pitivi/Makefile.am	Thu Oct 16 14:19:55 2008
@@ -27,6 +27,7 @@
 	serializable.py	\
 	settings.py 	\
 	signalgroup.py	\
+	signalinterface.py \
 	sourcelist.py 	\
 	threads.py	\
 	thumbnailer.py	\

Added: trunk/pitivi/signalinterface.py
==============================================================================
--- (empty file)
+++ trunk/pitivi/signalinterface.py	Thu Oct 16 14:19:55 2008
@@ -0,0 +1,155 @@
+# PiTiVi , Non-linear video editor
+#
+#       signalinterface.py
+#
+# Copyright (c) 2008, Edward Hervey <bilboed bilboed com>
+#
+# This program is free software; you can redistribute it and/or
+# modify it under the terms of the GNU Lesser General Public
+# License as published by the Free Software Foundation; either
+# version 2.1 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
+# Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public
+# License along with this program; if not, write to the
+# Free Software Foundation, Inc., 59 Temple Place - Suite 330,
+# Boston, MA 02111-1307, USA.
+
+# FIXME/IDEA : Add a decorator to easily add signals (ex: @signal(name="mysignal"))
+# FIXME/IDEA : Add a function to quickly define signals (a-la pygobject gsignals)
+# FIXME/IDEA : Use Weak dictionnaries for tracking connected callbacks/objects
+# FIXME/IDEA : Make specific exceptions !
+# FIXME : How to handle classes which are already using gobject (i.e. gst.Pipeline)
+
+"""
+Interfaces for event-based programming
+"""
+
+from random import randint
+
+class Signallable:
+    """
+    Signallable interface
+    """
+
+    class SignalGroup:
+        # internal
+        def __init__(self, signallable):
+            self.siglist = signallable.get_signals()
+            # self.ids is a dictionnary of
+            # key: signal name (string)
+            # value: list of:
+            #    (callback (callable),
+            #     args (list),
+            #     kwargs (dictionnary))
+            self.ids = {}
+            # self.handlers is a dictionnary of callback ids per
+            # signals.
+            self.handlers = {}
+            for signame in self.siglist.keys():
+                self.handlers[signame] = []
+
+        def connect(self, signame, cb, args, kwargs):
+            """ connect """
+            # get a unique id
+            if not signame in self.handlers.keys():
+                raise Exception("Signal %s does not exist" % signame)
+            if not callable(cb):
+                raise Exception("Provided callable '%r' is not callable" % cb)
+
+            uuid = randint(0, 2**64)
+            while uuid in self.ids:
+                uuid = randint(0, 2**64)
+
+            self.ids[uuid] = (cb, args, kwargs)
+            self.handlers[signame].append(uuid)
+            return uuid
+
+        def disconnect(self, sigid):
+            """ disconnect """
+            if not sigid in self.ids:
+                raise Exception("unknown signal id")
+            del self.ids[sigid]
+            for lists in self.handlers.itervalues():
+                if sigid in lists:
+                    lists.remove(sigid)
+
+        def emit(self, signame, *args, **kwargs):
+            """ emit """
+            # emits the signal,
+            # will concatenate the given args/kwargs with
+            # the ones supplied in .connect()
+            res = None
+            for sigid in self.handlers[signame]:
+                # cb: callable
+                cb, orar, kwar = self.ids[sigid]
+                ar = args[:] + orar
+                kw = kwargs.copy()
+                kw.update(kwar)
+                res = cb(*ar, **kw)
+            return res
+
+
+    # key : name (string)
+    # value : signature (list of any strings)
+    __signals__ = { }
+
+    def emit(self, signame, *args, **kwargs):
+        """
+        Emit the given signal.
+
+        The provided kwargs should contain *at-least* the arguments declared
+        in the signal declaration.
+
+        The object emitting the signal will be provided as the first
+        argument of the callback
+
+        Returns the first non-None return value given by the callbacks if they
+        provide any non-None return value.
+        """
+        if not hasattr(self, "_signal_group"):
+            # if there's no SignalGroup, that means nothing is
+            # connected
+            return None
+        return self._signal_group.emit(signame, self,
+                                       *args, **kwargs)
+
+    def connect(self, signame, cb, *args, **kwargs):
+        """
+        Connect a callback (with optional arguments) to the given
+        signal.
+
+        * signame : the name of the signal
+        * cb : the callback (needs to be a callable)
+        * args/kwargs : (optional) arguments
+        """
+        if not hasattr(self, "_signal_group"):
+            self._signal_group = self.SignalGroup(self)
+
+        return self._signal_group.connect(signame,
+                                           cb, args, kwargs)
+
+    def disconnect(self, sigid):
+        """
+        Disconnect signal using give signal id
+        """
+        if not hasattr(self, "_signal_group"):
+            raise Exception("This class doesn't have any signals !")
+
+        self._signal_group.disconnect(sigid)
+
+
+    @classmethod
+    def get_signals(cls):
+        """ Get the full list of signals implemented by this class """
+        sigs = {}
+        for cla in cls.mro():
+            if "__signals__" in cla.__dict__:
+                sigs.update(cla.__signals__)
+            if cla == Signallable:
+                break
+        return sigs

Modified: trunk/tests/Makefile.am
==============================================================================
--- trunk/tests/Makefile.am	(original)
+++ trunk/tests/Makefile.am	Thu Oct 16 14:19:55 2008
@@ -6,6 +6,7 @@
 	test_timeline_composition.py	\
 	test_timeline_objects.py	\
 	test_serializable.py		\
+	test_signallable.py		\
 	test_timeline_source.py		\
 	testcomplex.py
 

Added: trunk/tests/test_signallable.py
==============================================================================
--- (empty file)
+++ trunk/tests/test_signallable.py	Thu Oct 16 14:19:55 2008
@@ -0,0 +1,227 @@
+import unittest
+from pitivi.signalinterface import Signallable
+
+class myobject(object, Signallable):
+
+    __signals__ = {
+        "signal-oneargs" : ["firstarg"],
+        "signal-noargs" : []
+        }
+
+    def emit_signal_one_args(self, firstarg):
+        self.emit("signal-oneargs", firstarg)
+
+    def emit_signal_no_args(self):
+        self.emit("signal-noargs")
+
+class mysubobject(myobject):
+
+    __signals__ = {
+        "subobject-noargs" : None
+        }
+
+    def emit_sub_signal_no_args(self):
+        self.emit("subobject-noargs")
+
+class TestSignalisation(unittest.TestCase):
+    """
+    Test the proper behaviour of pitivi.signalinterface.Signallable
+    """
+
+    def setUp(self):
+        self.object = myobject()
+        self.subobject = mysubobject()
+        # internal checks to make sure these objects are clean
+        self.assertEquals(hasattr(self.object, "_signal_group"),
+                          False)
+        self.assertEquals(hasattr(self.subobject, "_signal_group"),
+                          False)
+
+        # To detect if signals were triggered
+        self.s_oneargs_triggered = 0
+        self.s_noargs_triggered = 0
+        self.s_subnoargs_triggered = 0
+
+        # values sent by signal
+        self.signal_oneargs_firstarg = None
+        self.signal_oneargs_signaller = None
+        self.signal_oneargs_args = None
+        self.signal_oneargs_kwargs = None
+
+        self.signal_noargs_signaller = None
+        self.signal_noargs_args = None
+        self.signal_noargs_kwargs = None
+
+        self.signal_subnoargs_signaller = None
+        self.signal_subnoargs_args = None
+        self.signal_subnoargs_kwargs = None
+
+    # default callbacks to be used
+    def _cb_oneargs(self, signaller, firstarg=None, *args, **kwargs):
+        self.s_oneargs_triggered +=1
+        self.signal_oneargs_signaller = signaller
+        self.signal_oneargs_firstarg = firstarg
+        self.signal_oneargs_args = args
+        self.signal_oneargs_kwargs = kwargs
+
+    def _cb_noargs(self, signaller, *args, **kwargs):
+        self.s_noargs_triggered += 1
+        self.signal_noargs_signaller = signaller
+        self.signal_noargs_args = args
+        self.signal_noargs_kwargs = kwargs
+
+    def _cb_subnoargs(self, signaller, *args, **kwargs):
+        self.s_subnoargs_triggered += 1
+        self.signal_suboargs_signaller = signaller
+        self.signal_subnoargs_args = args
+        self.signal_subnoargs_kwargs = kwargs
+
+
+    def test01_get_signals(self):
+        self.assertEquals(self.object.get_signals(),
+                          myobject.__signals__)
+        expected = myobject.__signals__
+        expected.update(mysubobject.__signals__)
+        self.assertEquals(self.subobject.get_signals(),
+                          expected)
+
+    def test02_connect(self):
+        def my_cb1(self, firstarg=None):
+            pass
+        # This should return a uuid
+        self.assert_(self.object.connect("signal-oneargs",
+                                         my_cb1))
+
+        # you can't connect to unexisting signals !
+        self.assertRaises(Exception,
+                          self.object.connect,
+                          "this-signal-doesn't-exist",
+                          my_cb1)
+
+        # you must give a callable as the cb argument
+        self.assertRaises(Exception,
+                          self.object.connect,
+                          "signal-oneargs",
+                          5)
+
+    def test03_disconnect(self):
+        def my_cb1(self, firstarg=None):
+            pass
+        sigid = self.object.connect("signal-oneargs", my_cb1)
+        self.assert_(sigid)
+        self.object.disconnect(sigid)
+
+        # disconnecting something already disconnected should
+        # trigger an exception
+        self.assertRaises(Exception,
+                          self.object.disconnect,
+                          sigid)
+
+        # disconnecting a unexisting signal should trigger
+        # an exception
+        self.assertRaises(Exception,
+                          self.object.disconnect,
+                          42)
+
+    def test04_emit01(self):
+        # signal : no arguments
+        # connect : no arguments
+        noargsid = self.object.connect("signal-noargs",
+                                       self._cb_noargs)
+        self.assert_(noargsid)
+        self.object.emit_signal_no_args()
+        self.assertEquals(self.s_noargs_triggered,1)
+        self.assertEquals(self.signal_noargs_args, ())
+        self.assertEquals(self.signal_noargs_signaller, self.object)
+        self.assertEquals(self.signal_noargs_kwargs, {})
+
+        # disconnect
+        self.object.disconnect(noargsid)
+
+        # let's make sure we're not called anymore
+        self.s_noargs_triggered = 0
+        self.object.emit_signal_no_args()
+        self.assertEquals(self.s_noargs_triggered, 0)
+
+    def test04_emit02(self):
+        # signal : no arguments
+        # connect : extra arguments
+        noargsid = self.object.connect("signal-noargs",
+                                       self._cb_noargs,
+                                       1,2,3,
+                                       myvalue=42)
+        self.assert_(noargsid)
+        self.object.emit_signal_no_args()
+        self.assertEquals(self.s_noargs_triggered, 1)
+        self.assertEquals(self.signal_noargs_args, (1,2,3))
+        self.assertEquals(self.signal_noargs_signaller, self.object)
+        self.assertEquals(self.signal_noargs_kwargs, {"myvalue":42})
+
+    def test04_emit03(self):
+        # signal : named argument
+        # connect : no arguments
+        oneargsigid = self.object.connect("signal-oneargs",
+                                            self._cb_oneargs)
+        self.assert_(oneargsigid)
+        self.object.emit_signal_one_args(firstarg="yep")
+        self.assertEquals(self.s_oneargs_triggered, 1)
+        self.assertEquals(self.signal_oneargs_signaller, self.object)
+        self.assertEquals(self.signal_oneargs_firstarg, "yep")
+        self.assertEquals(self.signal_oneargs_args, ())
+        self.assertEquals(self.signal_oneargs_kwargs, {})
+
+    def test04_emit04(self):
+        # signal : named argument
+        # connect : extra arguments
+        oneargsigid = self.object.connect("signal-oneargs",
+                                            self._cb_oneargs,
+                                          1,2,3, myvalue=42)
+        self.assert_(oneargsigid)
+        self.object.emit_signal_one_args(firstarg="yep")
+        self.assertEquals(self.s_oneargs_triggered, 1)
+        self.assertEquals(self.signal_oneargs_firstarg, "yep")
+        self.assertEquals(self.signal_oneargs_signaller, self.object)
+        self.assertEquals(self.signal_oneargs_args, (1,2,3))
+        self.assertEquals(self.signal_oneargs_kwargs, {"myvalue":42})
+
+    def test05_subclass_emit01(self):
+        # making sure a subclass can emit the parent classes
+        # signal
+        noargsid = self.subobject.connect("signal-noargs",
+                                          self._cb_noargs,
+                                          1,2,3,
+                                          myvalue=42)
+        self.assert_(noargsid)
+        self.subobject.emit_signal_no_args()
+        self.assertEquals(self.s_noargs_triggered, 1)
+        self.assertEquals(self.signal_noargs_signaller, self.subobject)
+        self.assertEquals(self.signal_noargs_args, (1,2,3))
+        self.assertEquals(self.signal_noargs_kwargs, {"myvalue":42})
+
+    def test06_multiple_emissions(self):
+        # connect two handlers to one signal
+        noargsid1 = self.object.connect("signal-noargs",
+                                        self._cb_noargs,
+                                        1,2,3,
+                                        myvalue=42)
+        self.assert_(noargsid1)
+        noargsid2 = self.object.connect("signal-noargs",
+                                        self._cb_noargs,
+                                        1,2,3,
+                                        myvalue=42)
+        self.assert_(noargsid2)
+
+        # emit the signal...
+        self.object.emit_signal_no_args()
+        # ...which should have called all the handlers
+        self.assertEquals(self.s_noargs_triggered, 2)
+        self.assertEquals(self.signal_noargs_args, (1,2,3))
+        self.assertEquals(self.signal_noargs_kwargs, {"myvalue":42})
+
+        self.object.disconnect(noargsid1)
+        self.object.disconnect(noargsid2)
+
+        self.object.emit_signal_no_args()
+        self.assertEquals(self.s_noargs_triggered, 2)
+
+    #FIXME : test return values on emission !