| Server IP : 138.197.176.125 / Your IP : 216.73.217.54 Web Server : Apache/2.4.41 (Ubuntu) System : Linux SuiteCRM-8 5.4.0-216-generic #236-Ubuntu SMP Fri Apr 11 19:53:21 UTC 2025 x86_64 User : root ( 0) PHP Version : 8.3.19 Disable Function : NONE MySQL : OFF | cURL : ON | WGET : ON | Perl : ON | Python : ON | Sudo : ON | Pkexec : ON Directory : /lib/python3/dist-packages/twisted/internet/test/__pycache__/ |
Upload File : |
U
�q�[94 � @ sr d Z ddlZzddlmZ W n ek
r< ddlmZ Y nX ddlmZ ddlmZ ddl m
Z
m Z m Z ddl
mZ ddlmZ dd lmZ dd
lmZmZmZ dd lmZ dd lmZmZ ee
e �G d
d� de ��Z G dd� de�Z dd� Z G dd� de �Z G dd� de e�Z!G dd� de e�Z"G dd� de�Z#G dd � d e�Z$z ddl%Z%W n ek
�rl d e$_&Y nX dS ) z%
Tests for L{twisted.internet.base}.
� N)�Queue)� implementer)�
ThreadPool)� IReactorTime�IReactorThreads�IResolverSimple)�DNSLookupError)� FirstOneWins)�Deferred)�ThreadedResolver� DelayedCall� ReactorBase)�Clock)�TestCase�SkipTestc @ s0 e Zd ZdZdd� Zdd� Zdd� Zdd � Zd
S ) � 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>( � z&FakeReactor.__init__.<locals>.<lambda>) r �_clockZ callLaterr r �startZ
getThreadPoolr � _threadCallsr r r r �__init__" s
zFakeReactor.__init__c O s | j �|||f� d S r )r �put�r �f�args�kwargsr r r �callFromThread- s zFakeReactor.callFromThreadc C s | j �� \}}}|||� d S r )r �getr r r r �_runThreadCalls1 s zFakeReactor._runThreadCallsc C s | j �� d S r )r �stopr r r r �_stop6 s zFakeReactor._stopN)�__name__�
__module__� __qualname__�__doc__r r# r% r' r r r r r s
r c @ s0 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�ZipZlookedUpr r �fakeGetHostByNameN s
z=ThreadedResolverTests.test_success.<locals>.fakeGetHostByName�
gethostbyname� N)r �
addCleanupr' �patch�socketr �
getHostByName� addCallbackr. r% � assertEqualr �advance�calls)r r0 �timeout�reactorZ
resolvedTor2 �resolver�dr r1 r � test_success? 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��d S )Nz ENOBUFS (this is a funny joke))�IOErrorr/ r r r r2 l s z=ThreadedResolverTests.test_failure.<locals>.fakeGetHostByNamer3 � some.namer4 N)r r5 r' r6 r7 r r8 �
assertFailurer r9 r. r% r: �lenr r; r< �r r= r> r2 Z
failedWithr? r@ r r r � test_failureb 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 � � � �d S r )r$ r/ ��resultr r r2 � s z=ThreadedResolverTests.test_timeout.<locals>.fakeGetHostByNamer3 rC r4 zThe I/O was errorfulN)r r5 r' r r6 r7 r r8 rD r r9 r. r r; r: rE r rB rF r rI r � test_timeout� s z"ThreadedResolverTests.test_timeoutc s g � t t�G � fdd�dt��}G dd� dt�}|� }|� }|�|� tt� �}|j�|d� |j�|d� |j�|d� |j�|d� |j�|d� | � t
� �d � | � t t t
� ��tgd � | � 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 r0 Ztimeouts�r< r r r8 � s
zOThreadedResolverTests.test_resolverGivenStr.<locals>.FakeResolver.getHostByNameN)r )r( r) r* r8 r rL r r � FakeResolver� s rM c @ s e Zd Zdd� ZdS )zFThreadedResolverTests.test_resolverGivenStr.<locals>.JustEnoughReactorc S s d S r r r r r r � installWaker� s zSThreadedResolverTests.test_resolverGivenStr.<locals>.JustEnoughReactor.installWakerN)r( r) r* rN r r r r �JustEnoughReactor� s rO zexample.exampleu vääntynyt.exampleu рф.examplez xn----7sbb4ac0ad0be6cf.xn--p1ai� r r4 zxn--vntynyt-5waa.example� zxn--p1ai.example� � N)r r �objectr
ZinstallResolverr r
Z nameResolverZresolveHostNamer: rE �list�map�type�str)r rM rO Zfaker> Zrecr rL r �test_resolverGivenStr� sF
� � � � � z+ThreadedResolverTests.test_resolverGivenStrN)r( r) r* r+ rA rG rK rY r r r r r, ; s
# !r, c C s dS )z8
Function used by L{DelayedCallTests.test_str}.
Nr r r r r �nothing� s rZ 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 s d S r r )�callr r r �noop� s z0DelayedCallMixin._getDelayedCallAt.<locals>.noopc S s d S r r r r r r r � r z4DelayedCallMixin._getDelayedCallAt.<locals>.<lambda>r N)r )r �timer^ 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 r4 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.
� �rR �ArP Nc S s dS �Ng �?r r r r r r � r z+DelayedCallMixin.test_str.<locals>.<lambda>z?<DelayedCall 0x%x [10.5s] called=0 cancelled=0 nothing(3, A=5)>)r rZ r: rX �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}.
�
)� rg � Nc S s dS )Ng�������?r r r r r r � r z,DelayedCallMixin.test_repr.<locals>.<lambda>)r rZ r: rX �reprrj 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�ra rb �
assertTrue� assertFalse�r ra rb r r r �test_lt s
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}.
Nrr ru r r r �test_le s
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}.
Nrr ru r r r �test_gt s
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}.
Nrr ru r r r �test_ge' s
zDelayedCallMixin.test_gec C s: | � | j| jk� | �| j| jk� | �| j| jk� dS )zD
A L{DelayedCall} instance is only equal to itself.
N)rt ra rb rs r r r r �test_eq4 s 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)rs ra rb rt r r r r �test_ne? s zDelayedCallMixin.test_neN)r( r) r* r+ r` rc rl rq rv rw rx ry rz r{ r r r r r[ � s
r[ c @ s e Zd ZdZdd� Zdd� ZdS )�DelayedCallNoDebugTestsr\ c C s | � tdd� t�| � dS )z!
Turn debug off.
�debugFN�r6 r r[ rc r r r r rc O s z DelayedCallNoDebugTests.setUpc C s> t dtdddidddd� �}d�t|��}| �t|�|� dS ) rd re rf rg rP Nc S s dS rh r r r r r r ] r z2DelayedCallNoDebugTests.test_str.<locals>.<lambda>zA<DelayedCall 0x{:x} [10.5s] called=0 cancelled=0 nothing(3, A=5)>)r rZ �formatri r: rX )r rk Zexpectedr r r rl W s ��z DelayedCallNoDebugTests.test_strN�r( r) r* r+ rc rl r r r r r| K s r| c @ s e Zd ZdZdd� Zdd� ZdS )�DelayedCallDebugTestsr\ c C s | � tdd� t�| � dS )z
Turn debug on.
r} TNr~ r r r r rc i s zDelayedCallDebugTests.setUpc C s> t dtdddidddd� �}d�t|��}| �t|�|� dS ) rd re rf rg rP Nc S s dS rh r r r r r r w r z0DelayedCallDebugTests.test_str.<locals>.<lambda>z\<DelayedCall 0x{:x} \[10.5s\] called=0 cancelled=0 nothing\(3, A=5\)
traceback at creation:)r rZ r ri Z assertRegexrX )r rk ZexpectedRegexpr r r rl q s �� �z DelayedCallDebugTests.test_strNr� r r r r r� e s r� c @ s e Zd ZdZdd� ZdS )� TestSpySignalCapturingReactorza
Subclass of ReactorBase to capture signals delivered to the
reactor for inspection.
c C s dS )z*
Required method, unused.
Nr r r r r rN � s z*TestSpySignalCapturingReactor.installWakerN)r( r) r* r+ rN r r r r r� � s r� c @ s0 e Zd ZdZdd� Zdd� Zdd� Zdd � Zd
S ) �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� ZassertIs� _exitSignal�r r> r r r � test_exitSignalDefaultsToNone� s z4ReactorBaseSignalTests.test_exitSignalDefaultsToNonec C s( t � }|�tjd� | �tj|j� dS )zn
ReactorBase's SIGINT handler saves the value of SIGINT to the
_exitSignal attribute.
N)r� ZsigInt�signal�SIGINT� assertEqualsr� r� r r r �test_captureSIGINT� s z)ReactorBaseSignalTests.test_captureSIGINTc C s( t � }|�tjd� | �tj|j� dS )zp
ReactorBase's SIGTERM handler saves the value of SIGTERM to the
_exitSignal attribute.
N)r� ZsigTermr� �SIGTERMr� r� r� r r r �test_captureSIGTERM� s z*ReactorBaseSignalTests.test_captureSIGTERMc C s: t td�std��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� ZsigBreakr� r� r� r� r r r �test_captureSIGBREAK� s
z+ReactorBaseSignalTests.test_captureSIGBREAKN)r( r) r* r+ r� r� r� r� r r r r r� � s
r� zsignal module not available)'r+ r7 r � ImportErrorZqueueZzope.interfacer Ztwisted.python.threadpoolr Ztwisted.internet.interfacesr r r Ztwisted.internet.errorr Ztwisted.internet._resolverr Ztwisted.internet.deferr
Ztwisted.internet.baser r r
Ztwisted.internet.taskr Ztwisted.trial.unittestr r rT r r, rZ r[ r| r� r� r� r� �skipr r r r �<module> s8 y 0