Na}{

Path: ~//usr/lib/python3/dist-packages/twisted/test/

File Content: `test_defgen.py`

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

"""
Tests for L{twisted.internet.defer.deferredGenerator} and related APIs.
"""


from twisted.internet import defer, reactor
from twisted.internet.defer import (
    Deferred,
    deferredGenerator,
    inlineCallbacks,
    returnValue,
    waitForDeferred,
)
from twisted.python.util import runWithWarningsSuppressed
from twisted.trial import unittest
from twisted.trial.util import suppress as SUPPRESS


def getThing():
    d = Deferred()
    reactor.callLater(0, d.callback, "hi")
    return d


def getOwie():
    d = Deferred()

    def CRAP():
        d.errback(ZeroDivisionError("OMG"))

    reactor.callLater(0, CRAP)
    return d


# NOTE: most of the tests in DeferredGeneratorTests are duplicated
# with slightly different syntax for the InlineCallbacksTests below.


class TerminalException(Exception):
    pass


class BaseDefgenTests:
    """
    This class sets up a bunch of test cases which will test both
    deferredGenerator and inlineCallbacks based generators. The subclasses
    DeferredGeneratorTests and InlineCallbacksTests each provide the actual
    generator implementations tested.
    """

    def testBasics(self):
        """
        Test that a normal deferredGenerator works.  Tests yielding a
        deferred which callbacks, as well as a deferred errbacks. Also
        ensures returning a final value works.
        """

        return self._genBasics().addCallback(self.assertEqual, "WOOSH")

    def testBuggy(self):
        """
        Ensure that a buggy generator properly signals a Failure
        condition on result deferred.
        """
        return self.assertFailure(self._genBuggy(), ZeroDivisionError)

    def testNothing(self):
        """Test that a generator which never yields results in None."""

        return self._genNothing().addCallback(self.assertEqual, None)

    def testHandledTerminalFailure(self):
        """
        Create a Deferred Generator which yields a Deferred which fails and
        handles the exception which results.  Assert that the Deferred
        Generator does not errback its Deferred.
        """
        return self._genHandledTerminalFailure().addCallback(self.assertEqual, None)

    def testHandledTerminalAsyncFailure(self):
        """
        Just like testHandledTerminalFailure, only with a Deferred which fires
        asynchronously with an error.
        """
        d = defer.Deferred()
        deferredGeneratorResultDeferred = self._genHandledTerminalAsyncFailure(d)
        d.errback(TerminalException("Handled Terminal Failure"))
        return deferredGeneratorResultDeferred.addCallback(self.assertEqual, None)

    def testStackUsage(self):
        """
        Make sure we don't blow the stack when yielding immediately
        available deferreds.
        """
        return self._genStackUsage().addCallback(self.assertEqual, 0)

    def testStackUsage2(self):
        """
        Make sure we don't blow the stack when yielding immediately
        available values.
        """
        return self._genStackUsage2().addCallback(self.assertEqual, 0)


def deprecatedDeferredGenerator(f):
    """
    Calls L{deferredGenerator} while suppressing the deprecation warning.

    @param f: Function to call
    @return: Return value of function.
    """
    return runWithWarningsSuppressed(
        [
            SUPPRESS(
                message="twisted.internet.defer.deferredGenerator was " "deprecated"
            )
        ],
        deferredGenerator,
        f,
    )


class DeferredGeneratorTests(BaseDefgenTests, unittest.TestCase):

    # First provide all the generator impls necessary for BaseDefgenTests
    @deprecatedDeferredGenerator
    def _genBasics(self):

        x = waitForDeferred(getThing())
        yield x
        x = x.getResult()

        self.assertEqual(x, "hi")

        ow = waitForDeferred(getOwie())
        yield ow
        try:
            ow.getResult()
        except ZeroDivisionError as e:
            self.assertEqual(str(e), "OMG")
        yield "WOOSH"
        return

    @deprecatedDeferredGenerator
    def _genBuggy(self):
        yield waitForDeferred(getThing())
        1 // 0

    @deprecatedDeferredGenerator
    def _genNothing(self):
        if False:
            yield 1

    @deprecatedDeferredGenerator
    def _genHandledTerminalFailure(self):
        x = waitForDeferred(defer.fail(TerminalException("Handled Terminal Failure")))
        yield x
        try:
            x.getResult()
        except TerminalException:
            pass

    @deprecatedDeferredGenerator
    def _genHandledTerminalAsyncFailure(self, d):
        x = waitForDeferred(d)
        yield x
        try:
            x.getResult()
        except TerminalException:
            pass

    def _genStackUsage(self):
        for x in range(5000):
            # Test with yielding a deferred
            x = waitForDeferred(defer.succeed(1))
            yield x
            x = x.getResult()
        yield 0

    _genStackUsage = deprecatedDeferredGenerator(_genStackUsage)

    def _genStackUsage2(self):
        for x in range(5000):
            # Test with yielding a random value
            yield 1
        yield 0

    _genStackUsage2 = deprecatedDeferredGenerator(_genStackUsage2)

    # Tests unique to deferredGenerator

    def testDeferredYielding(self):
        """
        Ensure that yielding a Deferred directly is trapped as an
        error.
        """
        # See the comment _deferGenerator about d.callback(Deferred).
        def _genDeferred():
            yield getThing()

        _genDeferred = deprecatedDeferredGenerator(_genDeferred)

        return self.assertFailure(_genDeferred(), TypeError)

    suppress = [
        SUPPRESS(message="twisted.internet.defer.waitForDeferred was " "deprecated")
    ]


class InlineCallbacksTests(BaseDefgenTests, unittest.TestCase):
    # First provide all the generator impls necessary for BaseDefgenTests

    def _genBasics(self):

        x = yield getThing()

        self.assertEqual(x, "hi")

        try:
            yield getOwie()
        except ZeroDivisionError as e:
            self.assertEqual(str(e), "OMG")
        returnValue("WOOSH")

    _genBasics = inlineCallbacks(_genBasics)

    def _genBuggy(self):
        yield getThing()
        1 / 0

    _genBuggy = inlineCallbacks(_genBuggy)

    def _genNothing(self):
        if False:
            yield 1

    _genNothing = inlineCallbacks(_genNothing)

    def _genHandledTerminalFailure(self):
        try:
            yield defer.fail(TerminalException("Handled Terminal Failure"))
        except TerminalException:
            pass

    _genHandledTerminalFailure = inlineCallbacks(_genHandledTerminalFailure)

    def _genHandledTerminalAsyncFailure(self, d):
        try:
            yield d
        except TerminalException:
            pass

    _genHandledTerminalAsyncFailure = inlineCallbacks(_genHandledTerminalAsyncFailure)

    def _genStackUsage(self):
        for x in range(5000):
            # Test with yielding a deferred
            yield defer.succeed(1)
        returnValue(0)

    _genStackUsage = inlineCallbacks(_genStackUsage)

    def _genStackUsage2(self):
        for x in range(5000):
            # Test with yielding a random value
            yield 1
        returnValue(0)

    _genStackUsage2 = inlineCallbacks(_genStackUsage2)

    # Tests unique to inlineCallbacks

    def testYieldNonDeferred(self):
        """
        Ensure that yielding a non-deferred passes it back as the
        result of the yield expression.

        @return: A L{twisted.internet.defer.Deferred}
        @rtype: L{twisted.internet.defer.Deferred}
        """

        def _test():
            yield 5
            returnValue(5)

        _test = inlineCallbacks(_test)

        return _test().addCallback(self.assertEqual, 5)

    def testReturnNoValue(self):
        """Ensure a standard python return results in a None result."""

        def _noReturn():
            yield 5
            return

        _noReturn = inlineCallbacks(_noReturn)

        return _noReturn().addCallback(self.assertEqual, None)

    def testReturnValue(self):
        """Ensure that returnValue works."""

        def _return():
            yield 5
            returnValue(6)

        _return = inlineCallbacks(_return)

        return _return().addCallback(self.assertEqual, 6)

    def test_nonGeneratorReturn(self):
        """
        Ensure that C{TypeError} with a message about L{inlineCallbacks} is
        raised when a non-generator returns something other than a generator.
        """

        def _noYield():
            return 5

        _noYield = inlineCallbacks(_noYield)

        self.assertIn("inlineCallbacks", str(self.assertRaises(TypeError, _noYield)))

    def test_nonGeneratorReturnValue(self):
        """
        Ensure that C{TypeError} with a message about L{inlineCallbacks} is
        raised when a non-generator calls L{returnValue}.
        """

        def _noYield():
            returnValue(5)

        _noYield = inlineCallbacks(_noYield)

        self.assertIn("inlineCallbacks", str(self.assertRaises(TypeError, _noYield)))


class DeprecateDeferredGeneratorTests(unittest.SynchronousTestCase):
    """
    Tests that L{DeferredGeneratorTests} and L{waitForDeferred} are
    deprecated.
    """

    def test_deferredGeneratorDeprecated(self):
        """
        L{deferredGenerator} is deprecated.
        """

        @deferredGenerator
        def decoratedFunction():
            yield None

        warnings = self.flushWarnings([self.test_deferredGeneratorDeprecated])
        self.assertEqual(len(warnings), 1)
        self.assertEqual(warnings[0]["category"], DeprecationWarning)
        self.assertEqual(
            warnings[0]["message"],
            "twisted.internet.defer.deferredGenerator was deprecated in "
            "Twisted 15.0.0; please use "
            "twisted.internet.defer.inlineCallbacks instead",
        )

    def test_waitForDeferredDeprecated(self):
        """
        L{waitForDeferred} is deprecated.
        """
        d = Deferred()
        waitForDeferred(d)

        warnings = self.flushWarnings([self.test_waitForDeferredDeprecated])
        self.assertEqual(len(warnings), 1)
        self.assertEqual(warnings[0]["category"], DeprecationWarning)
        self.assertEqual(
            warnings[0]["message"],
            "twisted.internet.defer.waitForDeferred was deprecated in "
            "Twisted 15.0.0; please use "
            "twisted.internet.defer.inlineCallbacks instead",
        )

Edit Rename Chmod Delete

FILE FOLDER INFO

Name	Size	Permission
__pycache__	---	0755
__init__.py	475 bytes	0644
cert.pem.no_trailing_newline	1414 bytes	0644
crash_test_dummy.py	549 bytes	0644
ignore_test_failure.py	31459 bytes	0644
iosim.py	18163 bytes	0644
key.pem.no_trailing_newline	1707 bytes	0644
mock_win32process.py	1299 bytes	0644
myrebuilder1.py	151 bytes	0644
myrebuilder2.py	151 bytes	0644
plugin_basic.py	901 bytes	0644
plugin_extra1.py	392 bytes	0644
plugin_extra2.py	550 bytes	0644
process_cmdline.py	123 bytes	0644
process_echoer.py	214 bytes	0644
process_fds.py	984 bytes	0644
process_getargv.py	234 bytes	0644
process_getenv.py	268 bytes	0644
process_linger.py	297 bytes	0644
process_reader.py	178 bytes	0644
process_signal.py	220 bytes	0644
process_stdinreader.py	739 bytes	0644
process_tester.py	779 bytes	0644
process_tty.py	130 bytes	0644
process_twisted.py	1182 bytes	0644
proto_helpers.py	1368 bytes	0644
reflect_helper_IE.py	60 bytes	0644
reflect_helper_VE.py	81 bytes	0644
reflect_helper_ZDE.py	48 bytes	0644
server.pem	4444 bytes	0644
ssl_helpers.py	1650 bytes	0644
stdio_test_consumer.py	1168 bytes	0644
stdio_test_halfclose.py	2047 bytes	0644
stdio_test_hostpeer.py	1089 bytes	0644
stdio_test_lastwrite.py	1154 bytes	0644
stdio_test_loseconn.py	1583 bytes	0644
stdio_test_producer.py	1484 bytes	0644
stdio_test_write.py	902 bytes	0644
stdio_test_writeseq.py	894 bytes	0644
test_abstract.py	3435 bytes	0644
test_adbapi.py	26175 bytes	0644
test_amp.py	110514 bytes	0644
test_application.py	34140 bytes	0644
test_compat.py	18172 bytes	0644
test_context.py	1464 bytes	0644
test_cooperator.py	21339 bytes	0644
test_defer.py	124782 bytes	0644
test_defgen.py	10634 bytes	0644
test_dict.py	1528 bytes	0644
test_dirdbm.py	6851 bytes	0644
test_error.py	8926 bytes	0644
test_factories.py	4567 bytes	0644
test_fdesc.py	7452 bytes	0644
test_finger.py	1894 bytes	0644
test_formmethod.py	3961 bytes	0644
test_ftp.py	130002 bytes	0644
test_ftp_options.py	2688 bytes	0644
test_htb.py	3154 bytes	0644
test_ident.py	6718 bytes	0644
test_internet.py	46469 bytes	0644
test_iosim.py	8908 bytes	0644
test_iutils.py	13632 bytes	0644
test_lockfile.py	15336 bytes	0644
test_log.py	36073 bytes	0644
test_logfile.py	17891 bytes	0644
test_loopback.py	14321 bytes	0644
test_main.py	2166 bytes	0644
test_memcache.py	25284 bytes	0644
test_modules.py	17428 bytes	0644
test_monkey.py	5519 bytes	0644
test_paths.py	70972 bytes	0644
test_pcp.py	12528 bytes	0644
test_persisted.py	13407 bytes	0644
test_plugin.py	25384 bytes	0644
test_policies.py	33061 bytes	0644
test_postfix.py	4427 bytes	0644
test_process.py	87203 bytes	0644
test_protocols.py	7334 bytes	0644
test_randbytes.py	3270 bytes	0644
test_rebuild.py	7323 bytes	0644
test_reflect.py	24469 bytes	0644
test_roots.py	1643 bytes	0644
test_shortcut.py	1962 bytes	0644
test_sip.py	25498 bytes	0644
test_sob.py	5667 bytes	0644
test_socks.py	17496 bytes	0644
test_ssl.py	23301 bytes	0644
test_sslverify.py	116627 bytes	0644
test_stateful.py	2015 bytes	0644
test_stdio.py	12730 bytes	0644
test_strerror.py	5247 bytes	0644
test_strports.py	1705 bytes	0644
test_task.py	46348 bytes	0644
test_tcp.py	65813 bytes	0644
test_tcp_internals.py	13036 bytes	0644
test_text.py	6367 bytes	0644
test_threadable.py	3338 bytes	0644
test_threadpool.py	21956 bytes	0644
test_threads.py	13211 bytes	0644
test_tpfile.py	1545 bytes	0644
test_twistd.py	73768 bytes	0644
test_twisted.py	6277 bytes	0644
test_udp.py	25012 bytes	0644
test_unix.py	13579 bytes	0644
test_usage.py	23311 bytes	0644
testutils.py	5178 bytes	0644

File Content: test_defgen.py

File Content: `test_defgen.py`