# Copyright (c) Twisted Matrix Laboratories.
# See LICENSE for details.

"""
Tests for implementations of L{IReactorWin32Events}.
"""

from thread import get_ident

try:
import win32event
except ImportError:
win32event = None

from zope.interface.verify import verifyObject

from twisted.python.failure import Failure
from twisted.python.threadable import isInIOThread
from twisted.internet.interfaces import IReactorWin32Events
from twisted.internet.defer import Deferred
from twisted.internet.test.reactormixins import ReactorBuilder


class Listener(object):
"""
L{Listener} is an object that can be added to a L{IReactorWin32Events}
reactor to receive callback notification when a Windows event is set. It
records what thread its callback is invoked in and fires a Deferred.

@ivar success: A flag which is set to C{True} when the event callback is
called.

@ivar logThreadID: The id of the thread in which the C{logPrefix} method is
called.

@ivar eventThreadID: The id of the thread in which the event callback is
called.

@ivar connLostThreadID: The id of the thread in which the C{connectionLost}
method is called.

@ivar _finished: The L{Deferred} which will be fired when the event callback
is called.
"""
success = False
logThreadID = eventThreadID = connLostThreadID = None

def __init__(self, finished):
self._finished = finished


def logPrefix(self):
self.logThreadID = get_ident()
return 'Listener'


def occurred(self):
self.success = True
self.eventThreadID = get_ident()
self._finished.callback(None)


def brokenOccurred(self):
raise RuntimeError("Some problem")


def returnValueOccurred(self):
return EnvironmentError("Entirely different problem")


def connectionLost(self, reason):
self.connLostThreadID = get_ident()
self._finished.errback(reason)



class Win32EventsTestsBuilder(ReactorBuilder):
"""
Builder defining tests relating to L{IReactorWin32Events}.
"""
requiredInterfaces = [IReactorWin32Events]

def test_interface(self):
"""
An instance of the reactor has all of the methods defined on
L{IReactorWin32Events}.
"""
reactor = self.buildReactor()
verifyObject(IReactorWin32Events, reactor)


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 = get_ident()
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)


def test_ioThreadDoesNotChange(self):
"""
Using L{IReactorWin32Events.addEvent} does not change which thread is
reported as the I/O thread.
"""
results = []
def check(ignored):
results.append(isInIOThread())
reactor.stop()
reactor = self.buildReactor()
event = win32event.CreateEvent(None, False, False, None)
finished = Deferred()
listener = Listener(finished)
finished.addCallback(check)
reactor.addEvent(event, listener, 'occurred')
reactor.callWhenRunning(win32event.SetEvent, event)
self.runReactor(reactor)
self.assertTrue(listener.success)
self.assertEqual([True], results)


def test_disconnectedOnError(self):
"""
If the event handler raises an exception, the event is removed from the
reactor and the handler's C{connectionLost} method is called in the I/O
thread and the exception is logged.
"""
reactorThreadID = get_ident()
reactor = self.buildReactor()
event = win32event.CreateEvent(None, False, False, None)

result = []
finished = Deferred()
finished.addBoth(result.append)
finished.addBoth(lambda ignored: reactor.stop())

listener = Listener(finished)
reactor.addEvent(event, listener, 'brokenOccurred')
reactor.callWhenRunning(win32event.SetEvent, event)
self.runReactor(reactor)

self.assertIsInstance(result[0], Failure)
result[0].trap(RuntimeError)

self.assertEqual(reactorThreadID, listener.connLostThreadID)
self.assertEqual(1, len(self.flushLoggedErrors(RuntimeError)))


def test_disconnectOnReturnValue(self):
"""
If the event handler returns a value, the event is removed from the
reactor and the handler's C{connectionLost} method is called in the I/O
thread.
"""
reactorThreadID = get_ident()
reactor = self.buildReactor()
event = win32event.CreateEvent(None, False, False, None)

result = []
finished = Deferred()
finished.addBoth(result.append)
finished.addBoth(lambda ignored: reactor.stop())

listener = Listener(finished)
reactor.addEvent(event, listener, 'returnValueOccurred')
reactor.callWhenRunning(win32event.SetEvent, event)
self.runReactor(reactor)

self.assertIsInstance(result[0], Failure)
result[0].trap(EnvironmentError)

self.assertEqual(reactorThreadID, listener.connLostThreadID)


def test_notDisconnectedOnShutdown(self):
"""
Event handlers added with L{IReactorWin32Events.addEvent} do not have
C{connectionLost} called on them if they are still active when the
reactor shuts down.
"""
reactor = self.buildReactor()
event = win32event.CreateEvent(None, False, False, None)
finished = Deferred()
listener = Listener(finished)
reactor.addEvent(event, listener, 'occurred')
reactor.callWhenRunning(reactor.stop)
self.runReactor(reactor)
self.assertIs(None, listener.connLostThreadID)

globals().update(Win32EventsTestsBuilder.makeTestCaseClasses())