Rendi Python unittest fail su eccezione da qualsiasi thread

Question

Sto utilizzando il framework unittest per automatizzare i test di integrazione di codice python multi-thread, hardware esterno e C. incorporato Nonostante il mio palese abuso di un framework per test di integrazione non funziona, funziona davvero bene. Tranne un problema: ho bisogno che il test fallisca se viene sollevata un'eccezione da uno qualsiasi dei thread spawn. Questo è possibile con la struttura più semplice?

Una soluzione semplice ma non lavorabile sarebbe a) refactoring del codice per evitare il multi-threading o b) testare ciascun thread separatamente. Non posso farlo perché il codice interagisce in modo asincrono con l'hardware esterno. Ho anche preso in considerazione l'implementazione di un tipo di messaggio che passa per inoltrare le eccezioni al thread di unittest principale. Ciò richiederebbe modifiche significative del test al codice in fase di test, e voglio evitarlo.

Tempo per un esempio. Posso modificare lo script di test qui sotto per fallire nell'eccezione sollevata in my_thread senza modificando la classe x.ExceptionRaiser?

import unittest 
import x 

class Test(unittest.TestCase): 
    def test_x(self): 
     my_thread = x.ExceptionRaiser() 
     # Test case should fail when thread is started and raises 
     # an exception. 
     my_thread.start() 
     my_thread.join() 

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

Answer 1

In un primo momento, sys.excepthook sembrava una soluzione. È un hook globale che viene chiamato ogni volta che viene lanciata un'eccezione non rilevata.

Sfortunatamente, questo non funziona. Perché? bene threading avvolge la tua funzione run nel codice che stampa i deliziosi traceback che vedi sullo schermo (notato come ti dice sempre Exception in thread {Name of your thread here}? questo è come è fatto).

Per farla breve, sembra che il nodulo threading ha un'importazione non documentata che fa qualcosa sulla falsariga di:

threading._format_exc = traceback.format_exc 

Non molto sorprendente, questa funzione viene chiamata solo quando viene generata un'eccezione da un filo d' run funzione.

Quindi cosa facciamo? Sostituire questa funzione con la nostra logica, e voilà:

import threading 
import os 

class GlobalExceptionWatcher(object): 
    def _store_excepthook(self): 
     ''' 
     Uses as an exception handlers which stores any uncaught exceptions. 
     ''' 
     formated_exc = self.__org_hook() 
     self._exceptions.append(formated_exc) 
     return formated_exc 

    def __enter__(self): 
     ''' 
     Register us to the hook. 
     ''' 
     self._exceptions = [] 
     self.__org_hook = threading._format_exc 
     threading._format_exc = self._store_excepthook 

    def __exit__(self, type, value, traceback): 
     ''' 
     Remove us from the hook, assure no exception were thrown. 
     ''' 
     threading._format_exc = self.__org_hook 
     if len(self._exceptions) != 0: 
      tracebacks = os.linesep.join(self._exceptions) 
      raise Exception('Exceptions in other threads: %s' % tracebacks) 

Usage:

my_thread = x.ExceptionRaiser() 
# will fail when thread is started and raises an exception. 
with GlobalExceptionWatcher(): 
    my_thread.start() 
    my_thread.join() 

Hai ancora bisogno di join da soli, ma in caso di uscita, responsabile contestuale del con-dichiarazione verifica la presenza di alcuna eccezione gettato in altri thread e solleverà un'eccezione in modo appropriato.

---

IL CODICE VIENE FORNITO "COSÌ COM'È", SENZA ALCUN TIPO DI GARANZIA, ESPLICITA O IMPLICITA

Questa è una non documentata, mod sorta-di-orribile. L'ho provato su linux e windows, e sembra funzionare. Usalo a tuo rischio.

Answer 2

ho incontrato questo problema me stesso, e l'unica soluzione che ho potuto venire in mente è sottoclasse discussione per includere un attributo per se o non termina senza un'eccezione non rilevata:

from threading import Thread 

class ErrThread(Thread): 
    """                                                
    A subclass of Thread that will log store exceptions if the thread does                                
    not exit normally                                             
    """ 
    def run(self): 
     try: 
      Thread.run(self) 
     except Exception as self.err: 
      pass 
     else: 
      self.err = None 


class TaskQueue(object): 
    """                                                
    A utility class to run ErrThread objects in parallel and raises and exception                              
    in the event that *any* of them fail.                                        
    """ 

    def __init__(self, *tasks): 

     self.threads = [] 

     for t in tasks: 
      try: 
       self.threads.append(ErrThread(**t)) ## passing in a dict of target and args 
      except TypeError: 
       self.threads.append(ErrThread(target=t)) 

    def run(self): 

     for t in self.threads: 
      t.start() 
     for t in self.threads: 
      t.join() 
      if t.err: 
       raise Exception('Thread %s failed with error: %s' % (t.name, t.err))