test_win32events.py 文件源码

def test_addEvent(self):
        """
        When an event which has been added to the reactor is set, the action
        associated with the event is invoked in the reactor thread.
        """
        reactorThreadID = getThreadID()
        reactor = self.buildReactor()
        event = win32event.CreateEvent(None, False, False, None)
        finished = Deferred()
        finished.addCallback(lambda ignored: reactor.stop())
        listener = Listener(finished)
        reactor.addEvent(event, listener, 'occurred')
        reactor.callWhenRunning(win32event.SetEvent, event)
        self.runReactor(reactor)
        self.assertTrue(listener.success)
        self.assertEqual(reactorThreadID, listener.logThreadID)
        self.assertEqual(reactorThreadID, listener.eventThreadID)