| 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/python/test/__pycache__/ |
Upload File : |
o
�b"( � @ s� d Z ddlZddlZddlZddlZddlmZmZmZmZ ddl m
Z
m Z m Z m
Z
m Z ddlmZ z
ddl mZmZ W n eyG dZY nw dZdd lmZ dd
lmZ dd lmZmZ dd lm Z dd
l m Z ddl m Z ddl!m"Z" ddl#m$Z$ e"�%� r�ddl m&Z& dZ'ndZ'zddl(m)Z)m*Z*m+Z+m,Z, W n
ey� dZ-dZ.Y nw dZ-dZ.G dd� d�Z/dd� Z0G dd� de1�Z2G dd � d e �Z3d d � Z4ee-e.�G d d � d e$��Z5ee-e.�G d!d"� d"e$��Z6dS )#z&
Tests for L{twisted.python.sendmsg}.
� N)�close�pathsep�pipe�read)�AF_INET�AF_INET6�
SOL_SOCKET�error�socket)�pack)�AF_UNIX�
socketpairTF)�skipIf)�reactor)�Deferred�inlineCallbacks)� ProcessDone)�ProcessProtocol)�FilePath)�platform)�TestCase)� MSG_DONTWAIT)�
SCM_RIGHTS�getSocketFamily�recvmsg�sendmsgz!Platform doesn't support sendmsg.� c @ s@ e Zd ZdZdd� Zdd� Zdd� Zdd � Zd
d � Zd d
� Z dS )� _FDHolderzP
A wrapper around a FD that will remember if it has been closed or not.
c C s
|| _ d S �N��_fd)�self�fd� r# �B/usr/lib/python3/dist-packages/twisted/python/test/test_sendmsg.py�__init__: s
z_FDHolder.__init__c C s | j S )z/
Return the fileno of this FD.
r �r! r# r# r$ �fileno= s z_FDHolder.filenoc C s | j r
t| j � d| _ dS dS )zH
Close the FD. If it's already been closed, do nothing.
N)r r r& r# r# r$ r C s
�z_FDHolder.closec C s, | j rt�d| j � d�t� | �� dS dS )z>
If C{self._fd} is unclosed, raise a warning.
zFD z was not closed!N)r �warnings�warn�ResourceWarningr r& r# r# r$ �__del__K s �z_FDHolder.__del__c C s | S r r# r& r# r# r$ � __enter__S s z_FDHolder.__enter__c C s | � � d S r )r )r! �exc_type� exc_value� tracebackr# r# r$ �__exit__V s z_FDHolder.__exit__N)
�__name__�
__module__� __qualname__�__doc__r% r' r r+ r, r0 r# r# r# r$ r 5 s r c C s t � \} }t| �t|�fS )zI
Create a pipe, and return the two FDs wrapped in L{_FDHolders}.
)r r )�r�wr# r# r$ � _makePipeZ s
r7 c @ s e Zd ZdZdefdd�ZdS )�ExitedWithStderrz,
A process exited with some stderr.
�returnc C s d� dgt| j� �}t|�S )zY
Dump the errors in a pretty way in the event of a subprocess traceback.
�
� )�join�list�args�repr�r! �resultr# r# r$ �__str__g s zExitedWithStderr.__str__N)r1 r2 r3 r4 �strrB r# r# r# r$ r8 b s r8 c @ s8 e Zd ZdZdd� Zdd� Zdd� Zdd � Zd
d � Zd S )
�StartStopProcessProtocola�
An L{IProcessProtocol} with a Deferred for events where the subprocess
starts and stops.
@ivar started: A L{Deferred} which fires with this protocol's
L{IProcessTransport} provider when it is connected to one.
@ivar stopped: A L{Deferred} which fires with the process output or a
failure if the process produces output on standard error.
@ivar output: A C{str} used to accumulate standard output.
@ivar errors: A C{str} used to accumulate standard error.
c C s t � | _t � | _d| _d| _d S )Nr; )r �started�stopped�output�errorsr& r# r# r$ r% s
z!StartStopProcessProtocol.__init__c C s | j �| j� d S r )rE �callback� transportr& r# r# r$ �connectionMade� � z'StartStopProcessProtocol.connectionMadec C � | j |7 _ d S r )rG �r! �datar# r# r$ � outReceived� rL z$StartStopProcessProtocol.outReceivedc C rM r )rH rN r# r# r$ � errReceived� rL z$StartStopProcessProtocol.errReceivedc C s6 |� t�r| j�| j� d S | j�t| j| j�� d S r )�checkr rF rI rG �errbackr8 rH )r! �reasonr# r# r$ � processEnded� s
z%StartStopProcessProtocol.processEndedN) r1 r2 r3 r4 r% rK rP rQ rU r# r# r# r$ rD o s rD c
C s� t tj��� j}ttj�}t t� tj���� j|d< t
� }t j |||t t
��| d ��� jd|f g|dddddd||id � |S )
z�
Start a script that is a peer of this test as a subprocess.
@param script: the module name of the script in this directory (no
package prefix, no '.py')
@type script: C{str}
@rtype: L{StartStopProcessProtocol}
�
PYTHONPATHz.pys %dr r6 � r5 � )�env�childFDs)r �sys�
executable�
asTextMode�path�dict�os�environr r< rD r � spawnProcess�__file__�sibling)�script�outputFD�pyExerY �ssppr# r# r$ �_spawn� s
�� ri c @ sf e Zd ZdZdd� Zdd� Zdd� Zdd � Zd
d � Zd d
� Z dd� Z
e e d�dd� �Z
edd� �ZdS )� SendmsgTestszB
Tests for the Python2/3 compatible L{sendmsg} interface.
c C s t t�\| _| _dS )z0
Create a pair of UNIX sockets.
N)r
r �inputrG r& r# r# r$ �setUp� s zSendmsgTests.setUpc C s | j �� | j�� dS )z4
Close the sockets opened by setUp.
N)rk r rG r& r# r# r$ �tearDown� s
zSendmsgTests.tearDownc C s4 | j �� | �tt| j d�}| �|jd tj� dS )z�
If the underlying C{sendmsg} call fails, L{send1msg} raises
L{socket.error} with its errno set to the underlying errno value.
� hello, worldr N� rk r � assertRaisesr r � assertEqualr> �errno�EBADF�r! �excr# r# r$ �test_syscallError� s
z SendmsgTests.test_syscallErrorc C s: | j �� | �tt| j ddgd�}| �|jd tj� dS )z�
The behavior when the underlying C{sendmsg} call fails is the same
whether L{sendmsg} is passed ancillary data or not.
rn )r r s 0123r Nro rt r# r# r$ �#test_syscallErrorWithControlMessage� s
�z0SendmsgTests.test_syscallErrorWithControlMessagec C sT d}| � t|�t| j|�� t| j�}| � |jd� | � |jd� | � |jg � dS )zI
L{recvmsg} will retrieve a message sent via L{sendmsg}.
�
hello, world!r N) rq �lenr rk r rG rO �flags� ancillary)r! �messagerA r# r# r$ �test_roundtrip� s
zSendmsgTests.test_roundtripc C s^ dd d }| j �d� t| j |�}| �|t|�k � t| jt|��}| �t|d �|� dS )zS
L{sendmsg} returns the number of bytes which it was able to send.
� xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx� � Fr N)rk � setblockingr �
assertTruery r rG rq )r! r| �sent�receivedr# r# r$ �test_shortsend� s zSendmsgTests.test_shortsendc C s0 t | jdg d� t| j�}| �|dg df� dS )z�
L{sendmsg} treats an empty ancillary data list the same way it treats
receiving no argument for the ancillary parameter at all.
rx r N)r rk r rG rq r@ r# r# r$ � test_roundtripEmptyAncillary� s
z)SendmsgTests.test_roundtripEmptyAncillaryz7MSG_DONTWAIT is only known to work as intended on Linuxc C sl t d�D ]*}z
t| jdtd� W q ty. } z| �|jd tj� W Y d}~ dS d}~ww | � d� dS )z�
The C{flags} argument to L{sendmsg} is passed on to the underlying
C{sendmsg} call, to affect it in whatever way is defined by those
flags.
i r~ )rz r NzHFailed to fill up the send buffer, or maybe send1msg blocked for a while)
�ranger rk r �OSErrorrq r> rr �EAGAIN�fail)r! �i�er# r# r$ �
test_flags� s
���zSendmsgTests.test_flagsc
c s� � t d| j�� �}|jV t� \}}| �|j� | �|j� |� t| jdt t
t d|�� �fg� W d � n1 s;w Y |j V | �
t|�� d�d� | �
t|�� d�d� dS )z�
Calling L{sendmsg} with SOL_SOCKET, SCM_RIGHTS, and a platform-endian
packed file descriptor number should send that file descriptor to a
different process, where it can be retrieved by using L{recv1msg}.
�pullpipes blonkr� Nr s Test fixture data: blonk.
r; )ri rG r' rE r7 �
addCleanupr r rk r r r rF rq r )r! rh �pipeOut�pipeInr# r# r$ �test_sendSubProcessFD s �
� �z"SendmsgTests.test_sendSubProcessFDN)r1 r2 r3 r4 rl rm rv rw r} r� r� r � dontWaitSkipr� r r� r# r# r# r$ rj � s
rj c @ s: e Zd ZdZdd� Zdd� Zdd� Zeed�d d
� �Z d S ) �GetSocketFamilyTestsz'
Tests for L{getSocketFamily}.
c C s t |�}| �|j� |S )z�
Create a new socket using the given address family and return that
socket's file descriptor. The socket will automatically be closed when
the test is torn down.
)r
r� r )r! �
addressFamily�sr# r# r$ �_socket3 s z GetSocketFamilyTests._socketc C � | � tt| �t��� dS )z�
When passed the file descriptor of a socket created with the C{AF_INET}
address family, L{getSocketFamily} returns C{AF_INET}.
N)rq r r r� r& r# r# r$ � test_inet= � z GetSocketFamilyTests.test_inetc C r� )z�
When passed the file descriptor of a socket created with the
C{AF_INET6} address family, L{getSocketFamily} returns C{AF_INET6}.
N)rq r r r� r& r# r# r$ �
test_inet6D r� z GetSocketFamilyTests.test_inet6z)Platform does not support AF_UNIX socketsc C r� )z�
When passed the file descriptor of a socket created with the C{AF_UNIX}
address family, L{getSocketFamily} returns C{AF_UNIX}.
N)rq r r r� r& r# r# r$ � test_unixK s z GetSocketFamilyTests.test_unixN)
r1 r2 r3 r4 r� r� r� r � nonUNIXSkipr� r# r# r# r$ r� - s
r� )7r4 rr r` r[ r( r r r r r
r r r r �structr r r
� ImportErrorr� �unittestr �twisted.internetr �twisted.internet.deferr r �twisted.internet.errorr �twisted.internet.protocolr �twisted.python.filepathr �twisted.python.runtimer �twisted.trial.unittestr �isLinuxr r� �twisted.python.sendmsgr r r r � doImportSkip�importSkipReasonr r7 � Exceptionr8 rD ri rj r� r# r# r# r$ �<module> sV � �%
& {