test_threads.py 文件源码

def test_asyncBlockingCallFromThread(self):
        """
        Test blockingCallFromThread as above, but be sure the resulting
        Deferred is not already fired.
        """
        def reactorFunc():
            d = defer.Deferred()
            reactor.callLater(0.1, d.callback, "egg")
            return d
        def cb(res):
            self.assertEqual(res[0][0], "egg")

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