_yieldfromtests.py.3only

# -*- 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.internet.task 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")