| Server IP : 209.38.156.173 / Your IP : 216.73.216.122 [ Web Server : Apache/2.4.52 (Ubuntu) System : Linux lakekumayuhotel 5.15.0-136-generic #147-Ubuntu SMP Sat Mar 15 15:53:30 UTC 2025 x86_64 User : root ( 0) PHP Version : 8.1.2-1ubuntu2.22 Disable Function : NONE Domains : 2 Domains MySQL : OFF | cURL : ON | WGET : ON | Perl : ON | Python : OFF | Sudo : ON | Pkexec : ON Directory : /lib/python3/dist-packages/twisted/internet/test/__pycache__/ |
Upload File : |
o
�bt5 � @ sn d Z ddlZddlmZ ddlmZmZ ddlmZ ddl m
Z
ddl m Z ddl
mZmZmZ dd lmZ dd
lmZ dd lmZmZmZ dd lmZ dd
lm Z ddl m Z m Z zddl Z!W n e"yk dZ Y nw e!Z e
ee�G dd� d��Z#G dd� de �Z$dd� Z%G dd� d�Z&G dd� de&e �Z'G dd� de&e �Z(G dd � d e�Z)ee d �G d d � d e ��Z*dS ) z%
Tests for L{twisted.internet.base}.
� N)�Queue)�Any�Callable)�skipIf)� implementer)� FirstOneWins)� DelayedCall� ReactorBase�ThreadedResolver)�Deferred)�DNSLookupError)�IReactorThreads� IReactorTime�IResolverSimple)�Clock)�
ThreadPool)�SkipTest�TestCasec @ sl e Zd ZdZdd� Zdedef fdd�Zdd � Zd
d � Z d d
� Z
dd� Z dedef fdd�Z dd� Z
dS )� FakeReactorzl
A fake reactor implementation which just supports enough reactor APIs for
L{ThreadedResolver}.
c s> t � � _� jj� _t� � _� j�� � fdd�� _t� � _d S )Nc s � j S �N)� _threadpool� ��selfr �A/usr/lib/python3/dist-packages/twisted/internet/test/test_base.py�<lambda>- s z&FakeReactor.__init__.<locals>.<lambda>) r �_clock� callLaterr r �start�
getThreadPoolr � _threadCallsr r r r �__init__' s
zFakeReactor.__init__�callable.c O s | j �|||f� d S r )r �put�r r" �args�kwargsr r r �callFromThread1 s zFakeReactor.callFromThreadc C s" | j �� \}}}||i |�� d S r )r �get)r �fr% r& r r r �_runThreadCalls4 s zFakeReactor._runThreadCallsc C s | j �� d S r )r �stopr r r r �_stop8 s zFakeReactor._stopc C � d S r r r r r r �getDelayedCalls; � zFakeReactor.getDelayedCallsc C r- r r r r r r �seconds? r/ zFakeReactor.secondsc O r- r r r$ r r r � callInThreadC r/ zFakeReactor.callInThreadc C r- r r )r �sizer r r �suggestThreadPoolSizeG r/ z!FakeReactor.suggestThreadPoolSizeN)�__name__�
__module__� __qualname__�__doc__r! r r r' r* r, r. r0 r1 r3 r r r r r s
r c @ �0 e Zd ZdZdd� Zdd� Zdd� Zdd � Zd
S ) �ThreadedResolverTestsz(
Tests for L{ThreadedResolver}.
c s� d� d}d}t � }| �|j� g �g }� �fdd�}| �td|� t|�}|�||f�}|�|j� |� � | �
�|g� | �
|� g� |j � |d � | �
|j j
g � dS ) z�
L{ThreadedResolver.getHostByName} returns a L{Deferred} which fires
with the value returned by the call to L{socket.gethostbyname} in the
threadpool of the reactor passed to L{ThreadedResolver.__init__}.
z 10.0.0.17zfoo.bar.example.com� c s �� | � � S r )�append��name��ip�lookedUpr r �fakeGetHostByNamea s
z=ThreadedResolverTests.test_success.<locals>.fakeGetHostByName�
gethostbyname� N)r �
addCleanupr, �patch�socketr
�
getHostByName� addCallbackr; r* � assertEqualr �advance�calls)r r= �timeout�reactor�
resolvedTorA �resolver�dr r> r � test_successQ s" z"ThreadedResolverTests.test_successc C s� d}t � }| �|j� dd� }| �td|� g }t|�}|�d|f�}| �|t� |� |j
� |� � | � t
|�d� |j�|d � | � |jjg � dS )z�
L{ThreadedResolver.getHostByName} returns a L{Deferred} which fires a
L{Failure} if the call to L{socket.gethostbyname} raises an exception.
r: c S s t d��)Nz ENOBUFS (this is a funny joke))�OSErrorr< r r r rA ~ � z=ThreadedResolverTests.test_failure.<locals>.fakeGetHostByNamerB � some.namerC N)r rD r, rE rF r
rG �
assertFailurer rH r; r* rI �lenr rJ rK �r rL rM rA �
failedWithrO rP r r r � test_failuret s z"ThreadedResolverTests.test_failurec s� d}t � }| �|j� t� � � fdd�}| �td|� g }t|�}|�d|f�}| �|t � |�
|j � |j �
|d � | �|g � |j �
d� | �t|�d� � �td�� dS ) z�
If L{socket.gethostbyname} does not complete before the specified
timeout elapsed, the L{Deferred} returned by
L{ThreadedResolver.getHostByName} fails with L{DNSLookupError}.
�
c s � � � �r )r( r< ��resultr r rA � rS z=ThreadedResolverTests.test_timeout.<locals>.fakeGetHostByNamerB rT rC zThe I/O was errorfulN)r rD r, r rE rF r
rG rU r rH r; r rJ rI rV r# �IOErrorrW r r[ r � test_timeout� s z"ThreadedResolverTests.test_timeoutc s g � t t�G � fdd�d��}G dd� dt�}|� }|� }|�|� tt� �}|j�|d� |j�|d� |j�|d� |j�|d� |j�|d� | �t � �d � | �t
t t � ��t
gd � | �d� d
� | �d� d � | �d � d
� | �d� d � | �d� d � dS )zm
L{ThreadedResolver.getHostByName} is passed L{str}, encoded using IDNA
if required.
c s e Zd Zd� fdd� ZdS )zAThreadedResolverTests.test_resolverGivenStr.<locals>.FakeResolverr c s � � |� t� S r )r; r )r r= �timeouts�rK r r rG � s
zOThreadedResolverTests.test_resolverGivenStr.<locals>.FakeResolver.getHostByNameN)r )r4 r5 r6 rG r r` r r � FakeResolver� s ra c @ s e Zd Zdd� ZdS )zFThreadedResolverTests.test_resolverGivenStr.<locals>.JustEnoughReactorc S r- r r r r r r � installWaker� � zSThreadedResolverTests.test_resolverGivenStr.<locals>.JustEnoughReactor.installWakerN)r4 r5 r6 rb r r r r �JustEnoughReactor� s rd zexample.exampleu vääntynyt.exampleu рф.examplez xn----7sbb4ac0ad0be6cf.xn--p1ai� r rC zxn--vntynyt-5waa.example� zxn--p1ai.example� � N)r r r �installResolverr r � nameResolver�resolveHostNamerI rV �list�map�type�str)r ra rd �fakerM �recr r` r �test_resolverGivenStr� s(
z+ThreadedResolverTests.test_resolverGivenStrN)r4 r5 r6 r7 rQ rY r^ rr r r r r r9 L s # !r9 c C � dS )z8
Function used by L{DelayedCallTests.test_str}.
Nr r r r r �nothing� � rt c @ s` e Zd ZdZdd� Zdd� Zdd� Zdd � Zd
d � Zd d
� Z dd� Z
dd� Z dd� Z dd� Z
dS )�DelayedCallMixin�
L{DelayedCall}
c C s dd� }t |dd� di ||d�S )z�
Get a L{DelayedCall} instance at a given C{time}.
@param time: The absolute time at which the returned L{DelayedCall}
will be scheduled.
c S r- r r )�callr r r �noop� rc z0DelayedCallMixin._getDelayedCallAt.<locals>.noopc S r- r r r r r r r � ru z4DelayedCallMixin._getDelayedCallAt.<locals>.<lambda>r N)r )r �timery r r r �_getDelayedCallAt� s z"DelayedCallMixin._getDelayedCallAtc C s | � d�| _| � d�| _dS )zb
Create two L{DelayedCall} instanced scheduled to run at different
times.
r rC N)r{ �zero�oner r r r �setUp� s zDelayedCallMixin.setUpc C s: t dtdddidddd� �}| �t|�dt|�f � dS ) ��
The string representation of a L{DelayedCall} instance, as returned by
L{str}, includes the unsigned id of the instance, as well as its state,
the function to be called, and the function arguments.
� �rg �Are Nc S rs �Ng �?r r r r r r � ru z+DelayedCallMixin.test_str.<locals>.<lambda>z?<DelayedCall 0x%x [10.5s] called=0 cancelled=0 nothing(3, A=5)>)r rt rI ro �id�r �dcr r r �test_str� s ��zDelayedCallMixin.test_strc C s4 t dtdddidddd� �}| �t|�t|�� dS )z�
The string representation of a L{DelayedCall} instance, as returned by
{repr}, is identical to that returned by L{str}.
�
)� r� � Nc S rs )Ng�������?r r r r r r ru z,DelayedCallMixin.test_repr.<locals>.<lambda>)r rt rI ro �reprr� r r r � test_repr s zDelayedCallMixin.test_reprc C sJ | j | j}}| �||k � | �||k � | �||k � | �||k � dS )z�
For two instances of L{DelayedCall} C{a} and C{b}, C{a < b} is true
if and only if C{a} is scheduled to run before C{b}.
N�r| r} �
assertTrue� assertFalse�r r| r} r r r �test_lt �
zDelayedCallMixin.test_ltc C sJ | j | j}}| �||k� | �||k� | �||k� | �||k� dS )z�
For two instances of L{DelayedCall} C{a} and C{b}, C{a <= b} is true
if and only if C{a} is scheduled to run before C{b} or at the same
time as C{b}.
Nr� r� r r r �test_le �
zDelayedCallMixin.test_lec C sJ | j | j}}| �||k� | �||k� | �||k� | �||k� dS )z�
For two instances of L{DelayedCall} C{a} and C{b}, C{a > b} is true
if and only if C{a} is scheduled to run after C{b}.
Nr� r� r r r �test_gt" r� zDelayedCallMixin.test_gtc C sJ | j | j}}| �||k� | �||k� | �||k� | �||k� dS )z�
For two instances of L{DelayedCall} C{a} and C{b}, C{a > b} is true
if and only if C{a} is scheduled to run after C{b} or at the same
time as C{b}.
Nr� r� r r r �test_ge- r� zDelayedCallMixin.test_gec C s: | � | j| jk� | �| j| jk� | �| j| jk� dS )zD
A L{DelayedCall} instance is only equal to itself.
N)r� r| r} r� r r r r �test_eq9 � zDelayedCallMixin.test_eqc C s: | � | j| jk� | �| j| jk� | �| j| jk� dS )zM
A L{DelayedCall} instance is not equal to any other object.
N)r� r| r} r� r r r r �test_neC r� zDelayedCallMixin.test_neN)r4 r5 r6 r7 r{ r~ r� r� r� r� r� r� r� r� r r r r rv � s
rv c @ � e Zd ZdZdd� Zdd� ZdS )�DelayedCallNoDebugTestsrw c C � | � tdd� t�| � dS )z!
Turn debug off.
�debugFN�rE r rv r~ r r r r r~ S � z DelayedCallNoDebugTests.setUpc C �> t dtdddidddd� �}d�t|��}| �t|�|� dS ) r r� r� r� re Nc S rs r� r r r r r r ` ru z2DelayedCallNoDebugTests.test_str.<locals>.<lambda>zA<DelayedCall 0x{:x} [10.5s] called=0 cancelled=0 nothing(3, A=5)>)r rt �formatr� rI ro )r r� �expectedr r r r� Z s
�z DelayedCallNoDebugTests.test_strN�r4 r5 r6 r7 r~ r� r r r r r� N � r� c @ r� )�DelayedCallDebugTestsrw c C r� )z
Turn debug on.
r� TNr� r r r r r~ m r� zDelayedCallDebugTests.setUpc C r� ) r r� r� r� re Nc S rs r� r r r r r r z ru z0DelayedCallDebugTests.test_str.<locals>.<lambda>z\<DelayedCall 0x{:x} \[10.5s\] called=0 cancelled=0 nothing\(3, A=5\)
traceback at creation:)r rt r� r� � assertRegexro )r r� �expectedRegexpr r r r� t s
�z DelayedCallDebugTests.test_strNr� r r r r r� h r� r� c @ s e Zd ZdZdd� ZdS )� TestSpySignalCapturingReactorza
Subclass of ReactorBase to capture signals delivered to the
reactor for inspection.
c C rs )z*
Required method, unused.
Nr r r r r rb � ru z*TestSpySignalCapturingReactor.installWakerN)r4 r5 r6 r7 rb r r r r r� � s r� zsignal module not availablec @ r8 ) �ReactorBaseSignalTestszE
Tests to exercise ReactorBase's signal exit reporting path.
c C s t � }| �d|j� dS )zI
The default value of the _exitSignal attribute is None.
N)r� �assertIs� _exitSignal�r rM r r r � test_exitSignalDefaultsToNone� s z4ReactorBaseSignalTests.test_exitSignalDefaultsToNonec C �( t � }|�tjd� | �tj|j� dS )zn
ReactorBase's SIGINT handler saves the value of SIGINT to the
_exitSignal attribute.
N)r� �sigInt�signal�SIGINT� assertEqualsr� r� r r r �test_captureSIGINT� � z)ReactorBaseSignalTests.test_captureSIGINTc C r� )zp
ReactorBase's SIGTERM handler saves the value of SIGTERM to the
_exitSignal attribute.
N)r� �sigTermr� �SIGTERMr� r� r� r r r �test_captureSIGTERM� r� z*ReactorBaseSignalTests.test_captureSIGTERMc C s: t td�s td��t� }|�tjd� | �tj|j� dS )zr
ReactorBase's SIGBREAK handler saves the value of SIGBREAK to the
_exitSignal attribute.
�SIGBREAKz$signal module does not have SIGBREAKN)�hasattrr� r r� �sigBreakr� r� r� r� r r r �test_captureSIGBREAK� s
z+ReactorBaseSignalTests.test_captureSIGBREAKN)r4 r5 r6 r7 r� r� r� r� r r r r r� � s r� )+r7 rF �queuer �typingr r �unittestr �zope.interfacer �twisted.internet._resolverr �twisted.internet.baser r r
�twisted.internet.deferr �twisted.internet.errorr �twisted.internet.interfacesr
r r �twisted.internet.taskr �twisted.python.threadpoolr �twisted.trial.unittestr r r� �_signal� ImportErrorr r9 rt rv r� r� r� r� r r r r �<module> s> �+ r