← Back
Editing: 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
Save File
Cancel