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_core.py (11125B)

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

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

import signal
import time
from types import FrameType
from typing import Callable, List, Optional, Tuple, Union, cast

from twisted.internet.abstract import FileDescriptor
from twisted.internet.defer import Deferred
from twisted.internet.error import ReactorAlreadyRunning, ReactorNotRestartable
from twisted.internet.test.reactormixins import ReactorBuilder
from twisted.python.failure import Failure
from twisted.trial.unittest import SynchronousTestCase

class SystemEventTestsBuilder(ReactorBuilder):
    """
    Builder defining tests relating to L{IReactorCore.addSystemEventTrigger}
    and L{IReactorCore.fireSystemEvent}.
    """

def test_stopWhenNotStarted(self) -> None:
        """
        C{reactor.stop()} raises L{RuntimeError} when called when the reactor
        has not been started.
        """
        reactor = self.buildReactor()
        cast(SynchronousTestCase, self).assertRaises(RuntimeError, reactor.stop)

def test_stopWhenAlreadyStopped(self) -> None:
        """
        C{reactor.stop()} raises L{RuntimeError} when called after the reactor
        has been stopped.
        """
        reactor = self.buildReactor()
        reactor.callWhenRunning(reactor.stop)
        self.runReactor(reactor)
        cast(SynchronousTestCase, self).assertRaises(RuntimeError, reactor.stop)

def test_callWhenRunningOrder(self) -> None:
        """
        Functions are run in the order that they were passed to
        L{reactor.callWhenRunning}.
        """
        reactor = self.buildReactor()
        events: List[str] = []
        reactor.callWhenRunning(events.append, "first")
        reactor.callWhenRunning(events.append, "second")
        reactor.callWhenRunning(reactor.stop)
        self.runReactor(reactor)
        cast(SynchronousTestCase, self).assertEqual(events, ["first", "second"])

def test_runningForStartupEvents(self) -> None:
        """
        The reactor is not running when C{"before"} C{"startup"} triggers are
        called and is running when C{"during"} and C{"after"} C{"startup"}
        triggers are called.
        """
        reactor = self.buildReactor()
        state = {}

def beforeStartup() -> None:
            state["before"] = reactor.running

def duringStartup() -> None:
            state["during"] = reactor.running

def afterStartup() -> None:
            state["after"] = reactor.running

testCase = cast(SynchronousTestCase, self)

reactor.addSystemEventTrigger("before", "startup", beforeStartup)
        reactor.addSystemEventTrigger("during", "startup", duringStartup)
        reactor.addSystemEventTrigger("after", "startup", afterStartup)
        reactor.callWhenRunning(reactor.stop)
        testCase.assertEqual(state, {})
        self.runReactor(reactor)
        testCase.assertEqual(state, {"before": False, "during": True, "after": True})

def test_signalHandlersInstalledDuringStartup(self) -> None:
        """
        Signal handlers are installed in responsed to the C{"during"}
        C{"startup"}.
        """
        reactor = self.buildReactor()
        phase: Optional[str] = None

def beforeStartup() -> None:
            nonlocal phase
            phase = "before"

def afterStartup() -> None:
            nonlocal phase
            phase = "after"

reactor.addSystemEventTrigger("before", "startup", beforeStartup)
        reactor.addSystemEventTrigger("after", "startup", afterStartup)

sawPhase = []

def fakeSignal(signum: int, action: Callable[[int, FrameType], None]) -> None:
            sawPhase.append(phase)

testCase = cast(SynchronousTestCase, self)

testCase.patch(signal, "signal", fakeSignal)
        reactor.callWhenRunning(reactor.stop)
        testCase.assertIsNone(phase)
        testCase.assertEqual(sawPhase, [])
        self.runReactor(reactor)
        testCase.assertIn("before", sawPhase)
        testCase.assertEqual(phase, "after")

def test_stopShutDownEvents(self) -> None:
        """
        C{reactor.stop()} fires all three phases of shutdown event triggers
        before it makes C{reactor.run()} return.
        """
        reactor = self.buildReactor()
        events = []
        reactor.addSystemEventTrigger(
            "before", "shutdown", lambda: events.append(("before", "shutdown"))
        )
        reactor.addSystemEventTrigger(
            "during", "shutdown", lambda: events.append(("during", "shutdown"))
        )
        reactor.addSystemEventTrigger(
            "after", "shutdown", lambda: events.append(("after", "shutdown"))
        )
        reactor.callWhenRunning(reactor.stop)
        self.runReactor(reactor)
        cast(SynchronousTestCase, self).assertEqual(
            events,
            [("before", "shutdown"), ("during", "shutdown"), ("after", "shutdown")],
        )

def test_shutdownFiresTriggersAsynchronously(self) -> None:
        """
        C{"before"} C{"shutdown"} triggers are not run synchronously from
        L{reactor.stop}.
        """
        reactor = self.buildReactor()
        events: List[str] = []
        reactor.addSystemEventTrigger(
            "before", "shutdown", events.append, "before shutdown"
        )

def stopIt() -> None:
            reactor.stop()
            events.append("stopped")

testCase = cast(SynchronousTestCase, self)

reactor.callWhenRunning(stopIt)
        testCase.assertEqual(events, [])
        self.runReactor(reactor)
        testCase.assertEqual(events, ["stopped", "before shutdown"])

def test_shutdownDisconnectsCleanly(self) -> None:
        """
        A L{IFileDescriptor.connectionLost} implementation which raises an
        exception does not prevent the remaining L{IFileDescriptor}s from
        having their C{connectionLost} method called.
        """
        lostOK = [False]

# Subclass FileDescriptor to get logPrefix
        class ProblematicFileDescriptor(FileDescriptor):
            def connectionLost(self, reason: Failure) -> None:
                raise RuntimeError("simulated connectionLost error")

class OKFileDescriptor(FileDescriptor):
            def connectionLost(self, reason: Failure) -> None:
                lostOK[0] = True

testCase = cast(SynchronousTestCase, self)
        reactor = self.buildReactor()

# Unfortunately, it is necessary to patch removeAll to directly control
        # the order of the returned values.  The test is only valid if
        # ProblematicFileDescriptor comes first.  Also, return these
        # descriptors only the first time removeAll is called so that if it is
        # called again the file descriptors aren't re-disconnected.
        fds = iter([ProblematicFileDescriptor(), OKFileDescriptor()])
        reactor.removeAll = lambda: fds
        reactor.callWhenRunning(reactor.stop)
        self.runReactor(reactor)
        testCase.assertEqual(len(testCase.flushLoggedErrors(RuntimeError)), 1)
        testCase.assertTrue(lostOK[0])

def test_multipleRun(self) -> None:
        """
        C{reactor.run()} raises L{ReactorAlreadyRunning} when called when
        the reactor is already running.
        """
        events: List[str] = []

testCase = cast(SynchronousTestCase, self)

def reentrantRun() -> None:
            testCase.assertRaises(ReactorAlreadyRunning, reactor.run)
            events.append("tested")

reactor = self.buildReactor()
        reactor.callWhenRunning(reentrantRun)
        reactor.callWhenRunning(reactor.stop)
        self.runReactor(reactor)
        testCase.assertEqual(events, ["tested"])

def test_runWithAsynchronousBeforeStartupTrigger(self) -> None:
        """
        When there is a C{'before'} C{'startup'} trigger which returns an
        unfired L{Deferred}, C{reactor.run()} starts the reactor and does not
        return until after C{reactor.stop()} is called
        """
        events = []

def trigger() -> Deferred[object]:
            events.append("trigger")
            d: Deferred[object] = Deferred()
            d.addCallback(callback)
            reactor.callLater(0, d.callback, None)
            return d

def callback(ignored: object) -> None:
            events.append("callback")
            reactor.stop()

reactor = self.buildReactor()
        reactor.addSystemEventTrigger("before", "startup", trigger)
        self.runReactor(reactor)
        cast(SynchronousTestCase, self).assertEqual(events, ["trigger", "callback"])

def test_iterate(self) -> None:
        """
        C{reactor.iterate()} does not block.
        """
        reactor = self.buildReactor()
        t = reactor.callLater(5, reactor.crash)

start = time.time()
        reactor.iterate(0)  # Shouldn't block
        elapsed = time.time() - start

cast(SynchronousTestCase, self).assertTrue(elapsed < 2)
        t.cancel()

def test_crash(self) -> None:
        """
        C{reactor.crash()} stops the reactor and does not fire shutdown
        triggers.
        """
        reactor = self.buildReactor()
        events = []
        reactor.addSystemEventTrigger(
            "before", "shutdown", lambda: events.append(("before", "shutdown"))
        )
        reactor.callWhenRunning(reactor.callLater, 0, reactor.crash)
        self.runReactor(reactor)
        testCase = cast(SynchronousTestCase, self)
        testCase.assertFalse(reactor.running)
        testCase.assertFalse(
            events, "Shutdown triggers invoked but they should not have been."
        )

def test_runAfterCrash(self) -> None:
        """
        C{reactor.run()} restarts the reactor after it has been stopped by
        C{reactor.crash()}.
        """
        events: List[Union[str, Tuple[str, bool]]] = []

def crash() -> None:
            events.append("crash")
            reactor.crash()

reactor = self.buildReactor()
        reactor.callWhenRunning(crash)
        self.runReactor(reactor)

def stop() -> None:
            events.append(("stop", reactor.running))
            reactor.stop()

reactor.callWhenRunning(stop)
        self.runReactor(reactor)
        cast(SynchronousTestCase, self).assertEqual(events, ["crash", ("stop", True)])

def test_runAfterStop(self) -> None:
        """
        C{reactor.run()} raises L{ReactorNotRestartable} when called when
        the reactor is being run after getting stopped priorly.
        """
        events: List[str] = []

testCase = cast(SynchronousTestCase, self)

def restart() -> None:
            testCase.assertRaises(ReactorNotRestartable, reactor.run)
            events.append("tested")

reactor = self.buildReactor()
        reactor.callWhenRunning(reactor.stop)
        reactor.addSystemEventTrigger("after", "shutdown", restart)
        self.runReactor(reactor)
        testCase.assertEqual(events, ["tested"])

globals().update(SystemEventTestsBuilder.makeTestCaseClasses())

cmd: