Python unittest et multithread

Question

J'utilise python's unittest et j'aimerais écrire un test qui démarre quelques threads et attend qu'ils se terminent. Les threads exécutent une fonction qui a quelques unittest assertions. Si l'une des assertions échoue, je souhaite que le test échoue. Cela ne semble pas être le cas.

EDIT: exemple minimal runnable (de 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 in thread Thread-1:                                          
Traceback (most recent call last):                                                  
    File "/usr/lib64/python2.7/threading.py", line 813, in __bootstrap_inner                                        
    self.run()                                                       
    File "/usr/lib64/python2.7/threading.py", line 766, in run                                           
    self.__target(*self.__args, **self.__kwargs)                                              
    File "main.py", line 7, in <lambda>                                                 
    t = threading.Thread(target=lambda: self.fail())                                             
    File "/usr/lib64/python2.7/unittest/case.py", line 450, in fail                                          
    raise self.failureException(msg)                                                 
AssertionError: None                                                     

ok                                                          

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

OK  

Answer 1

assertions Python unittest sont communiquées par des exceptions, vous devez vous assurer que les exceptions se terminent dans le fil principal. Donc, pour un fil qui signifie que vous devez exécuter .join(), comme cela jeter l'exception du fil sur dans le thread principal:

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

Assurez-vous également que vous n'avez pas try/except blocs qui pourraient manger jusqu'à l'exception avant le unittest peut les enregistrer.

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

Answer 2

Dans mon thread principal, je détecte les sous-processus qui échouent en vérifiant leur code de sortie (non-zéro est l'échec).

proc.join() 
self.assertEqual(proc.exitcode, 0, 'Sub-process failed, check output for stack trace')