test_threads.py 文件源码

def test_deferredFailure(self):
        """
        Check that L{threads.deferToThreadPool} return a failure object with an
        appropriate exception instance when the called function raises an
        exception.
        """
        class NewError(Exception):
            pass
        def raiseError():
            raise NewError()
        d = threads.deferToThreadPool(reactor, self.tp, raiseError)
        return self.assertFailure(d, NewError)