Python unittest и многопоточность

Question

Я использую python's unittest и хотел бы написать тест, который запускает несколько потоков и ждет их завершения. Нити выполняют функцию, которая имеет некоторые утверждения unittest. Если какое-либо из утверждений терпит неудачу, я хочу, чтобы тест был неудачным. Кажется, это не так.

EDIT: Минимальный работоспособный пример (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() 

и выход:

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 утверждений доводятся исключениями, поэтому вы должны убедиться, что исключения конца в основном потоке. Таким образом, для потока, который означает, что вы должны запустить .join(), как будет бросать исключение из потока более в основной поток:

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

Также убедитесь, что у вас нет каких-либо try/except блоков, которые могут съедающие исключение до unittest может их зарегистрировать.

Edit: self.fail() действительно не сообщается, когда вызывается из потока, даже если .join() присутствует. Не уверен, что с этим.

Answer 2

В моем основном потоке я обнаруживаю, что подпроцессы терпят неудачу, проверяя их код выхода (отличное от нуля).

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