test_threads.py 文件源码

def testCallInThread(self):
        waiter = threading.Event()
        result = []
        def threadedFunc():
            result.append(threadable.isInIOThread())
            waiter.set()

        reactor.callInThread(threadedFunc)
        waiter.wait(120)
        if not waiter.isSet():
            self.fail("Timed out waiting for event.")
        else:
            self.assertEquals(result, [False])