7 votes

Python unittest et multithreading

J'utilise le unittest de python et j'aimerais écrire un test qui démarre quelques threads et attend qu'ils se terminent. Les threads exécutent une fonction qui contient certaines assertions unittest. Si l'une des assertions échoue, je veux que le test échoue également. Ce n'est pas le cas.

ÉDITER: Exemple minimal d'exécution (python3)

import unittest
import threading

class MyTests(unittest.TestCase):

 def test_sample(self):
  t = threading.Thread(target=lambda: self.fail())
  t.start()
  t.join()

if __name__ == '__main__':
    unittest.main()

et la sortie est:

sh-4.3$ python main.py -v                                                                                                                                                                                                              
test_sample (__main__.MyTests) ... Exception dans le thread Thread-1:                                                                                                                                                                       
Traceback (most recent call last):                                                                                                                                                                                                     
  Fichier "/usr/lib64/python2.7/threading.py", ligne 813, dans __bootstrap_inner                                                                                                                                                             
    self.run()                                                                                                                                                                                                                         
  Fichier "/usr/lib64/python2.7/threading.py", ligne 766, dans run                                                                                                                                                                           
    self.__target(*self.__args, **self.__kwargs)                                                                                                                                                                                       
  Fichier "main.py", ligne 7, dans                                                                                                                                                                                                   
    t = threading.Thread(target=lambda: self.fail())                                                                                                                                                                                   
  Fichier "/usr/lib64/python2.7/unittest/case.py", ligne 450, dans fail                                                                                                                                                                      
    raise self.failureException(msg)                                                                                                                                                                                                   
AssertionError: None                                                                                                                                                                                                                   

ok                                                                                                                                                                                                                                     

----------------------------------------------------------------------                                                                                                                                                                 
Ran 1 test in 0.002s                                                                                                                                                                                                                   

OK

2voto

Grumbel Points 1296

Les assertions de unittest en Python sont communiquées par des exceptions, donc vous devez vous assurer que les exceptions se retrouvent dans le thread principal. Pour un thread, cela signifie que vous devez exécuter .join(), car cela lancera l'exception du thread dans le thread principal :

    t = threading.Thread(target=lambda: self.assertTrue(False))
    t.start()
    t.join()

Assurez-vous également de ne pas avoir de blocs try/except qui pourraient absorber l'exception avant que unittest ne puisse les enregistrer.

Éditer: self.fail() n'est en effet pas communiqué lorsqu'il est appelé depuis un thread, même si .join() est présent. Pas sûr de ce qui se passe avec ça.

1voto

Fred S Points 863

Votre test ne échoue pas pour la même raison que ce code imprime "aucune exception"

import threading

def raise_err():
    raise Exception()

try:
    t = threading.Thread(target=raise_err)
    t.start()
    t.join()
    print('aucune exception')
except:
    print('exception attrapée')

Lorsque unittest exécute votre fonction de test, il détermine si le test a réussi ou échoué en vérifiant si l'exécution du code entraîne une exception. Si l'exception se produit à l'intérieur du thread, il n'y a toujours pas d'exception dans le thread principal.

Vous pourriez faire quelque chose comme cela si vous pensez que vous DEVEZ obtenir un résultat de réussite/échec en exécutant quelque chose dans un thread. Mais ce n'est vraiment pas la manière dont unittest est conçu pour fonctionner, et il y a probablement une façon beaucoup plus facile de faire ce que vous essayez d'accomplir.

import threading
import unittest

def raise_err():
    raise Exception()
def no_err():
    return

class Runner():

    def __init__(self):
        self.threads = {}
        self.thread_results = {}

    def add(self, target, name):
        self.threads[name] = threading.Thread(target = self.run, args = [target, name])
        self.threads[name].start()

    def run(self, target, name):
        self.thread_results[name] = 'fail'
        target()
        self.thread_results[name] = 'pass'

    def check_result(self, name):
        self.threads[name].join()
        assert(self.thread_results[name] == 'pass')

runner = Runner()

class MyTests(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        runner.add(raise_err, 'test_raise_err')
        runner.add(no_err, 'test_no_err')

    def test_raise_err(self):
        runner.check_result('test_raise_err')

    def test_no_err(self):
        runner.check_result('test_no_err')

if __name__ == '__main__':
    unittest.main()

0voto

Kieren Anderson Points 314

Dans mon thread principal, je détecte les sous-processus qui échouent en vérifiant leur code de sortie (non nul signifie échec).

proc.join()
self.assertEqual(proc.exitcode, 0, 'Le sous-processus a échoué, vérifiez la sortie pour la trace de la pile')

Prograide.com

Prograide est une communauté de développeurs qui cherche à élargir la connaissance de la programmation au-delà de l'anglais.
Pour cela nous avons les plus grands doutes résolus en français et vous pouvez aussi poser vos propres questions ou résoudre celles des autres.

Powered by:

X