1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107
|
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() |
Partager