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() ?
--
http://mail.python.org/mailman/listinfo/python-list
