← Back to branch summary

~plane1/maus/devel_624

« back to all changes in this revision

Viewing changes to third_party/nose-0.11.3/lib/python/nose/twistedtools.py

  • Committer: tunnell
  • Date: 2010-09-30 13:56:05 UTC
  • Revision ID: tunnell@itchy-20100930135605-wxbkfgy75p0sndk3
add third party

Show diffs side-by-side

added added

removed removed

Lines of Context:
third_party/nose-0.11.3/lib/python/nose/twistedtools.py
 
1
"""
 
2
Twisted integration
 
3
-------------------
 
4
 
 
5
This module provides a very simple way to integrate your tests with the
 
6
Twisted_ event loop.
 
7
 
 
8
You must import this module *before* importing anything from Twisted itself!
 
9
 
 
10
Example::
 
11
 
 
12
  from nose.twistedtools import reactor, deferred
 
13
  
 
14
  @deferred()
 
15
  def test_resolve():
 
16
      return reactor.resolve("www.python.org")
 
17
 
 
18
Or, more realistically::
 
19
 
 
20
  @deferred(timeout=5.0)
 
21
  def test_resolve():
 
22
      d = reactor.resolve("www.python.org")
 
23
      def check_ip(ip):
 
24
          assert ip == "67.15.36.43"
 
25
      d.addCallback(check_ip)
 
26
      return d
 
27
 
 
28
.. _Twisted: http://twistedmatrix.com/trac/
 
29
"""
 
30
 
 
31
import sys
 
32
from Queue import Queue, Empty
 
33
from nose.tools import make_decorator, TimeExpired
 
34
 
 
35
__all__ = [
 
36
    'threaded_reactor', 'reactor', 'deferred', 'TimeExpired',
 
37
    'stop_reactor'
 
38
]
 
39
 
 
40
_twisted_thread = None
 
41
 
 
42
def threaded_reactor():
 
43
    """
 
44
    Start the Twisted reactor in a separate thread, if not already done.
 
45
    Returns the reactor.
 
46
    The thread will automatically be destroyed when all the tests are done.
 
47
    """
 
48
    global _twisted_thread
 
49
    try:
 
50
        from twisted.internet import reactor
 
51
    except ImportError:
 
52
        return None, None
 
53
    if not _twisted_thread:
 
54
        from twisted.python import threadable
 
55
        from threading import Thread
 
56
        _twisted_thread = Thread(target=lambda: reactor.run( \
 
57
                installSignalHandlers=False))
 
58
        _twisted_thread.setDaemon(True)
 
59
        _twisted_thread.start()
 
60
    return reactor, _twisted_thread
 
61
 
 
62
# Export global reactor variable, as Twisted does
 
63
reactor, reactor_thread = threaded_reactor()
 
64
 
 
65
 
 
66
def stop_reactor():
 
67
    """Stop the reactor and join the reactor thread until it stops.
 
68
    Call this function in teardown at the module or package level to
 
69
    reset the twisted system after your tests. You *must* do this if
 
70
    you mix tests using these tools and tests using twisted.trial.
 
71
    """
 
72
    global _twisted_thread
 
73
    reactor.stop()
 
74
    reactor_thread.join()
 
75
    for p in reactor.getDelayedCalls():
 
76
        if p.active():
 
77
            p.cancel()
 
78
    _twisted_thread = None
 
79
 
 
80
 
 
81
def deferred(timeout=None):
 
82
    """
 
83
    By wrapping a test function with this decorator, you can return a
 
84
    twisted Deferred and the test will wait for the deferred to be triggered.
 
85
    The whole test function will run inside the Twisted event loop.
 
86
 
 
87
    The optional timeout parameter specifies the maximum duration of the test.
 
88
    The difference with timed() is that timed() will still wait for the test
 
89
    to end, while deferred() will stop the test when its timeout has expired.
 
90
    The latter is more desireable when dealing with network tests, because
 
91
    the result may actually never arrive.
 
92
 
 
93
    If the callback is triggered, the test has passed.
 
94
    If the errback is triggered or the timeout expires, the test has failed.
 
95
 
 
96
    Example::
 
97
    
 
98
        @deferred(timeout=5.0)
 
99
        def test_resolve():
 
100
            return reactor.resolve("nose.python-hosting.com")
 
101
 
 
102
    Attention! If you combine this decorator with other decorators (like
 
103
    "raises"), deferred() must be called *first*!
 
104
 
 
105
    In other words, this is good::
 
106
        
 
107
        @raises(DNSLookupError)
 
108
        @deferred()
 
109
        def test_error():
 
110
            return reactor.resolve("xxxjhjhj.biz")
 
111
 
 
112
    and this is bad::
 
113
        
 
114
        @deferred()
 
115
        @raises(DNSLookupError)
 
116
        def test_error():
 
117
            return reactor.resolve("xxxjhjhj.biz")
 
118
    """
 
119
    reactor, reactor_thread = threaded_reactor()
 
120
    if reactor is None:
 
121
        raise ImportError("twisted is not available or could not be imported")
 
122
    # Check for common syntax mistake
 
123
    # (otherwise, tests can be silently ignored
 
124
    # if one writes "@deferred" instead of "@deferred()")
 
125
    try:
 
126
        timeout is None or timeout + 0
 
127
    except TypeError:
 
128
        raise TypeError("'timeout' argument must be a number or None")
 
129
 
 
130
    def decorate(func):
 
131
        def wrapper(*args, **kargs):
 
132
            q = Queue()
 
133
            def callback(value):
 
134
                q.put(None)
 
135
            def errback(failure):
 
136
                # Retrieve and save full exception info
 
137
                try:
 
138
                    failure.raiseException()
 
139
                except:
 
140
                    q.put(sys.exc_info())
 
141
            def g():
 
142
                try:
 
143
                    d = func(*args, **kargs)
 
144
                    try:
 
145
                        d.addCallbacks(callback, errback)
 
146
                    # Check for a common mistake and display a nice error
 
147
                    # message
 
148
                    except AttributeError:
 
149
                        raise TypeError("you must return a twisted Deferred "
 
150
                                        "from your test case!")
 
151
                # Catch exceptions raised in the test body (from the
 
152
                # Twisted thread)
 
153
                except:
 
154
                    q.put(sys.exc_info())
 
155
            reactor.callFromThread(g)
 
156
            try:
 
157
                error = q.get(timeout=timeout)
 
158
            except Empty:
 
159
                raise TimeExpired("timeout expired before end of test (%f s.)"
 
160
                                  % timeout)
 
161
            # Re-raise all exceptions
 
162
            if error is not None:
 
163
                exc_type, exc_value, tb = error
 
164
                raise exc_type, exc_value, tb
 
165
        wrapper = make_decorator(func)(wrapper)
 
166
        return wrapper
 
167
    return decorate
 
168