Submit
Path:
~
/
/
proc
/
thread-self
/
root
/
opt
/
alt
/
python35
/
lib
/
python3.5
/
site-packages
/
nose
/
__pycache__
/
File Content:
twistedtools.cpython-35.pyc
%ߍ]� � @ s� d Z d d l Z d d l m Z m Z d d l m Z m Z d d d d d g Z d a d d � Z e � \ Z Z d d � Z d d d � Z d S) av 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/ � N)�Queue�Empty)�make_decorator�TimeExpired�threaded_reactor�reactor�deferredr �stop_reactorc s� y d d l m � Wn t k r, d SYn Xt s� d d l m } d d l m } | d � f d d � � a t j d � t j � � t f S)z� 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. r )r N)� threadable)�Thread�targetc s � j d d � S)NZinstallSignalHandlersF)�run� )r r �/twistedtools.py�<lambda>8 s z"threaded_reactor.<locals>.<lambda>T)NN) Ztwisted.internetr �ImportError�_twisted_threadZtwisted.pythonr � threadingr � setDaemon�start)r r r )r r r * s c C sZ d d � } t j | � t j � x* t j � D] } | j � r0 | j � q0 Wd a d S)a 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. c S s t j � d S)z0Helper for calling stop from withing the thread.N)r �stopr r r r r J s z"stop_reactor.<locals>.stop_reactorN)r �callFromThread�reactor_thread�joinZgetDelayedCallsZactive�cancelr )r �pr r r r B s c sy t � \ � } � d k r'