Make Python Unittest fehlgeschlagen auf Ausnahme von jedem Thread

Question

Ich verwende das Unittest-Framework, um Integrationstests von Multi-Threaded Python-Code, externe Hardware und Embedded C zu automatisieren. Trotz meiner eklatanten Missbrauch eines Unittesting-Framework für Integrationstests, funktioniert es wirklich Gut. Bis auf ein Problem: Ich muss den Test fehlschlagen, wenn eine Ausnahme von einem der erzeugten Threads ausgelöst wird. Ist das mit dem Unittest-Framework möglich?

Eine einfache, aber nicht praktikable Lösung wäre, entweder den Code umzuformen, um Multithreading zu vermeiden, oder b) jeden Thread einzeln zu testen. Ich kann das nicht, weil der Code asynchron mit der externen Hardware interagiert. Ich habe auch überlegt, eine Art von Nachrichtenübergabe zu implementieren, um die Ausnahmen an den Haupt-Unittest-Thread weiterzuleiten. Dies würde erhebliche testbezogene Änderungen an dem getesteten Code erfordern, und das möchte ich vermeiden.

Zeit für ein Beispiel. Kann ich das folgende Testskript so ändern, dass es bei der Exception in my_thread ohne, die die x.ExceptionRaiser-Klasse ändert, fehlschlägt?

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

Zuerst sah sys.excepthook wie eine Lösung aus. Es ist ein globaler Hook, der jedes Mal aufgerufen wird, wenn eine nicht abgefangene Ausnahme ausgelöst wird.

Leider funktioniert das nicht. Warum? Nun threading Wraps Ihre run Funktion in Code, der die schönen Tracebacks gedruckt, die Sie auf dem Bildschirm sehen (gemerkt, wie es Ihnen immer sagt Exception in thread {Name of your thread here}? So wird es gemacht).

Lange Rede kurzer Sinn, scheint es, dass die threading Knötchen eine nicht dokumentierte Import hat, die von etwas entlang der Linien tut:

threading._format_exc = traceback.format_exc 

Nicht sehr überraschend, dass diese Funktion nur dann aufgerufen, wenn eine Ausnahme von einem Thread ausgelöst wird, run Funktion.

Was machen wir also? Ersetzen Sie diese Funktion mit unserer Logik und 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) 

Verbrauch:

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

Sie müssen noch zu join selbst, aber beim Austritt, der mit-Erklärung des Kontext-Manager prüft für jede Ausnahme in andere Threads geworfen werden und eine Ausnahme angemessen auslösen.

---

DER CODE "AS IS" ZUR VERFÜGUNG GESTELLT, ohne Gewährleistung irgendeiner Art, AUSDRÜCKLICH ODER IMPLIZIT

Dies ist eine nicht dokumentierte, sort-of-schreckliche Hack. Ich habe es auf Linux und Windows getestet, und es scheint zu funktionieren. Verwenden Sie es auf eigene Gefahr.

Answer 2

Ich habe über dieses Problem selbst kommen, und die einzige Lösung, die ich habe in der Lage zu kommen mit ist Subklassen Thema ein Attribut enthalten, ob es ohne eine abgefangene Ausnahme beendet:

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