Python unittest and multithreading

Question

Używam Pythona unittest i chciałbym napisać test, który rozpoczyna kilka wątków i czeka na ich zakończenie. Wątki wykonują funkcję z pewnymi asercjami unittest. Jeśli któreś z twierdzeń nie powiedzie się, żałuję, że test się nie powiódł. Wydaje się, że tak nie jest.

EDIT: Minimal runnable przykład (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() 

i wyjście jest:

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

Python unittest twierdzenia są przekazywane przez wyjątkami, więc trzeba się upewnić, że wyjątki końca w głównym wątku. Tak dla wątku, który oznacza, że ​​trzeba uruchomić .join(), jako że rzuci wyjątek od wątku nad do głównego wątku:

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

również upewnić się, że nie masz żadnych try/except bloki, które może jeść wyjątek przed zarejestrowaniem ich przez unittest.

Edycja: self.fail() rzeczywiście nie jest komunikowana po wywołaniu z wątku, nawet jeśli występuje .join(). Nie jestem pewien co z tym.

Answer 2

W moim głównym wątku wykrywam podprocesy ulegające awarii, sprawdzając ich kod wyjścia (niezerowy jest uszkodzony).

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