pytest_twisted.py

"""
Allows to test twisted applications with pytest.

Last modified: 2010-12-27 18:15:20 by ralf

Notes: twisted's asynchronous behavior may have influence on the order of test-functions

TODO:
    + credits to Ralf Schmitt See: http://twistedmatrix.com/pipermail/twisted-python/2007-February/014872.html
    + get test to work
"""

import sys

from twisted.internet import reactor, defer
from twisted.python import failure, log
from greenlet import greenlet

def _start_twisted_logging():
    """Enables twisted internal logging"""

    class Logger(object):
        """late-bound sys.stdout"""
        def write(self, msg):
            sys.stdout.write(msg)

        def flush(self):
            sys.stdout.flush()
            # sys.stdout will be changed by py.test later.
    log.startLogging(Logger(), setStdout=0)

def _run_twisted(logging=False):
    """Start twisted mainloop and initialize recursive calling of doit()."""

    # make twisted copy traceback...
    failure.Failure.cleanFailure = lambda *args: None
    if logging:
        _start_twisted_logging()

    def fix_signal_handling():
        # see http://twistedmatrix.com/trac/ticket/733
        import signal
        if hasattr(signal, "siginterrupt"):
            signal.siginterrupt(signal.SIGCHLD, False)

    def start():
        fix_signal_handling()
        doit(None)

    # recursively called for each test-function/method due done()
    def doit(val): # val always None
        # switch context to wait that wrapper() passes back to test-method
        res = gr_tests.switch(val)
        if res is None:
            reactor.stop()
            return

        def done(res):
            reactor.callLater(0.0, doit, None) # recursive call of doit()

        def err(res):
            reactor.callLater(0.0, doit, res)

        # the test-function *may* return a deferred
        # here the test-function will actually been called
        # done() is finalizing a test-process by assuring recursive invoking
        # of doit()
        defer.maybeDeferred(res).addCallback(done).addErrback(err)
    # initially preparing the calling of doit() and starting the reactor
    reactor.callLater(0, start)
    reactor.run()

def pytest_addoption(parser):
    parser.addoption('--twisted-logging', action='store_true', default=False,
                     dest='twisted_logging',
                     help="switch on twisted internal logging")

def pytest_configure(config):
    twisted_logging = config.getvalue("twisted_logging")
    gr_twisted.switch(twisted_logging)

def pytest_unconfigure(config):
    gr_twisted.switch(None)

def pytest_pyfunc_call(pyfuncitem): # (__multicall__, pyfuncitem):
    def doit():
        if 1: #not __multicall__.execute():
            testfunction = pyfuncitem.obj
            if pyfuncitem._isyieldedfunction():
                return testfunction(*pyfuncitem._args)
            else:
                funcargs = pyfuncitem.funcargs
                return testfunction(**funcargs)

    res = gr_twisted.switch(doit)

    if gr_twisted.dead:
        import pytest
        pytest.exit("twisted reactor stopped.")

    if res is not None:
        res.raiseException()

    return True # indicates that we performed the function call


gr_twisted  = greenlet(_run_twisted)
gr_tests    = greenlet.getcurrent()

# ===============================================================================
# plugin tests
# ===============================================================================

def test_generic(testdir):
    testdir.makepyfile('''
        def test_pass():
            pass
        from twisted.internet import defer, reactor
        from twisted.python import failure
        from twisted.python import log


        def test_no_deferred():
            assert True is True

        def test_deferred():
            log.msg("test_deferred() called")
            d = defer.Deferred()
            def done():
                log.msg("test_deferred.done() CALLBACK DONE")
                d.callback(None)

            reactor.callLater(2.5, done)
            log.msg("test_deferred() returning deferred: %r" % (d,))
            return d

        def test_deferred2():
            log.msg("test_deferred2() called")
            d = defer.Deferred()
            def done():
                log.msg("test_deferred2.done() CALLBACK DONE")
                d.callback(None)

            reactor.callLater(2.5, done)
            log.msg("test_deferred2() returning deferred: %r" % (d,))
            return d

        def test_deferred4():
            log.msg("test_deferred4() called")
            from twisted.web.client import getPage
            def printContents(contents):
                assert contents == ""

            deferred = getPage('http://twistedmatrix.com/')
            deferred.addCallback(printContents)
            return deferred

        def test_deferred3():
            log.msg("test_deferred3() called")
            d = defer.Deferred()
            def done():
                log.msg("test_deferred3.done() CALLBACK DONE")
                d.callback(None)

            reactor.callLater(2.5, done)
            log.msg("test_deferred3() returning deferred: %r" % (d,))
            return d

        class TestTwistedSetupMethod:
            def setup_method(self, method):
                log.msg("TestTwistedSetupMethod.setup_method() called")

            def test_deferred(self):
                log.msg("TestTwistedSetupMethod.test_deferred() called")
                d = defer.Deferred()
                def done():
                    log.msg("TestTwistedSetupMethod.test_deferred() CALLBACK DONE")
                    d.callback(None)

                reactor.callLater(2.5, done)
                log.msg("TestTwistedSetupMethod.test_deferred() returning deferred: %r" % (d,))
                return d


        def test_defer_fail():
            def fun():
                log.msg("provoking NameError")
                rsdfg
            return defer.maybeDeferred(fun)
    ''')
    testdir.runpytest("-T")
    # XXX: what to do?
    # s = testdir.tmpdir.join("event.log").read()
    # assert s.find("TestrunFinish") != -1