Make Python unittest fail on exception from any thread

Question

Estoy usando el framework unittest para automatizar pruebas de integración de código python multiproceso, hardware externo y C integrado. A pesar de mi flagrante abuso de un framework de pruebas unitarias para pruebas de integración, funciona realmente bien. Excepto por un problema: necesito que la prueba falle si se produce una excepción de cualquiera de los hilos engendrados. ¿Es esto posible con el framework unittest?

Una solución simple pero no viable sería a) a refactorizar el código para evitar el multi-threading o b) probar cada thread por separado. No puedo hacer eso porque el código interactúa asincrónicamente con el hardware externo. También consideré implementar algún tipo de mensaje que pase para reenviar las excepciones al hilo principal de la unidad. Esto requeriría cambios significativos relacionados con las pruebas del código que se está probando, y quiero evitar eso.

La hora de un ejemplo. ¿Puedo modificar el script de prueba a continuación para que falle en la excepción planteada en my_thread sin modificando la clase 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

Al principio, sys.excepthook parecía una solución. Es un gancho global que se invoca cada vez que se lanza una excepción no detectada.

Lamentablemente, esto no funciona. ¿Por qué? así threading envuelve su función run en el código que imprime los encantadores tracebacks que ve en la pantalla (se dio cuenta de que siempre le dice Exception in thread {Name of your thread here}? así es como se hace).

Para resumir, parece que el threading nódulo tiene una importación indocumentado, que hace algo a lo largo de las líneas de:

threading._format_exc = traceback.format_exc 

No es muy sorprendente que esta función sólo se llama cuando se produce una excepción de un hilo de run función.

Entonces, ¿qué hacemos? Reemplazar esta función con nuestra lógica, y 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) 

Uso:

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

usted todavía tiene que join sí mismo, pero al salir, gestor de contexto del con-declaración comprobará si hay alguna excepción lanzado en otros hilos, y levantará una excepción apropiadamente.

---

LA código se proporciona "tal cual", sin garantía de ningún tipo, expresa o implícita

Este es un truco indocumentado, tipo-de-horrible. Lo probé en Linux y Windows, y parece funcionar. Úselo bajo su propio riesgo.

Answer 2

me he encontrado con este problema a mí mismo, y la única solución que he sido capaz de llegar a la subclasificación se rosca para incluir un atributo de si es o no termina sin una excepción no capturada:

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))