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_inlinecb.py (11482B)

# -*- test-case-name: twisted.internet.test.test_inlinecb -*-
# Copyright (c) Twisted Matrix Laboratories.
# See LICENSE for details.

"""
Tests for L{twisted.internet.defer.inlineCallbacks}.

Some tests for inlineCallbacks are defined in L{twisted.test.test_defgen} as
well.
"""

from twisted.internet.defer import (
    CancelledError,
    Deferred,
    inlineCallbacks,
    returnValue,
)
from twisted.trial.unittest import SynchronousTestCase, TestCase

class StopIterationReturnTests(TestCase):
    """
    On Python 3.4 and newer generator functions may use the C{return} statement
    with a value, which is attached to the L{StopIteration} exception that is
    raised.

L{inlineCallbacks} will use this value when it fires the C{callback}.
    """

def test_returnWithValue(self):
        """
        If the C{return} statement has a value it is propagated back to the
        L{Deferred} that the C{inlineCallbacks} function returned.
        """
        environ = {"inlineCallbacks": inlineCallbacks}
        exec(
            """
@inlineCallbacks
def f(d):
    yield d
    return 14
        """,
            environ,
        )
        d1 = Deferred()
        d2 = environ["f"](d1)
        d1.callback(None)
        self.assertEqual(self.successResultOf(d2), 14)

class NonLocalExitTests(TestCase):
    """
    It's possible for L{returnValue} to be (accidentally) invoked at a stack
    level below the L{inlineCallbacks}-decorated function which it is exiting.
    If this happens, L{returnValue} should report useful errors.

If L{returnValue} is invoked from a function not decorated by
    L{inlineCallbacks}, it will emit a warning if it causes an
    L{inlineCallbacks} function further up the stack to exit.
    """

def mistakenMethod(self):
        """
        This method mistakenly invokes L{returnValue}, despite the fact that it
        is not decorated with L{inlineCallbacks}.
        """
        returnValue(1)

def assertMistakenMethodWarning(self, resultList):
        """
        Flush the current warnings and assert that we have been told that
        C{mistakenMethod} was invoked, and that the result from the Deferred
        that was fired (appended to the given list) is C{mistakenMethod}'s
        result.  The warning should indicate that an inlineCallbacks function
        called 'inline' was made to exit.
        """
        self.assertEqual(resultList, [1])
        warnings = self.flushWarnings(offendingFunctions=[self.mistakenMethod])
        self.assertEqual(len(warnings), 1)
        self.assertEqual(warnings[0]["category"], DeprecationWarning)
        self.assertEqual(
            warnings[0]["message"],
            "returnValue() in 'mistakenMethod' causing 'inline' to exit: "
            "returnValue should only be invoked by functions decorated with "
            "inlineCallbacks",
        )

def test_returnValueNonLocalWarning(self):
        """
        L{returnValue} will emit a non-local exit warning in the simplest case,
        where the offending function is invoked immediately.
        """

@inlineCallbacks
        def inline():
            self.mistakenMethod()
            returnValue(2)
            yield 0

d = inline()
        results = []
        d.addCallback(results.append)
        self.assertMistakenMethodWarning(results)

def test_returnValueNonLocalDeferred(self):
        """
        L{returnValue} will emit a non-local warning in the case where the
        L{inlineCallbacks}-decorated function has already yielded a Deferred
        and therefore moved its generator function along.
        """
        cause = Deferred()

@inlineCallbacks
        def inline():
            yield cause
            self.mistakenMethod()
            returnValue(2)

effect = inline()
        results = []
        effect.addCallback(results.append)
        self.assertEqual(results, [])
        cause.callback(1)
        self.assertMistakenMethodWarning(results)

class ForwardTraceBackTests(SynchronousTestCase):
    def test_forwardTracebacks(self):
        """
        Chained inlineCallbacks are forwarding the traceback information
        from generator to generator.

A first simple test with a couple of inline callbacks.
        """

@inlineCallbacks
        def erroring():
            yield "forcing generator"
            raise Exception("Error Marker")

@inlineCallbacks
        def calling():
            yield erroring()

d = calling()
        f = self.failureResultOf(d)
        tb = f.getTraceback()
        self.assertIn("in erroring", tb)
        self.assertIn("in calling", tb)
        self.assertIn("Error Marker", tb)

def test_forwardLotsOfTracebacks(self):
        """
        Several Chained inlineCallbacks gives information about all generators.

A wider test with a 4 chained inline callbacks.

Application stack-trace should be reported, and implementation details
        like "throwExceptionIntoGenerator" symbols are omitted from the stack.

Note that the previous test is testing the simple case, and this one is
        testing the deep recursion case.

That case needs specific code in failure.py to accomodate to stack
        breakage introduced by throwExceptionIntoGenerator.

Hence we keep the two tests in order to sort out which code we
        might have regression in.
        """

@inlineCallbacks
        def erroring():
            yield "forcing generator"
            raise Exception("Error Marker")

@inlineCallbacks
        def calling3():
            yield erroring()

@inlineCallbacks
        def calling2():
            yield calling3()

@inlineCallbacks
        def calling():
            yield calling2()

