/usr/lib/python3/dist-packages/nose/twistedtools.py is in python3-nose 1.1.2-3.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 | """
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
reactor.stop()
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).with_traceback(tb)
wrapper = make_decorator(func)(wrapper)
return wrapper
return decorate
|