test_threads.py 文件源码

def test_asyncErrorBlockingCallFromThread(self):
        """
        Test error report for blockingCallFromThread as above, but be sure the
        resulting Deferred is not already fired.
        """
        def reactorFunc():
            d = defer.Deferred()
            reactor.callLater(0.1, d.errback, RuntimeError("spam"))
            return d
        def cb(res):
            self.assertIsInstance(res[1][0], RuntimeError)
            self.assertEqual(res[1][0].args[0], "spam")

        return self._testBlockingCallFromThread(reactorFunc).addCallback(cb)