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_defer_yieldfrom.py (4848B)

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

"""
Tests for C{yield from} support in Deferreds.
"""

import types

from twisted.internet.defer import Deferred, ensureDeferred, fail, succeed
from twisted.internet.task import Clock
from twisted.trial.unittest import TestCase

class YieldFromTests(TestCase):
    """
    Tests for using Deferreds in conjunction with PEP-380.
    """

def test_ensureDeferred(self):
        """
        L{ensureDeferred} will turn a coroutine into a L{Deferred}.
        """

def run():
            d = succeed("foo")
            res = yield from d
            return res

# It's a generator...
        r = run()
        self.assertIsInstance(r, types.GeneratorType)

# Now it's a Deferred.
        d = ensureDeferred(r)
        self.assertIsInstance(d, Deferred)

# The Deferred has the result we want.
        res = self.successResultOf(d)
        self.assertEqual(res, "foo")

def test_DeferredfromCoroutine(self):
        """
        L{Deferred.fromCoroutine} will turn a coroutine into a L{Deferred}.
        """

def run():
            d = succeed("bar")
            yield from d
            res = yield from run2()
            return res

def run2():
            d = succeed("foo")
            res = yield from d
            return res

# It's a generator...
        r = run()
        self.assertIsInstance(r, types.GeneratorType)

# Now it's a Deferred.
        d = Deferred.fromCoroutine(r)
        self.assertIsInstance(d, Deferred)

# The Deferred has the result we want.
        res = self.successResultOf(d)
        self.assertEqual(res, "foo")

def test_basic(self):
        """
        L{Deferred.fromCoroutine} allows a function to C{yield from} a
        L{Deferred}.
        """

def run():
            d = succeed("foo")
            res = yield from d
            return res

d = Deferred.fromCoroutine(run())
        res = self.successResultOf(d)
        self.assertEqual(res, "foo")

def test_exception(self):
        """
        An exception in a generator scheduled with L{Deferred.fromCoroutine}
        will cause the returned L{Deferred} to fire with a failure.
        """

def run():
            d = succeed("foo")
            yield from d
            raise ValueError("Oh no!")

d = Deferred.fromCoroutine(run())
        res = self.failureResultOf(d)
        self.assertEqual(type(res.value), ValueError)
        self.assertEqual(res.value.args, ("Oh no!",))

def test_twoDeep(self):
        """
        An exception in a generator scheduled with L{Deferred.fromCoroutine}
        will cause the returned L{Deferred} to fire with a failure.
        """
        reactor = Clock()
        sections = []

def runone():
            sections.append(2)
            d = Deferred()
            reactor.callLater(1, d.callback, None)
            yield from d
            sections.append(3)
            return "Yay!"

def run():
            sections.append(1)
            result = yield from runone()
            sections.append(4)
            d = Deferred()
            reactor.callLater(1, d.callback, None)
            yield from d
            sections.append(5)
            return result

d = Deferred.fromCoroutine(run())

reactor.advance(0.9)
        self.assertEqual(sections, [1, 2])

reactor.advance(0.1)
        self.assertEqual(sections, [1, 2, 3, 4])

reactor.advance(0.9)
        self.assertEqual(sections, [1, 2, 3, 4])

reactor.advance(0.1)
        self.assertEqual(sections, [1, 2, 3, 4, 5])

res = self.successResultOf(d)
        self.assertEqual(res, "Yay!")

def test_reraise(self):
        """
        Yielding from an already failed Deferred will raise the exception.
        """

def test():
            try:
                yield from fail(ValueError("Boom"))
            except ValueError as e:
                self.assertEqual(e.args, ("Boom",))
                return 1
            return 0

res = self.successResultOf(Deferred.fromCoroutine(test()))
        self.assertEqual(res, 1)

def test_chained(self):
        """
        Yielding from a paused & chained Deferred will give the result when it
        has one.
        """
        reactor = Clock()

def test():
            d = Deferred()
            d2 = Deferred()
            d.addCallback(lambda ignored: d2)

d.callback(None)
            reactor.callLater(0, d2.callback, "bye")
            res = yield from d
            return res

d = Deferred.fromCoroutine(test())
        reactor.advance(0.1)

res = self.successResultOf(d)
        self.assertEqual(res, "bye")

cmd: