모든 스레드 예외로 인해 Python unittest가 실패하게 만들기

Question

unittest 프레임 워크를 사용하여 멀티 스레드 파이썬 코드, 외부 하드웨어 및 내장 된 C의 통합 테스트를 자동화하고 있습니다. 통합 테스팅을위한 unittesting 프레임 워크의 맹렬한 남용에도 불구하고, 잘. 하나의 문제를 제외하고는 : 스폰 된 스레드에서 예외가 발생하면 테스트가 실패해야합니다. 이것은 unittest 프레임 워크에서 가능합니까?

단순하지만 실용적이지 않은 솔루션은 a) 멀티 스레딩을 피하기 위해 코드를 리팩터링하거나 b) 각 스레드를 개별적으로 테스트하는 것입니다. 코드가 외부 하드웨어와 비동기 적으로 상호 작용하기 때문에 그렇게 할 수 없습니다. 또한 예외를 메인 unittest 스레드로 전달하기 위해 어떤 종류의 메시지 전달을 구현하는 것을 고려했습니다. 이를 위해서는 테스트중인 코드에 상당한 테스트 관련 변경이 필요하므로이를 피하고 싶습니다.

예를 들면 시간입니다. 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

처음에는 sys.excepthook가 솔루션처럼 보입니다. 캐치되지 않는 예외가 발생할 때마다 호출되는 전역 후크입니다.

불행히도, 이것은 작동하지 않습니다. 왜? 잘 threading은 run 함수를 화면에 표시되는 멋진 추적 코드를 인쇄하는 코드로 래핑합니다 (항상 Exception in thread {Name of your thread here}을 알려주는 방법을 알아 차 렸습니다).

길고도 짧은 이야기, threading 결절이의 라인을 따라 뭔가하지 문서화되지 않은 수입이 나타납니다 :

threading._format_exc = traceback.format_exc 

을하지 매우 놀랍게도, 예외가 스레드의에서 던진 때이 기능 만이라고합니다 run 기능.

그럼 어떻게해야합니까? 이 우리의 논리와 기능, 함으로써 해결할 수 교체 :

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) 

사용법 :

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

당신은 여전히 ​​join 자신이 필요하지만, 종료시에,로 문의 컨텍스트 매니저가 예외를 확인합니다 다른 스레드에서 throw되며 적절하게 예외를 발생시킵니다.

---

코드는 어떠한 보증 EXPRESS없이 "그대로"제공 OR 이것이 불법 정렬 오브 끔찍한 해킹

암시한다. 내가 리눅스와 윈도우에서 그것을 테스트하고, 그것은 작동하는 것 같습니다. 자신의 책임하에 사용하십시오.

Answer 2

나는이 문제를 통해 자신을 왔어요, 그리고 유일한 해결책은 나는 캐치되지 않는 예외없이 종료 여부에 대한 속성을 포함하는 스레드를 서브 클래스된다 마련 할 수있었습니다 :

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