<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>   &quot;&quot;&quot; <br>   decorator function catching the argument <br>   &quot;&quot;&quot; <br>   def timeout_function(func): <br>       &quot;&quot;&quot; <br>       decorator function <br>
       &quot;&quot;&quot; <br>       @wraps(func) <br>       def _timeout_function(self): <br>           &quot;&quot;&quot; <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>           &quot;&quot;&quot; <br>           def timeout_handler(signum, frame): <br>               print &#39;Timeout (%s sec) reached&#39; % 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(&quot;Tests.%s&quot; % <br>self.__class__.__name__) <br>       self.__logger.setLevel(logging.DEBUG) <br>
<p>   def setUp(self): <br>       &#39;&#39;&#39; <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>
       &#39;&#39;&#39; <br>       pass <br>
<p>   def tearDown(self): <br>       &#39;&#39;&#39; <br>       &#39;&#39;&#39; <br>
<p>   @timeout(5) <br>   def test_001(self): <br>       &#39;&#39;&#39; <br>       &#39;&#39;&#39; <br>       self.__<a href="http://logger.info">logger.info</a>(&#39;[CHECKPOINT] test_001&#39;) <br>       try: <br>           self.__<a href="http://logger.info">logger.info</a>(&#39;entering a waiting loop&#39;) <br>
           self.loop.run() <br>           self.__<a href="http://logger.info">logger.info</a>(&#39;dummy log, should not appear&#39;) <br>           self.fail() <br>
<p>       except KeyboardInterrupt: <br>           self.__logger.exception(&#39;Catching a Ctrl+c event (user or <br>timeout)&#39;) <br>       except : <br>           self.__logger.exception(&#39;Unexpected error&#39;) <br>
           self.fail() <br>
<p>   @timeout(5) <br>   def test_002(self): <br>       &#39;&#39;&#39; <br>       &#39;&#39;&#39; <br>       def loop_quit(loop): <br>           loop.quit() <br>           return False <br>
<p>       self.__<a href="http://logger.info">logger.info</a>(&#39;[CHECKPOINT] test_002&#39;) <br>       try: <br>           self.__<a href="http://logger.info">logger.info</a>(&#39;entering a waiting loop&#39;) <br>           gobject.timeout_add(1000, loop_quit, self.loop) <br>
           self.loop.run() <br>           self.__<a href="http://logger.info">logger.info</a>(&#39;exiting the loop&#39;) <br>
<p>       except KeyboardInterrupt: <br>           self.__logger.exception(&#39;Catching a Ctrl+c event (user or <br>timeout)&#39;) <br>           self.fail() <br>       except : <br>           self.__logger.exception(&#39;Unexpected error&#39;) <br>
           self.fail() <br>[/code] <br>
<p>If I start a unittest campaign like this : <br>[code] <br>if __name__ == &quot;__main__&quot;: <br>   #Add the test you want to run <br>   suite = unittest.TestSuite() <br>
<p>   #To choose a list of tests, comment those you don&#39;t want to run <br>   suite.addTest(Test_loopRun_And_Timeout(&#39;test_002&#39;)) <br>   suite.addTest(Test_loopRun_And_Timeout(&#39;test_001&#39;)) <br>   unittest.TextTestRunner(verbosity=0).run(suite) <br>
   print &#39;done&#39; <br>[/code] <br>the result give me 2 tests OK <br>
<p>Now If I launch this <br>[code] <br>if __name__ == &quot;__main__&quot;: <br>   #Add the test you want to run <br>   suite = unittest.TestSuite() <br>
<p>   #To choose a list of tests, comment those you don&#39;t want to run <br>   suite.addTest(Test_loopRun_And_Timeout(&#39;test_001&#39;)) <br>   suite.addTest(Test_loopRun_And_Timeout(&#39;test_002&#39;)) <br>   unittest.TextTestRunner(verbosity=0).run(suite) <br>
   print &#39;done&#39; <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 &#39;timeout&#39; 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">