FM / test

/usr/lib/python3/dist-packages/twisted/internet/test/

/usr/lib/python3/dist-packages/twisted/internet/test

Name	Size	Mode	Actions
fake_CAs/	-	0755	rm
__pycache__/	-	0755	rm
connectionmixins.py	20184	0644	edit dl rm
fakeendpoint.py	1663	0644	edit dl rm
modulehelpers.py	1672	0644	edit dl rm
process_cli.py	547	0644	edit dl rm
process_connectionlost.py	126	0644	edit dl rm
process_gireactornocompat.py	816	0644	edit dl rm
process_helper.py	1282	0644	edit dl rm
reactormixins.py	14609	0644	edit dl rm
test_abstract.py	1976	0644	edit dl rm
test_address.py	8266	0644	edit dl rm
test_asyncioreactor.py	9327	0644	edit dl rm
test_base.py	13684	0644	edit dl rm
test_baseprocess.py	2588	0644	edit dl rm
test_core.py	11125	0644	edit dl rm
test_default.py	3419	0644	edit dl rm
test_defer_await.py	6514	0644	edit dl rm
test_defer_yieldfrom.py	4848	0644	edit dl rm
test_endpoints.py	148720	0644	edit dl rm
test_epollreactor.py	7322	0644	edit dl rm
test_error.py	1090	0644	edit dl rm
test_fdset.py	13497	0644	edit dl rm
test_filedescriptor.py	2761	0644	edit dl rm
test_gireactor.py	6516	0644	edit dl rm
test_glibbase.py	2217	0644	edit dl rm
test_inlinecb.py	11482	0644	edit dl rm
test_inotify.py	18562	0644	edit dl rm
test_iocp.py	7700	0644	edit dl rm
test_kqueuereactor.py	1853	0644	edit dl rm
test_main.py	1331	0644	edit dl rm
test_newtls.py	6532	0644	edit dl rm
test_pollingfile.py	1297	0644	edit dl rm
test_posixbase.py	9718	0644	edit dl rm
test_posixprocess.py	11203	0644	edit dl rm
test_process.py	36732	0644	edit dl rm
test_protocol.py	18520	0644	edit dl rm
test_resolver.py	19610	0644	edit dl rm
test_serialport.py	1988	0644	edit dl rm
test_sigchld.py	3922	0644	edit dl rm
test_socket.py	9443	0644	edit dl rm
test_stdio.py	6597	0644	edit dl rm
test_tcp.py	106621	0644	edit dl rm
test_testing.py	16239	0644	edit dl rm
test_threads.py	8734	0644	edit dl rm
test_time.py	3758	0644	edit dl rm
test_tls.py	13349	0644	edit dl rm
test_udp.py	16881	0644	edit dl rm
test_udp_internals.py	4837	0644	edit dl rm
test_unix.py	34854	0644	edit dl rm
test_win32events.py	6473	0644	edit dl rm
test_win32serialport.py	5330	0644	edit dl rm
_posixifaces.py	4391	0644	edit dl rm
_win32ifaces.py	3975	0644	edit dl rm
__init__.py	112	0644	edit dl rm

Edit: /usr/lib/python3/dist-packages/twisted/internet/test/test_win32events.py (6473B)

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

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

try:
    import win32event  # type: ignore[import]
except ImportError:
    win32event = None

from zope.interface.verify import verifyObject

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

class Listener:
    """
    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 = getThreadID()
        return "Listener"

def occurred(self):
        self.success = True
        self.eventThreadID = getThreadID()
        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 = getThreadID()
        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 = 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)

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 = getThreadID()
        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 = getThreadID()
        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.assertIsNone(listener.connLostThreadID)

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

cmd: