1 files changed, 173 insertions, 0 deletions
diff --git a/lib/spack/external/nose/twistedtools.py b/lib/spack/external/nose/twistedtools.py
new file mode 100644
index 0000000000..8d9c6ffe9b
--- /dev/null
+++ b/lib/spack/external/nose/twistedtools.py
@@ -0,0 +1,173 @@
+"""
+Twisted integration
+-------------------
+
+This module provides a very simple way to integrate your tests with the
+Twisted_ event loop.
+
+You must import this module *before* importing anything from Twisted itself!
+
+Example::
+
+  from nose.twistedtools import reactor, deferred
+  
+  @deferred()
+  def test_resolve():
+      return reactor.resolve("www.python.org")
+
+Or, more realistically::
+
+  @deferred(timeout=5.0)
+  def test_resolve():
+      d = reactor.resolve("www.python.org")
+      def check_ip(ip):
+          assert ip == "67.15.36.43"
+      d.addCallback(check_ip)
+      return d
+
+.. _Twisted: http://twistedmatrix.com/trac/
+"""
+
+import sys
+from Queue import Queue, Empty
+from nose.tools import make_decorator, TimeExpired
+
+__all__ = [
+    'threaded_reactor', 'reactor', 'deferred', 'TimeExpired',
+    'stop_reactor'
+]
+
+_twisted_thread = None
+
+def threaded_reactor():
+    """
+    Start the Twisted reactor in a separate thread, if not already done.
+    Returns the reactor.
+    The thread will automatically be destroyed when all the tests are done.
+    """
+    global _twisted_thread
+    try:
+        from twisted.internet import reactor
+    except ImportError:
+        return None, None
+    if not _twisted_thread:
+        from twisted.python import threadable
+        from threading import Thread
+        _twisted_thread = Thread(target=lambda: reactor.run( \
+                installSignalHandlers=False))
+        _twisted_thread.setDaemon(True)
+        _twisted_thread.start()
+    return reactor, _twisted_thread
+
+# Export global reactor variable, as Twisted does
+reactor, reactor_thread = threaded_reactor()
+
+
+def stop_reactor():
+    """Stop the reactor and join the reactor thread until it stops.
+    Call this function in teardown at the module or package level to
+    reset the twisted system after your tests. You *must* do this if
+    you mix tests using these tools and tests using twisted.trial.
+    """
+    global _twisted_thread
+
+    def stop_reactor():
+        '''Helper for calling stop from withing the thread.'''
+        reactor.stop()
+
+    reactor.callFromThread(stop_reactor)
+    reactor_thread.join()
+    for p in reactor.getDelayedCalls():
+        if p.active():
+            p.cancel()
+    _twisted_thread = None
+
+
+def deferred(timeout=None):
+    """
+    By wrapping a test function with this decorator, you can return a
+    twisted Deferred and the test will wait for the deferred to be triggered.
+    The whole test function will run inside the Twisted event loop.
+
+    The optional timeout parameter specifies the maximum duration of the test.
+    The difference with timed() is that timed() will still wait for the test
+    to end, while deferred() will stop the test when its timeout has expired.
+    The latter is more desireable when dealing with network tests, because
+    the result may actually never arrive.
+
+    If the callback is triggered, the test has passed.
+    If the errback is triggered or the timeout expires, the test has failed.
+
+    Example::
+    
+        @deferred(timeout=5.0)
+        def test_resolve():
+            return reactor.resolve("www.python.org")
+
+    Attention! If you combine this decorator with other decorators (like
+    "raises"), deferred() must be called *first*!
+
+    In other words, this is good::
+        
+        @raises(DNSLookupError)
+        @deferred()
+        def test_error():
+            return reactor.resolve("xxxjhjhj.biz")
+
+    and this is bad::
+        
+        @deferred()
+        @raises(DNSLookupError)
+        def test_error():
+            return reactor.resolve("xxxjhjhj.biz")
+    """
+    reactor, reactor_thread = threaded_reactor()
+    if reactor is None:
+        raise ImportError("twisted is not available or could not be imported")
+    # Check for common syntax mistake
+    # (otherwise, tests can be silently ignored
+    # if one writes "@deferred" instead of "@deferred()")
+    try:
+        timeout is None or timeout + 0
+    except TypeError:
+        raise TypeError("'timeout' argument must be a number or None")
+
+    def decorate(func):
+        def wrapper(*args, **kargs):
+            q = Queue()
+            def callback(value):
+                q.put(None)
+            def errback(failure):
+                # Retrieve and save full exception info
+                try:
+                    failure.raiseException()
+                except:
+                    q.put(sys.exc_info())
+            def g():
+                try:
+                    d = func(*args, **kargs)
+                    try:
+                        d.addCallbacks(callback, errback)
+                    # Check for a common mistake and display a nice error
+                    # message
+                    except AttributeError:
+                        raise TypeError("you must return a twisted Deferred "
+                                        "from your test case!")
+                # Catch exceptions raised in the test body (from the
+                # Twisted thread)
+                except:
+                    q.put(sys.exc_info())
+            reactor.callFromThread(g)
+            try:
+                error = q.get(timeout=timeout)
+            except Empty:
+                raise TimeExpired("timeout expired before end of test (%f s.)"
+                                  % timeout)
+            # Re-raise all exceptions
+            if error is not None:
+                exc_type, exc_value, tb = error
+                raise exc_type, exc_value, tb
+        wrapper = make_decorator(func)(wrapper)
+        return wrapper
+    return decorate
+