File: //proc/self/root/lib/python3/dist-packages/twisted/test/__pycache__/test_threadable.cpython-38.pyc
U
W[� � @ s� d Z ddlmZmZ ddlZddlZz ddlZW n ek
rH dZY nX dZddl m
Z
ddl m Z ddl
mZ G dd � d �Ze�e� G d
d � d e j�ZG d d
� d
e j�ZdS )z)
Tests for L{twisted.python.threadable}.
� )�division�absolute_importNz Platform lacks thread support)�_PY3)�unittest)�
threadablec @ s" e Zd ZdgZdZdZdd� ZdS )�
TestObject�aMethod���� c C sL t d�D ]>}| j| j | _| _| j| j | _| jdkstd| jf ��qd S )N�
r zz == %d, not 0 as expected)�range�y�x�z�AssertionError)�self�i� r �>/usr/lib/python3/dist-packages/twisted/test/test_threadable.pyr s zTestObject.aMethodN)�__name__�
__module__� __qualname__Z synchronizedr r
r r r r r r s r c @ sH e Zd Zdd� Zdd� Zdd� Zdd� Zed k r<ee_ee_d
d � Z d S ) �SynchronizationTestsc C sb t r2ttdd�dk r^| �tjt�� � t�d� n,ttdd�dk r^| �tjt�� � t�d� dS )z�
Reduce the CPython check interval so that thread switches happen much
more often, hopefully exercising more possible race conditions. Also,
delay actual test startup until the reactor has been started.
�getswitchintervalNgH�����z>�getcheckinterval� )r �getattr�sysZ
addCleanup�setswitchintervalr �setcheckintervalr �r r r r �setUp' s zSynchronizationTests.setUpc C s | � dtjj� dS )zk
The name of a synchronized method is inaffected by the synchronization
decorator.
r N)Z assertEqualr r r r r r r �test_synchronizedName7 s z*SynchronizationTests.test_synchronizedNamec sT t �� g � tj� fdd�d�}|�� |�� | �� d d� | �t �� d� dS )z�
L{threadable.isInIOThread} returns C{True} if and only if it is called
in the same thread as L{threadable.registerAsIOThread}.
c s � � t�� �S �N)�appendr � isInIOThreadr �Z
foreignResultr r �<lambda>G � z8SynchronizationTests.test_isInIOThread.<locals>.<lambda>��targetr z#Non-IO thread reported as IO threadz#IO thread reported as not IO threadN) r ZregisterAsIOThread� threading�Thread�start�joinZ assertFalseZ
assertTruer% )r �tr r& r �test_isInIOThread? s
� � �z&SynchronizationTests.test_isInIOThreadc sl t � �g � � �fdd�}g }td�D ]"}tj|d�}|�|� |�� q$|D ] }|�� qL� rht�� ��d S )Nc
sP zt d�D ] } ��� q
W n0 tk
rJ } z� �t|�� W 5 d }~X Y nX d S �Ni� )r r r r$ �str)r �e��errors�or r �callMethodLotsU s
zHSynchronizationTests.testThreadedSynchronization.<locals>.callMethodLots� r) ) r r r+ r, r$ r- r. r ZFailTest)r r7 Zthreadsr r/ r r4 r �testThreadedSynchronizationP s
z0SynchronizationTests.testThreadedSynchronizationNc C s t � }td�D ] }|�� qd S r1 )r r r )r r6 r r r r � testUnthreadedSynchronizationm s z2SynchronizationTests.testUnthreadedSynchronization)
r r r r! r" r0 r9 �
threadingSkip�skipr: r r r r r & s r c @ s* e Zd Zdd� Zedk r ee_dd� ZdS )�SerializationTestsc C s4 t �� }t|�}t�|�}t�|�}| �||� d S r# )r ZXLock�type�pickle�dumps�loadsZassertIsInstance)r �lockZlockType�
lockPickleZnewLockr r r � testPicklingu s
z SerializationTests.testPicklingNc C s( d}t �|�}t �|d�}t �|� d S )Ns6 ctwisted.python.threadable
unpickle_lock
p0
(tp1
Rp2
.� )r? rA r@ )r rC rB Z newPickler r r �testUnpickling� s
z!SerializationTests.testUnpickling)r r r rD r; r< rF r r r r r= t s r= )�__doc__Z
__future__r r r r? r+ � ImportErrorr; Ztwisted.python.compatr Z
twisted.trialr Ztwisted.pythonr r Z synchronizeZSynchronousTestCaser r= r r r r �<module> s
N