/usr/lib/python3/dist-packages/twisted/internet/test
Edit: /usr/lib/python3/dist-packages/twisted/internet/test/_yieldfromtests.py.3only (4338B)
# -*- test-case-name: twisted.internet.test.test_coroutines -*-
# Copyright (c) Twisted Matrix Laboratories.
# See LICENSE for details.
"""
Tests for C{yield from} support in Deferreds.
These tests can only work and be imported on Python 3!
"""
import types
from twisted.internet.defer import Deferred, ensureDeferred, fail
from twisted.trial.unittest import TestCase
from twisted.test.proto_helpers import Clock
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 = Deferred()
d.callback("bar")
yield from d
res = yield from run2()
return res
def run2():
d = Deferred()
d.callback("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_basic(self):
"""
L{ensureDeferred} allows a function to C{yield from} a L{Deferred}.
"""
def run():
d = Deferred()
d.callback("foo")
res = yield from d
return res
d = ensureDeferred(run())
res = self.successResultOf(d)
self.assertEqual(res, "foo")
def test_exception(self):
"""
An exception in a generator wrapped with L{ensureDeferred} will cause
the returned L{Deferred} to fire with a failure.
"""
def run():
d = Deferred()
d.callback("foo")
yield from d
raise ValueError("Oh no!")
d = ensureDeferred(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 wrapped with L{ensureDeferred} 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 = ensureDeferred(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(ensureDeferred(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 = ensureDeferred(test())
reactor.advance(0.1)
res = self.successResultOf(d)
self.assertEqual(res, "bye")