d = calling()
        f = self.failureResultOf(d)
        tb = f.getTraceback()
        self.assertIn("in erroring", tb)
        self.assertIn("in calling", tb)
        self.assertIn("in calling2", tb)
        self.assertIn("in calling3", tb)
        self.assertNotIn("throwExceptionIntoGenerator", tb)
        self.assertIn("Error Marker", tb)
        self.assertIn("in erroring", f.getTraceback())

class UntranslatedError(Exception):
    """
    Untranslated exception type when testing an exception translation.
    """

class TranslatedError(Exception):
    """
    Translated exception type when testing an exception translation.
    """

class DontFail(Exception):
    """
    Sample exception type.
    """

def __init__(self, actual):
        Exception.__init__(self)
        self.actualValue = actual

class CancellationTests(SynchronousTestCase):
    """
    Tests for cancellation of L{Deferred}s returned by L{inlineCallbacks}.
    For each of these tests, let:
        - C{G} be a generator decorated with C{inlineCallbacks}
        - C{D} be a L{Deferred} returned by C{G}
        - C{C} be a L{Deferred} awaited by C{G} with C{yield}
    """

def setUp(self):
        """
        Set up the list of outstanding L{Deferred}s.
        """
        self.deferredsOutstanding = []

def tearDown(self):
        """
        If any L{Deferred}s are still outstanding, fire them.
        """
        while self.deferredsOutstanding:
            self.deferredGotten()

@inlineCallbacks
    def sampleInlineCB(self, getChildDeferred=None):
        """
        Generator for testing cascade cancelling cases.

@param getChildDeferred: Some callable returning L{Deferred} that we
            awaiting (with C{yield})
        """
        if getChildDeferred is None:
            getChildDeferred = self.getDeferred
        try:
            x = yield getChildDeferred()
        except UntranslatedError:
            raise TranslatedError()
        except DontFail as df:
            x = df.actualValue - 2
        returnValue(x + 1)

def getDeferred(self):
        """
        A sample function that returns a L{Deferred} that can be fired on
        demand, by L{CancellationTests.deferredGotten}.

@return: L{Deferred} that can be fired on demand.
        """
        self.deferredsOutstanding.append(Deferred())
        return self.deferredsOutstanding[-1]

def deferredGotten(self, result=None):
        """
        Fire the L{Deferred} returned from the least-recent call to
        L{CancellationTests.getDeferred}.

@param result: result object to be used when firing the L{Deferred}.
        """
        self.deferredsOutstanding.pop(0).callback(result)

def test_cascadeCancellingOnCancel(self):
        """
        When C{D} cancelled, C{C} will be immediately cancelled too.
        """
        childResultHolder = ["FAILURE"]

def getChildDeferred():
            d = Deferred()

def _eb(result):
                childResultHolder[0] = result.check(CancelledError)
                return result

d.addErrback(_eb)
            return d

d = self.sampleInlineCB(getChildDeferred=getChildDeferred)
        d.addErrback(lambda result: None)
        d.cancel()
        self.assertEqual(
            childResultHolder[0],
            CancelledError,
            "no cascade cancelling occurs",
        )

def test_errbackCancelledErrorOnCancel(self):
        """
        When C{D} cancelled, CancelledError from C{C} will be errbacked
        through C{D}.
        """
        d = self.sampleInlineCB()
        d.cancel()
        self.assertRaises(
            CancelledError,
            self.failureResultOf(d).raiseException,
        )

def test_errorToErrorTranslation(self):
        """
        When C{D} is cancelled, and C raises a particular type of error, C{G}
        may catch that error at the point of yielding and translate it into
        a different error which may be received by application code.
        """

def cancel(it):
            it.errback(UntranslatedError())

a = Deferred(cancel)
        d = self.sampleInlineCB(lambda: a)
        d.cancel()
        self.assertRaises(
            TranslatedError,
            self.failureResultOf(d).raiseException,
        )

def test_errorToSuccessTranslation(self):
        """
        When C{D} is cancelled, and C{C} raises a particular type of error,
        C{G} may catch that error at the point of yielding and translate it
        into a result value which may be received by application code.
        """

def cancel(it):
            it.errback(DontFail(4321))

a = Deferred(cancel)
        d = self.sampleInlineCB(lambda: a)
        results = []
        d.addCallback(results.append)
        d.cancel()
        self.assertEquals(results, [4320])

def test_asynchronousCancellation(self):
        """
        When C{D} is cancelled, it won't reach the callbacks added to it by
        application code until C{C} reaches the point in its callback chain
        where C{G} awaits it.  Otherwise, application code won't be able to
        track resource usage that C{D} may be using.
        """
        moreDeferred = Deferred()

def deferMeMore(result):
            result.trap(CancelledError)
            return moreDeferred

def deferMe():
            d = Deferred()
            d.addErrback(deferMeMore)
            return d

d = self.sampleInlineCB(getChildDeferred=deferMe)
        d.cancel()
        self.assertNoResult(d)
        moreDeferred.callback(6543)
        self.assertEqual(self.successResultOf(d), 6544)

cmd: