opt
/
hc_python
/
lib
/
python3.12
/
site-packages
/
nose
/
__pycache__
/
Go to Home Directory
+
Upload
Create File
root@0UT1S:~$
Execute
By Order of Mr.0UT1S
[DIR] ..
N/A
__init__.cpython-312.pyc
634 bytes
Rename
Delete
__main__.cpython-312.pyc
450 bytes
Rename
Delete
case.cpython-312.pyc
16.52 KB
Rename
Delete
commands.cpython-312.pyc
7.20 KB
Rename
Delete
config.cpython-312.pyc
30.73 KB
Rename
Delete
core.cpython-312.pyc
16.78 KB
Rename
Delete
exc.cpython-312.pyc
572 bytes
Rename
Delete
failure.cpython-312.pyc
2.29 KB
Rename
Delete
importer.cpython-312.pyc
7.76 KB
Rename
Delete
inspector.cpython-312.pyc
8.42 KB
Rename
Delete
loader.cpython-312.pyc
26.48 KB
Rename
Delete
proxy.cpython-312.pyc
10.12 KB
Rename
Delete
pyversion.cpython-312.pyc
9.99 KB
Rename
Delete
result.cpython-312.pyc
8.92 KB
Rename
Delete
selector.cpython-312.pyc
11.74 KB
Rename
Delete
suite.cpython-312.pyc
27.26 KB
Rename
Delete
twistedtools.cpython-312.pyc
6.74 KB
Rename
Delete
util.cpython-312.pyc
27.33 KB
Rename
Delete
� ��g� � �d � d Z ddlZddlmZmZ ddlmZmZ g d�Zda d� Z e � \ ZZd� Z d d�Zy) 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 �� �� ddl m� t sAddlm} ddlm} |�fd��� at j d� t j � �t fS # t $ r Y yw xY w) 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 )NN)� threadable)�Threadc �( �� � j d�� S )NF)�installSignalHandlers)�runr s ��@/opt/hc_python/lib/python3.12/site-packages/nose/twistedtools.py�<lambda>z"threaded_reactor.<locals>.<lambda>8 s �� ����&+� 1<� 1-� )�targetT) �twisted.internetr �ImportError�_twisted_thread�twisted.pythonr � threadingr � setDaemon�start)r r r s @r r r * s^ �� ��,� �-�$� � )-� .���!�!�$�'������O�#�#�� � ���s �A � A$�#A$c �� � d� } t j | � t j � t j � D ]# }|j � s�|j � �% day)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 �, � t j � y)z0Helper for calling stop from withing the thread.N)r �stop� r r r z"stop_reactor.<locals>.stop_reactorJ s � ����r N)r �callFromThread�reactor_thread�join�getDelayedCalls�active�cancelr )r �ps r r r B sO � �� ���<�(����� � $� $� &���8�8�:� �H�H�J� '� �Or c � � �� t � \ �}��t d� � � du xs � dz �� fd�}|S # t $ r t d� �w xY w)a 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") Nz1twisted is not available or could not be importedr z+'timeout' argument must be a number or Nonec �: �� � � ��fd�} t � � |� }|S )Nc � �� ���� � t � � � fd��� fd��� ��� �� fd�}�j |� � j ��� }|�|\ }}} ||� j |� �y # t $ r t d�z � �w xY w)Nc �( �� �j d � y �N)�put)�value�qs �r �callbackz=deferred.<locals>.decorate.<locals>.wrapper.<locals>.callback� s �� ����d�r c �| �� | j � y # �j t j � � Y y xY wr, )�raiseExceptionr- �sys�exc_info)�failurer/ s �r �errbackz<deferred.<locals>.decorate.<locals>.wrapper.<locals>.errback� s- �� �*��*�*�,��*��E�E�#�,�,�.�)�s � �%;c �� �� ��i ���} | j ��� y # t $ r t d� �w xY w# �j t j � � Y y xY w)Nz7you must return a twisted Deferred from your test case!)�addCallbacks�AttributeError� TypeErrorr- r3 r4 )�d�argsr0 r6 �func�kargsr/ s ������r �gz6deferred.<locals>.decorate.<locals>.wrapper.<locals>.g� sj �� �*��d�,�e�,�A�@����x��9�� *� @�'� )?� @� @�@�� *��E�E�#�,�,�.�)�s �7 � �4�7 �%A)�timeoutz*timeout expired before end of test (%f s.))r r! �getr r �with_traceback) r<