File: //lib/python3/dist-packages/twisted/test/__pycache__/test_unix.cpython-310.pyc
o
�b 5 � @ s d Z ddlZddlZddlZddlmZ ddlmZmZm Z m
Z
m Z m Z m
Z
ddlmZ ddlmZ ddlmZ ddlmZmZ dd lmZ G d
d � d e j�Zee
�e d� d �G d
d� dej��Z G dd� de j �Z G dd� de j �Z ee
�!e d� d�G dd� dej��Z"dS )zK
Tests for implementations of L{IReactorUNIX} and L{IReactorUNIXDatagram}.
� N)�skipIf)�address�defer�error�
interfaces�protocol�reactor�utils)�lockfile)�
networkString)�FilePath)�MyClientFactory�MyServerFactory)�unittestc @ � e Zd Zdd� Zdd� ZdS )� FailedConnectionClientFactoryc C s
|| _ d S �N)�onFail)�selfr � r �8/usr/lib/python3/dist-packages/twisted/test/test_unix.py�__init__ �
z&FailedConnectionClientFactory.__init__c C s | j �|� d S r )r �errback)r � connector�reasonr r r �clientConnectionFailed s z4FailedConnectionClientFactory.clientConnectionFailedN)�__name__�
__module__� __qualname__r r r r r r r � r �1This reactor does not support UNIX domain socketsc @ sh e Zd ZdZe�ed�sdZdd� Zdd� Z dd � Z
d
d � Z d d
� Z dd� Z
dd� Zdd� Zdd� ZdS )�UnixSocketTestsz
Test unix sockets.
Nr! c s� �� � }�� � � t� �t�� }�_t�|��}��|j� t � t j
t j �}��|j � |�
� � |�|� � ��fdd�}|�|� |S )z�
The address passed to the server factory's C{buildProtocol} method and
the address returned by the connected protocol's transport's C{getPeer}
method match the address the client socket is bound to.
c s0 t �� �}���j|g� ��| j�� |� d S r )r � UNIXAddress� assertEqual�
peerAddresses� transport�getPeer)�proto�expected��peernamer �
serverFactoryr r �
cbConnMade; s
z1UnixSocketTests.test_peerBind.<locals>.cbConnMade)�mktempr r �Deferred�protocolConnectionMader �
listenUNIX�
addCleanup�
stopListening�socket�AF_UNIX� SOCK_STREAM�close�bind�connect� addCallback)r �filename�connMade�unixPort�
unixSocketr- r r* r �
test_peerBind* s
z UnixSocketTests.test_peerBindc s� �� � �t� }t�� }||_t��|�}��|j� t � � t�� }|� _t�
�� � t� ||g�}� ��fdd�}|� |� |S )z�
L{IReactorUNIX.connectUNIX} can be used to connect a client to a server
started with L{IReactorUNIX.listenUNIX}.
c s6 | \}}�� � jt���g� |j�� |j�� d S r )r$ r% r r# r& �loseConnection)�args�serverProtocol�clientProtocol��
clientFactoryr; r r r � allConnectedT s �
z1UnixSocketTests.test_dumber.<locals>.allConnected)
r. r r r/ r0 r r1 r2 r3 r
� connectUNIX�
gatherResultsr: )r r, �serverConnMader= �clientConnMade�drF r rD r � test_dumberC s
zUnixSocketTests.test_dumberc s� �� � �t� }t�� }||_tj�|dd����t� �d �� t
� � t�� }|� _tj �� dd� t� ||g�}� ���fdd�}|�
|� ��fdd �}|�
|� |S )
z�
A lockfile is created and locked when L{IReactorUNIX.listenUNIX} is
called and released when the Deferred returned by the L{IListeningPort}
provider's C{stopListening} method is called back.
T��wantPID�.lock� ��checkPIDc s: | \}}�� � jt���g� |j�� |j�� ��� S r )r$ r% r r# r& r@ r3 )rA rB � clientProto�rE r; r r= r r �
_portStuffy s �
z0UnixSocketTests.test_pidFile.<locals>._portStuffc s �� t�� d �d� d S )NrO �locked)� assertFalser
�isLocked��ignored)r; r r r �_check� s z,UnixSocketTests.test_pidFile.<locals>._check)r. r r r/ r0 r r1 �
assertTruer
rX r
rG rH r: )r r, rI rJ rK rU r[ r rT r � test_pidFileb s
z UnixSocketTests.test_pidFilec sR | � � � t� �tj� �dd�}| jtjtj� �dd� � �fdd�}|�� �|�S )z�
L{IReactorUNIX.listenUNIX} raises L{error.CannotListenError} if passed
the name of a file on which a server is already listening.
TrM c s t j� �dd�}|�� S �NTrM )r r1 r3 )�ignr= �r; r, r r �stoppedListening� s z<UnixSocketTests.test_socketLocking.<locals>.stoppedListening) r. r r r1 � assertRaisesr �CannotListenErrorr3 r: )r r= ra r r` r �test_socketLocking� s �z"UnixSocketTests.test_socketLockingc C sj | � � | _td| jf �}dttj�tj��� � ji}ttj
�� � j}t j |dd|f|d�}|�
|� |S )Nzmfrom twisted.internet import protocol, reactor
reactor.listenUNIX(%r, protocol.ServerFactory(),wantPID=True)
s
PYTHONPATHs -us -c)�env)r. r; r r �os�pathsep�join�sys�path� asBytesMode�
executabler �getProcessValuer: )r �callback�sourcere �pyExerK r r r �_uncleanSocketTest� s
��
z"UnixSocketTests._uncleanSocketTestc � � fdd�}� � |�S )a"
If passed C{True} for the C{wantPID} parameter, a server can be started
listening with L{IReactorUNIX.listenUNIX} when passed the name of a
file on which a previous server which has not exited cleanly has been
listening using the C{wantPID} option.
c s t j� jt� dd�}|�� S r^ )r r1 r; r r3 )r_ �p�r r r �ranStupidChild� s zGUnixSocketTests.test_uncleanServerSocketLocking.<locals>.ranStupidChild�rq �r ru r rt r � test_uncleanServerSocketLocking� s
z/UnixSocketTests.test_uncleanServerSocketLockingc rr )z�
If passed C{True} for the C{checkPID} parameter, a client connection
attempt made with L{IReactorUNIX.connectUNIX} fails with
L{error.BadFileError}.
c s0 t �� }t|�}tj� j|dd� � �|tj�S )NTrQ ) r r/ r r rG r; �
assertFailurer � BadFileError)r_ rK �frt r r ru � s zCUnixSocketTests.test_connectToUncleanServer.<locals>.ranStupidChildrv rw r rt r �test_connectToUncleanServer� s
z+UnixSocketTests.test_connectToUncleanServerc �p �� � }t�||��d� � d|�d�}��t��|� ��t��|� t��j�}� ��fdd�}|� |� |S )z~
Test the C{__str__} and C{__repr__} implementations of a UNIX port when
used with the given factory.
�<� on �>c �0 d� � d�}�� t��|� �� t��|� d S �Nr~ z (not listening)>�r$ �repr�str�r_ �unconnectedString�� factoryNamer r= r r ra � � z3UnixSocketTests._reprTest.<locals>.stoppedListening)
r. r r1 r$ r� r� r �
maybeDeferredr3 r: )r r, r� r; �connectedStringrK ra r r� r � _reprTest� �
zUnixSocketTests._reprTestc C �( G dd� d�}| � |t� | �|� d�S )a!
The two string representations of the L{IListeningPort} returned by
L{IReactorUNIX.listenUNIX} contains the name of the new-style factory
class being used and the filename on which the port is listening or
indicates that the port is not listening.
c @ r )zEUnixSocketTests.test_reprWithNewStyleFactory.<locals>.NewStyleFactoryc S � d S r r rt r r r �doStart� � zMUnixSocketTests.test_reprWithNewStyleFactory.<locals>.NewStyleFactory.doStartc S r� r r rt r r r �doStop� r� zLUnixSocketTests.test_reprWithNewStyleFactory.<locals>.NewStyleFactory.doStopN)r r r r� r� r r r r �NewStyleFactory� r r� z&twisted.test.test_unix.NewStyleFactory��assertIsInstance�typer� )r r� r r r � test_reprWithNewStyleFactory� �
�z,UnixSocketTests.test_reprWithNewStyleFactory)r r r �__doc__r � IReactorUNIXr �skipr? rL r] rd rq rx r| r� r� r r r r r" s , r" c @ s8 e Zd Zd ZZdZdd� Zdd� Zdd� Zd d
� Z dS ) � ClientProtoFNc C � t �� | _t �� | _d S r )r r/ �deferredStarted�deferredGotBackrt r r r r �
zClientProto.__init__c C �
d| _ d S �NT��stoppedrt r r r � stopProtocol r zClientProto.stopProtocolc C � d| _ | j�d � d S r� ��startedr� rn rt r r r �
startProtocol � zClientProto.startProtocolc C s || _ | j�d � d S r )�gotbackr� rn )r �datar r r �datagramReceived r� z ClientProto.datagramReceived)
r r r r� r� r� r r� r� r� r r r r r� s r� c @ s< e Zd Zd ZZd ZZdd� Zdd� Zdd� Z d d
� Z
dS ) � ServerProtoFNc C r� r )r r/ r� �deferredGotWhatrt r r r r r� zServerProto.__init__c C r� r� r� rt r r r r� r zServerProto.stopProtocolc C r� r� r� rt r r r r� r� zServerProto.startProtocolc C s* || _ | j�d|� || _| j�d � d S )N� hi back)�gotfromr& �write�gotwhatr� rn )r r� �addrr r r r� $ s z ServerProto.datagramReceived) r r r r� r� r� r� r r� r� r� r r r r r� s r� z3This reactor does not support UNIX datagram socketsc @ s0 e Zd ZdZdd� Zdd� Zdd� Zdd � Zd
S ) �DatagramUnixSocketTestsz%
Test datagram UNIX sockets.
c s� �� � � �� � }t� �t� �t�|��}��|j� tj|�� d�}��|j� t� �j
�j
g�}��fdd�}� ���fdd�}|� |� |� |� |S )zf
Test that a datagram can be sent to and received by a server and vice
versa.
)� bindAddressc s � j �d� t��j� jg�S )N� hi)r& r� r rH r� r� rY )�cp�spr r r� D s z4DatagramUnixSocketTests.test_exchange.<locals>.writec s. �� d�j� �� � �j� �� d�j� d S )Nr� r� )r$ r� r� r� rY ��
clientaddrr� r r� r r �_cbTestExchangeH s z>DatagramUnixSocketTests.test_exchange.<locals>._cbTestExchange) r. r� r� r �listenUNIXDatagramr2 r3 �connectUNIXDatagramr rH r� r: )r �
serveraddr�s�crK r� r� r r� r �
test_exchange4 s
z%DatagramUnixSocketTests.test_exchangec C sD | � � }t� }t�||�}| �tjtj||� |�� t� |� dS )z�
L{IReactorUNIXDatagram.listenUNIXDatagram} raises
L{error.CannotListenError} if the unix socket specified is already in
use.
N)
r. r� r r� rb r rc r3 rf �unlink)r r� rs r� r r r �test_cannotListenQ s z)DatagramUnixSocketTests.test_cannotListenc r} )z�
Test the C{__str__} and C{__repr__} implementations of a UNIX datagram
port when used with the given protocol.
r~ r r� c r� r� r� r� �� protocolNamer r= r r ra n r� z;DatagramUnixSocketTests._reprTest.<locals>.stoppedListening)
r. r r� r$ r� r� r r� r3 r: )r � serverProtor� r; r� � stopDeferredra r r� r r� ` r� z!DatagramUnixSocketTests._reprTestc C r� )a2
The two string representations of the L{IListeningPort} returned by
L{IReactorUNIXDatagram.listenUNIXDatagram} contains the name of the
new-style protocol class being used and the filename on which the port
is listening or indicates that the port is not listening.
c @ r )zODatagramUnixSocketTests.test_reprWithNewStyleProtocol.<locals>.NewStyleProtocolc S r� r r )r r& r r r �makeConnection r� z^DatagramUnixSocketTests.test_reprWithNewStyleProtocol.<locals>.NewStyleProtocol.makeConnectionc S r� r r rt r r r r� � r� zVDatagramUnixSocketTests.test_reprWithNewStyleProtocol.<locals>.NewStyleProtocol.doStopN)r r r r� r� r r r r �NewStyleProtocol~ r r� z'twisted.test.test_unix.NewStyleProtocolr� )r r� r r r � test_reprWithNewStyleProtocolv r� z5DatagramUnixSocketTests.test_reprWithNewStyleProtocolN)r r r r� r� r� r� r� r r r r r� + s r� )#r� rf r4 ri r r �twisted.internetr r r r r r r �twisted.pythonr
�twisted.python.compatr �twisted.python.filepathr �twisted.test.test_tcpr
r �
twisted.trial�
ClientFactoryr r� �TestCaser" �ConnectedDatagramProtocolr� �DatagramProtocolr� �IReactorUNIXDatagramr� r r r r �<module> s2 $ � ` �