test_threads.py 文件源码

def test_blockingCallFromThread(self):
        """
        Test blockingCallFromThread facility: create a thread, call a function
        in the reactor using L{threads.blockingCallFromThread}, and verify the
        result returned.
        """
        def reactorFunc():
            return defer.succeed("foo")
        def cb(res):
            self.assertEqual(res[0][0], "foo")

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