problème de timeout avec utilisation des modules dbus & gobject

Bonjour,

J'utilise python pour réaliser des tests d'une appli qui possède une interface dbus.

Je me suis basé sur pas mal de ressources web (notamment ce tutoriel) pour commencer à coder.

Mon problème :
Dans une campagne de test unitaire si :
- il y a plus de trois testcases
- un test qui était dans un une boucle d'attente ( loop.run() ) sort via le décorateur de fonction timeout
Alors le prochain test qui appelra la loop.run() entre dans un état 'bizarre' :
- le Ctrl+c ne fonctionne pas
- le mécanisme de timeout ne fonctionne pas
- la sortie de boucle normale ne fonctionne pas
Seul un kill violent arrete le test.

Voici un exemple de code allégé mais qui reproduit bien le problème

Code:

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()

Maintenant si j'ajoute pour tester:

Code:

1
2
3
4
5
6
7
8
9
10
 
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'

OK

si :
test_001 puis test_002 => fail sur le test 002 (mais on sort proprement)
test_001, test_002, test_001, test_001 => au loop.run() dans test_002 on est en état d'attente infini sans sortie propre (sauf un kill du process)

Note : je ne mets que 2 tests, et j'ajoute plusieurs fois le même test, mais cela fonctionne pareil avec plusieurs fonctions de test différente (fait ici pour alléger le code)

le problème semble assez fourbe :calim2: et je ne vois pas l'erreur du code (je penche pour le décorateur timeout mais je ne vois pas où)

Si vous avec des pistes, je suis preneur :ccool: