| Server IP : 104.129.129.72 / Your IP : 216.73.216.229 Web Server : LiteSpeed System : Linux 123answers 5.15.0-143-generic #153-Ubuntu SMP Fri Jun 13 19:10:45 UTC 2025 x86_64 User : answe8064 ( 1002) PHP Version : 8.2.28 Disable Function : NONE MySQL : OFF | cURL : ON | WGET : ON | Perl : ON | Python : OFF | Sudo : ON | Pkexec : ON Directory : /lib/python3/dist-packages/twisted/test/__pycache__/ |
Upload File : |
o
�b
� @ s� d Z ddlZddlZddlmZ zddlZW n ey! dZY nw dZddlm Z ddl
m Z m Z G dd � d �Z
e �e
� G d
d � d e �ZG d d
� d
e �ZdS )z)
Tests for L{twisted.python.threadable}.
� N)�skipIfTF)�
threadable)�FailTest�SynchronousTestCasec @ s" e Zd ZdgZdZdZdd� ZdS )�
TestObject�aMethod���� c C sL t d�D ] }| j| j| _| _| j| j | _| jdks#J d| jf ��qd S )N�
r zz == %d, not 0 as expected)�range�y�x�z)�self�i� r �>/usr/lib/python3/dist-packages/twisted/test/test_threadable.pyr s
�zTestObject.aMethodN)�__name__�
__module__� __qualname__� synchronizedr
r r r r r r r s
r c @ sH e Zd Zdd� Zdd� Zeed�dd� �Zeed�dd � �Zd
d � Z d S )
�SynchronizationTestsc C s | � 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.
gH�����z>N)�
addCleanup�sys�setswitchinterval�getswitchinterval�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)� assertEqualr r r r r r r �test_synchronizedName2 s z*SynchronizationTests.test_synchronizedName�!Platform does not support threadsc 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 ��
foreignResultr r �<lambda>B s z8SynchronizationTests.test_isInIOThread.<locals>.<lambda>��targetr z#Non-IO thread reported as IO threadz#IO thread reported as not IO threadN) r �registerAsIOThread� threading�Thread�start�join� assertFalse�
assertTruer# )r �tr r$ r �test_isInIOThread9 s
��z&SynchronizationTests.test_isInIOThreadc sj t � �g � � �fdd�}g }td�D ]}tj|d�}|�|� |�� q|D ]}|�� q&� r3t� ��d S )Nc
sR zt d�D ]} ��� qW d S ty( } z� �t|�� W Y d }~d S d }~ww �Ni� )r r �AssertionErrorr" �str)r �e��errors�or r �callMethodLotsQ s
� ��zHSynchronizationTests.testThreadedSynchronization.<locals>.callMethodLots� r' )r r r* r+ r" r, r- r )r r9 �threadsr
r0 r r6 r �testThreadedSynchronizationK s
�z0SynchronizationTests.testThreadedSynchronizationc C s t � }td�D ]}|�� qd S r2 )r r r )r r8 r r r r � testUnthreadedSynchronizationd s
�z2SynchronizationTests.testUnthreadedSynchronizationN)
r r r r r r �
threadingSkipr1 r< r= r r r r r ( s
r c @ s& e Zd Zeed�dd� �Zdd� ZdS )�SerializationTestsr c C s4 t �� }t|�}t�|�}t�|�}| �||� d S r! )r �XLock�type�pickle�dumps�loads�assertIsInstance)r �lock�lockType�
lockPickle�newLockr r r � testPicklingk s
z SerializationTests.testPicklingc C s( d}t �|�}t �|d�}t �|� d S )Ns6 ctwisted.python.threadable
unpickle_lock
p0
(tp1
Rp2
.� )rB rD rC )r rH rF � newPickler r r �testUnpicklings s
z!SerializationTests.testUnpicklingN)r r r r r> rJ rM r r r r r? j s
r? )�__doc__rB r �unittestr r* � ImportErrorr> �twisted.pythonr �twisted.trial.unittestr r r � synchronizer r? r r r r �<module> s �
B