<div id="inbdy"><a name="msg_c7dcfb46e4ddd581"></a><font class="fixed_width" face="Courier, Monospaced">Hi, <br>Here is a sample code that reproduces the issue : <br>[code] <br>import logging <br>import unittest <br>import signal <br>
import gobject <br>import dbus <br>from functools import wraps <br>from dbus.mainloop.glib import DBusGMainLoop <br>
<p>class TimeoutException(Exception): <br> pass <br>
<p>def timeout(timeout_time=1800): <br> """ <br> decorator function catching the argument <br> """ <br> def timeout_function(func): <br> """ <br> decorator function <br>
""" <br> @wraps(func) <br> def _timeout_function(self): <br> """ <br> create a signal handler <br> set the timeout with the argument given while calling the <br>
decorator @timeout <br> call the function <br> catch a timeout exception if necessary <br> """ <br> def timeout_handler(signum, frame): <br> print 'Timeout (%s sec) reached' % str(timeout_time) <br>
raise TimeoutException() <br>
<p> old_handler = signal.signal(signal.SIGALRM, <br>timeout_handler) <br> signal.alarm(timeout_time) # triger alarm in timeout_time <br>seconds <br> try: <br> retval = func(self) <br>
finally: <br> signal.signal(signal.SIGALRM, old_handler) <br> signal.alarm(0) <br> return retval <br> return _timeout_function <br> return timeout_function <br>
<p>class Test_loopRun_And_Timeout(unittest.TestCase): <br> def __init__(self,*args,**kwargs): <br> super(Test_loopRun_And_Timeout, self).__init__(*args,**kwargs) <br> dbus_loop = DBusGMainLoop(set_as_default=True) <br>
self.bus = dbus.SessionBus(private=True,mainloop=dbus_loop) <br> self.loop = gobject.MainLoop() <br>
<p> logging.basicConfig() <br> self.__logger = logging.getLogger("Tests.%s" % <br>self.__class__.__name__) <br> self.__logger.setLevel(logging.DEBUG) <br>
<p> def setUp(self): <br> ''' <br> in this part, mediarouter can not be created <br> Setup are all launch in // <br> So if 50 tests are run in this class, 50 mediarouters are <br>created <br>
''' <br> pass <br>
<p> def tearDown(self): <br> ''' <br> ''' <br>
<p> @timeout(5) <br> def test_001(self): <br> ''' <br> ''' <br> self.__<a href="http://logger.info">logger.info</a>('[CHECKPOINT] test_001') <br> try: <br> self.__<a href="http://logger.info">logger.info</a>('entering a waiting loop') <br>
self.loop.run() <br> self.__<a href="http://logger.info">logger.info</a>('dummy log, should not appear') <br> self.fail() <br>
<p> except KeyboardInterrupt: <br> self.__logger.exception('Catching a Ctrl+c event (user or <br>timeout)') <br> except : <br> self.__logger.exception('Unexpected error') <br>
self.fail() <br>
<p> @timeout(5) <br> def test_002(self): <br> ''' <br> ''' <br> def loop_quit(loop): <br> loop.quit() <br> return False <br>
<p> self.__<a href="http://logger.info">logger.info</a>('[CHECKPOINT] test_002') <br> try: <br> self.__<a href="http://logger.info">logger.info</a>('entering a waiting loop') <br> gobject.timeout_add(1000, loop_quit, self.loop) <br>
self.loop.run() <br> self.__<a href="http://logger.info">logger.info</a>('exiting the loop') <br>
<p> except KeyboardInterrupt: <br> self.__logger.exception('Catching a Ctrl+c event (user or <br>timeout)') <br> self.fail() <br> except : <br> self.__logger.exception('Unexpected error') <br>
self.fail() <br>[/code] <br>
<p>If I start a unittest campaign like this : <br>[code] <br>if __name__ == "__main__": <br> #Add the test you want to run <br> suite = unittest.TestSuite() <br>
<p> #To choose a list of tests, comment those you don't want to run <br> suite.addTest(Test_loopRun_And_Timeout('test_002')) <br> suite.addTest(Test_loopRun_And_Timeout('test_001')) <br> unittest.TextTestRunner(verbosity=0).run(suite) <br>
print 'done' <br>[/code] <br>the result give me 2 tests OK <br>
<p>Now If I launch this <br>[code] <br>if __name__ == "__main__": <br> #Add the test you want to run <br> suite = unittest.TestSuite() <br>
<p> #To choose a list of tests, comment those you don't want to run <br> suite.addTest(Test_loopRun_And_Timeout('test_001')) <br> suite.addTest(Test_loopRun_And_Timeout('test_002')) <br> unittest.TextTestRunner(verbosity=0).run(suite) <br>
print 'done' <br>[/code] <br>1 OK (test_001) <br>1 Fail (test_002 goes on timeout) <br>
<p>And if I am using more than 3 Testcases, the code is going to run in <br>infinite loop (Ctrl+C or timeout decorator does not work, only a kill <br>works) <br>
<p>Is there an issue using the 'timeout' decorator with the loop.run() ? </p>
<p>Maybe (and certainly) I badly handle the gobject.Mainloop ?</p>
<p>Eudes<br></p></p></p></p></p></p></p></p></p></p></p></p></p></p></p></p></p></p></font></div><br style="FONT-SIZE: 8px" clear="all">