root/galaxy-central/eggs/nose-0.11.1-py2.6.egg/nose/twistedtools.py

リビジョン 3, 5.2 KB (コミッタ: kohda, 14 年 前)

Install Unix tools  http://hannonlab.cshl.edu/galaxy_unix_tools/galaxy.html

行番号 
1"""
2Twisted integration
3-------------------
4
5This module provides a very simple way to integrate your tests with the
6Twisted_ event loop.
7
8You must import this module *before* importing anything from Twisted itself!
9
10Example::
11
12  from nose.twistedtools import reactor, deferred
13 
14  @deferred()
15  def test_resolve():
16      return reactor.resolve("www.python.org")
17
18Or, 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
31import sys
32from Queue import Queue, Empty
33from 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
42def 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
63reactor, reactor_thread = threaded_reactor()
64
65
66def 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    _twisted_thread = None
76
77
78def deferred(timeout=None):
79    """
80    By wrapping a test function with this decorator, you can return a
81    twisted Deferred and the test will wait for the deferred to be triggered.
82    The whole test function will run inside the Twisted event loop.
83
84    The optional timeout parameter specifies the maximum duration of the test.
85    The difference with timed() is that timed() will still wait for the test
86    to end, while deferred() will stop the test when its timeout has expired.
87    The latter is more desireable when dealing with network tests, because
88    the result may actually never arrive.
89
90    If the callback is triggered, the test has passed.
91    If the errback is triggered or the timeout expires, the test has failed.
92
93    Example::
94   
95        @deferred(timeout=5.0)
96        def test_resolve():
97            return reactor.resolve("nose.python-hosting.com")
98
99    Attention! If you combine this decorator with other decorators (like
100    "raises"), deferred() must be called *first*!
101
102    In other words, this is good::
103       
104        @raises(DNSLookupError)
105        @deferred()
106        def test_error():
107            return reactor.resolve("xxxjhjhj.biz")
108
109    and this is bad::
110       
111        @deferred()
112        @raises(DNSLookupError)
113        def test_error():
114            return reactor.resolve("xxxjhjhj.biz")
115    """
116    reactor, reactor_thread = threaded_reactor()
117    if reactor is None:
118        raise ImportError("twisted is not available or could not be imported")
119    # Check for common syntax mistake
120    # (otherwise, tests can be silently ignored
121    # if one writes "@deferred" instead of "@deferred()")
122    try:
123        timeout is None or timeout + 0
124    except TypeError:
125        raise TypeError("'timeout' argument must be a number or None")
126
127    def decorate(func):
128        def wrapper(*args, **kargs):
129            q = Queue()
130            def callback(value):
131                q.put(None)
132            def errback(failure):
133                # Retrieve and save full exception info
134                try:
135                    failure.raiseException()
136                except:
137                    q.put(sys.exc_info())
138            def g():
139                try:
140                    d = func(*args, **kargs)
141                    try:
142                        d.addCallbacks(callback, errback)
143                    # Check for a common mistake and display a nice error
144                    # message
145                    except AttributeError:
146                        raise TypeError("you must return a twisted Deferred "
147                                        "from your test case!")
148                # Catch exceptions raised in the test body (from the
149                # Twisted thread)
150                except:
151                    q.put(sys.exc_info())
152            reactor.callFromThread(g)
153            try:
154                error = q.get(timeout=timeout)
155            except Empty:
156                raise TimeExpired("timeout expired before end of test (%f s.)"
157                                  % timeout)
158            # Re-raise all exceptions
159            if error is not None:
160                exc_type, exc_value, tb = error
161                raise exc_type, exc_value, tb
162        wrapper = make_decorator(func)(wrapper)
163        return wrapper
164    return decorate
165
Note: リポジトリブラウザについてのヘルプは TracBrowser を参照してください。