[svn r63796] Initial import for pytest_twisted.

--HG--
branch : trunk

contrib/pytest_twisted/__init__.py

@@ -0,0 +1,114 @@
+"""
+TODO:
+    + credits to Ralf Schmitt See: http://twistedmatrix.com/pipermail/twisted-python/2007-February/014872.html
+    + add option for twisted logging
+    + write tests
+"""
+
+import os
+import sys
+
+import py
+try:
+    from twisted.internet.defer import Deferred
+except ImportError:
+    print "To use the twisted option you have to install twisted."
+    sys.exit(0)
+try:
+    from greenlet import greenlet
+except ImportError:
+    print "Since pylib 1.0 greenlet are removed and separately packaged: " \
+            "http://pypi.python.org/pypi/greenlet" 
+    sys.exit(0)
+
+DIR_CUR = str(py.path.local())
+
+
+def _start_twisted_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.
+
+    import twisted.python.log
+    twisted.python.log.startLogging(Logger(), setStdout=0)
+
+def _run_twisted():
+    """greenlet: run twisted mainloop"""
+    from twisted.internet import reactor, defer
+    from twisted.python import log, failure
+    failure.Failure.cleanFailure = lambda *args: None # make twisted copy traceback...
+    _start_twisted_logging() # XXX: add py.test option
+    
+    def doit(val):
+        res = gr_tests.switch(val)
+        if res is None:
+            reactor.stop()
+            return
+            
+        def done(res):
+            reactor.callLater(0.0, doit, None)
+
+        def err(res):
+            reactor.callLater(0.0, doit, res)
+            
+        defer.maybeDeferred(res).addCallback(done).addErrback(err)
+        
+    reactor.callLater(0.0, doit, None)
+    reactor.run()
+
+
+class TwistedPlugin:
+    """Allows to test twisted applications with pytest."""
+
+    def pytest_addoption(self, parser):
+        parser.addoption("--twisted", dest="twisted", 
+            help="Allows to test twisted applications with pytest.")
+
+    def pytest_configure(self, config):
+        twisted = config.getvalue("twisted")
+        if twisted:
+            print "Twisted plugin switched on"
+            gr_twisted.switch()
+
+    def pytest_unconfigure(self, config):
+        gr_twisted.switch(None)
+
+    def pytest_pyfunc_call(self, pyfuncitem, *args, **kwargs):
+        def wrapper(func):
+            res = func.obj()
+            if isinstance(res, Deferred):
+                res = gr_twisted.switch(func.obj)
+                if res:
+                    res.raiseException()
+            return res
+        pyfuncitem = wrapper(pyfuncitem)    
+
+
+gr_twisted  = greenlet(_run_twisted)
+gr_tests    = greenlet.getcurrent()
+
+# ===============================================================================
+# plugin tests 
+# ===============================================================================
+
+# XXX: write test
+'''
+def test_generic(plugintester):
+    plugintester.apicheck(EventlogPlugin)
+
+    testdir = plugintester.testdir()
+    testdir.makepyfile("""
+        def test_pass():
+            pass
+    """)
+    testdir.runpytest("--twisted")
+    s = testdir.tmpdir.join("event.log").read()
+    assert s.find("TestrunStart") != -1
+    assert s.find("ItemTestReport") != -1
+    assert s.find("TestrunFinish") != -1
+'''