FM / test

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

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

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

Edit: /usr/lib/python3/dist-packages/twisted/test/test_defgen.py (10634B)

# 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",
        )

cmd: