decorator issue with modules dbus & gobject

[Eudes Elie]

Hi,
Here is a sample code that reproduces the issue :
[code]
import logging
import unittest
import signal
import gobject
import dbus
from functools import wraps
from dbus.mainloop.glib import DBusGMainLoop

class TimeoutException(Exception):
   pass

def timeout(timeout_time=1800):
   """
   decorator function catching the argument
   """
   def timeout_function(func):
       """
       decorator function
       """
       @wraps(func)
       def _timeout_function(self):
           """
           create a signal handler
           set the timeout with the argument given while calling the
decorator @timeout
           call the function
           catch a timeout exception if necessary
           """
           def timeout_handler(signum, frame):
               print 'Timeout (%s sec) reached' % str(timeout_time)
               raise TimeoutException()

           old_handler = signal.signal(signal.SIGALRM,
timeout_handler)
           signal.alarm(timeout_time) # triger alarm in timeout_time
seconds
           try:
               retval = func(self)
           finally:
               signal.signal(signal.SIGALRM, old_handler)
           signal.alarm(0)
           return retval
       return _timeout_function
   return timeout_function

class Test_loopRun_And_Timeout(unittest.TestCase):
   def __init__(self,*args,**kwargs):
       super(Test_loopRun_And_Timeout, self).__init__(*args,**kwargs)
       dbus_loop = DBusGMainLoop(set_as_default=True)
       self.bus = dbus.SessionBus(private=True,mainloop=dbus_loop)
       self.loop = gobject.MainLoop()

       logging.basicConfig()
       self.__logger = logging.getLogger("Tests.%s" %
self.__class__.__name__)
       self.__logger.setLevel(logging.DEBUG)

   def setUp(self):
       '''
       in this part, mediarouter can not be created
       Setup are all launch in //
       So if 50 tests are run in this class, 50 mediarouters are
created
       '''
       pass

   def tearDown(self):
       '''
       '''

   @timeout(5)
   def test_001(self):
       '''
       '''
       self.__logger.info('[CHECKPOINT] test_001')
       try:
           self.__logger.info('entering a waiting loop')
           self.loop.run()
           self.__logger.info('dummy log, should not appear')
           self.fail()

       except KeyboardInterrupt:
           self.__logger.exception('Catching a Ctrl+c event (user or
timeout)')
       except :
           self.__logger.exception('Unexpected error')
           self.fail()

   @timeout(5)
   def test_002(self):
       '''
       '''
       def loop_quit(loop):
           loop.quit()
           return False

       self.__logger.info('[CHECKPOINT] test_002')
       try:
           self.__logger.info('entering a waiting loop')
           gobject.timeout_add(1000, loop_quit, self.loop)
           self.loop.run()
           self.__logger.info('exiting the loop')

       except KeyboardInterrupt:
           self.__logger.exception('Catching a Ctrl+c event (user or
timeout)')
           self.fail()
       except :
           self.__logger.exception('Unexpected error')
           self.fail()
[/code]

If I start a unittest campaign like this :
[code]
if __name__ == "__main__":
   #Add the test you want to run
   suite = unittest.TestSuite()

   #To choose a list of tests, comment those you don't want to run
   suite.addTest(Test_loopRun_And_Timeout('test_002'))
   suite.addTest(Test_loopRun_And_Timeout('test_001'))
   unittest.TextTestRunner(verbosity=0).run(suite)
   print 'done'
[/code]
the result give me 2 tests OK

Now If I launch this
[code]
if __name__ == "__main__":
   #Add the test you want to run
   suite = unittest.TestSuite()

   #To choose a list of tests, comment those you don't want to run
   suite.addTest(Test_loopRun_And_Timeout('test_001'))
   suite.addTest(Test_loopRun_And_Timeout('test_002'))
   unittest.TextTestRunner(verbosity=0).run(suite)
   print 'done'
[/code]
1 OK (test_001)
1 Fail (test_002 goes on timeout)

And if I am using more than 3 Testcases, the code is going to run in
infinite loop (Ctrl+C or timeout decorator does not work, only a kill
works)

Is there an issue using the 'timeout' decorator with the loop.run() ?

Maybe (and certainly) I badly handle the gobject.Mainloop ?

Eudes
-------------- next part --------------
An HTML attachment was scrubbed...
URL: <http://lists.freedesktop.org/archives/dbus/attachments/20110823/b9ae56aa/attachment-0001.htm>