Discussion :

[Makiavelik]

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 :

Sélectionner tout - Visualiser dans une fenêtre à part

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 :

Sélectionner tout - Visualiser dans une fenêtre à part

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

[Makiavelik]

Bonjour,

j'ai un pseudo work around qui surcharge la méthode run() de l'objet Mainloop :
on lui passe en paramètre un timeout, on lance une callback qui vérifiera qu'au bout de ce timeout la Mainloop n'est plus 'running' sinon on lève une exception de type Timeout.

Pour le moment ca fonctionne dans un module mais dans une campagne ... je vais approfondir.

Si vous avez des idées sur le problème initial je suis toujours preneur de réponses