test_threads.py 文件源码

def test_callInThread(self):
        """
        Test callInThread functionality: set a C{threading.Event}, and check
        that it's not in the main thread.
        """
        def cb(ign):
            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.assertEqual(result, [False])
        return self._waitForThread().addCallback(cb)