← Back
Editing: test_unix.cpython-38.pyc
U W[5; � @ s* d Z ddlmZmZ ddlZddlZddlZddlZddlm Z m Z m Z m Z m Z ddlmZmZ ddlmZ ddlmZmZ ddlmZ dd lmZ dd lmZmZ G d d � d e j �Z G d d� dej �Z G dd� de j �Z!G dd� de j"�Z#G dd� dej �Z$e �%e d��sde _&e �'e d��s&de$_&dS )zK Tests for implementations of L{IReactorUNIX} and L{IReactorUNIXDatagram}. � )�division�absolute_importN)� interfaces�reactor�protocol�error�address)�defer�utils)�lockfile)�_PY3� networkString)�FilePath)�unittest)�MyServerFactory�MyClientFactoryc @ s 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__ s z&FailedConnectionClientFactory.__init__c C s | j �|� d S r )r Zerrback)r Z connector�reasonr r r �clientConnectionFailed s z4FailedConnectionClientFactory.clientConnectionFailedN)�__name__� __module__� __qualname__r r r r r r r s r c @ sj e Zd ZdZdd� Zdd� Zdd� Zdd � Zd d � Zd d � Z dd� Z dd� Z dd� Z e r^de _dd� ZdS )�UnixSocketTestsz Test unix sockets. 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� transportZgetPeer)�protoZexpected�Zpeernamer � serverFactoryr r � cbConnMade6 s z1UnixSocketTests.test_peerBind.<locals>.cbConnMade)�mktempr r �Deferred�protocolConnectionMader � listenUNIX� addCleanup� stopListening�socketZAF_UNIXZ SOCK_STREAM�closeZbindZconnect� addCallback)r �filenameZconnMade�unixPortZ 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�serverProtocolZclientProtocol�� clientFactoryr0 r r r � allConnectedN s � z1UnixSocketTests.test_dumber.<locals>.allConnected) r' r r r( r) r r* r+ r, r � connectUNIX� gatherResultsr/ )r r% �serverConnMader1 �clientConnMade�dr8 r r6 r � test_dumber> 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�ZwantPID�.lock� �ZcheckPIDc s: | \}}�� � jt���g� |j�� |j�� ��� S r )r r! r r r" r3 r, )r4 r5 Z clientProto�r7 r0 r r1 r r � _portStuffq s � z0UnixSocketTests.test_pidFile.<locals>._portStuffc s �� t�� d �d� d S )Nr@ �locked)Z assertFalser �isLocked�Zignored)r0 r r r �_check~ s z,UnixSocketTests.test_pidFile.<locals>._check)r' r r r( r) r r* Z assertTruer rF r r9 r: r/ )r r% r; r< r= rD rH r rC r � test_pidFile[ 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. Tr? c s t j� �dd�}|�� S �NTr? )r r* r, )�ignr1 �r0 r% r r �stoppedListening� s z<UnixSocketTests.test_socketLocking.<locals>.stoppedListening) r' r r r* � assertRaisesr �CannotListenErrorr, r/ )r r1 rM r rL 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' r0 r r �os�pathsep�join�sys�pathZ asBytesMode� executabler ZgetProcessValuer/ )r �callback�sourcerQ ZpyExer= r r r �_uncleanSocketTest� s � � z"UnixSocketTests._uncleanSocketTestc s � 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 rJ )r r* r0 r r, )rK �p�r r r �ranStupidChild� s zGUnixSocketTests.test_uncleanServerSocketLocking.<locals>.ranStupidChild�rZ �r r] r r\ r � test_uncleanServerSocketLocking� s z/UnixSocketTests.test_uncleanServerSocketLockingc s � fdd�}� � |�S )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 )NTrB ) r r( r r r9 r0 Z assertFailurer Z BadFileError)rK r= �fr\ r r r] � s zCUnixSocketTests.test_connectToUncleanServer.<locals>.ranStupidChildr^ r_ r r\ r �test_connectToUncleanServer� s z+UnixSocketTests.test_connectToUncleanServerc sj �� � }t�||��d� |f }��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. � <%s on %r>c s. d� f }�� t��|� �� t��|� d S �Nz<%s (not listening)>�r �repr�str�rK ZunconnectedString�� factoryNamer r1 r r rM � s z3UnixSocketTests._reprTest.<locals>.stoppedListening) r' r r* r rf rg r � maybeDeferredr, r/ )r r% rj r0 �connectedStringr= rM r ri r � _reprTest� s zUnixSocketTests._reprTestc C s* G dd� d�}| � |tj� | �|� d�S )a The two string representations of the L{IListeningPort} returned by L{IReactorUNIX.listenUNIX} contains the name of the classic factory class being used and the filename on which the port is listening or indicates that the port is not listening. c @ s e Zd Zdd� Zdd� ZdS )zCUnixSocketTests.test_reprWithClassicFactory.<locals>.ClassicFactoryc S s d S r r r\ r r r �doStart� s zKUnixSocketTests.test_reprWithClassicFactory.<locals>.ClassicFactory.doStartc S s d S r r r\ r r r �doStop� s zJUnixSocketTests.test_reprWithClassicFactory.<locals>.ClassicFactory.doStopN�r r r rn ro r r r r �ClassicFactory� s rq z%twisted.test.test_unix.ClassicFactory��assertIsInstance�typesZ ClassTyperm )r rq r r r �test_reprWithClassicFactory� s �z+UnixSocketTests.test_reprWithClassicFactory�)Classic classes do not exist on Python 3.c C s* G dd� dt �}| �|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 @ s e Zd Zdd� Zdd� ZdS )zEUnixSocketTests.test_reprWithNewStyleFactory.<locals>.NewStyleFactoryc S s d S r r r\ r r r rn � s zMUnixSocketTests.test_reprWithNewStyleFactory.<locals>.NewStyleFactory.doStartc S s d S r r r\ r r r ro � s zLUnixSocketTests.test_reprWithNewStyleFactory.<locals>.NewStyleFactory.doStopNrp r r r r �NewStyleFactory� s rw z&twisted.test.test_unix.NewStyleFactory��objectrs �typerm )r rw r r r � test_reprWithNewStyleFactory� s �z,UnixSocketTests.test_reprWithNewStyleFactoryN)r r r �__doc__r2 r> rI rP rZ r` rb rm ru r �skipr{ r r r r r " s )�r c @ s8 e Zd Zd ZZdZdd� Zdd� Zdd� Zd d � Z dS ) � ClientProtoFNc C s t �� | _t �� | _d S r )r r( �deferredStarted�deferredGotBackr\ r r r r s zClientProto.__init__c C s d| _ d S �NT��stoppedr\ r r r � stopProtocol s zClientProto.stopProtocolc C s d| _ | j�d � d S r� ��startedr rX r\ r r r � startProtocol s zClientProto.startProtocolc C s || _ | j�d � d S r )�gotbackr� rX )r �datar r r �datagramReceived s 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 s t �� | _t �� | _d S r )r r( r �deferredGotWhatr\ r r r r s zServerProto.__init__c C s d| _ d S r� r� r\ r r r r� $ s zServerProto.stopProtocolc C s d| _ | j�d � d S r� r� r\ r r r r� ' s zServerProto.startProtocolc C s* || _ | j�d|� || _| j�d � d S )N� hi back)�gotfromr" �write�gotwhatr� rX )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� c @ sB e Zd ZdZdd� Zdd� Zdd� Zdd � Zer6d e_ d d � Z d 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. )Z bindAddressc s � j �d� t��j� jg�S )N� hi)r" r� r r: r� r� rG )�cp�spr r r� F s �z4DatagramUnixSocketTests.test_exchange.<locals>.writec s. �� d�j� �� � �j� �� d�j� d S )Nr� r� )r r� r� r� rG �Z clientaddrr� r r� r r �_cbTestExchangeK s z>DatagramUnixSocketTests.test_exchange.<locals>._cbTestExchange) r' r� r~ r �listenUNIXDatagramr+ r, ZconnectUNIXDatagramr r: r r/ )r Z serveraddr�s�cr= r� r� r r� r � test_exchange7 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� rN r rO r, rR �unlink)r r� r[ r� r r r �test_cannotListenU s �z)DatagramUnixSocketTests.test_cannotListenc sj �� � }t�||��d� |f }��t��|� ��t��|� t��j�}� ��fdd�}|� |� |S )z� Test the C{__str__} and C{__repr__} implementations of a UNIX datagram port when used with the given protocol. rc c s. d� f }�� t��|� �� t��|� d S rd re rh �� protocolNamer r1 r r rM r s z;DatagramUnixSocketTests._reprTest.<locals>.stoppedListening) r' r r� r rf rg r rk r, r/ )r Z serverProtor� r0 rl Z stopDeferredrM r r� r rm e s z!DatagramUnixSocketTests._reprTestc C s* G dd� d�}| � |tj� | �|� d�S )a0 The two string representations of the L{IListeningPort} returned by L{IReactorUNIXDatagram.listenUNIXDatagram} contains the name of the classic protocol class being used and the filename on which the port is listening or indicates that the port is not listening. c @ s e Zd Zdd� Zdd� ZdS )zMDatagramUnixSocketTests.test_reprWithClassicProtocol.<locals>.ClassicProtocolc S s d S r r �r r" r r r �makeConnection� s z\DatagramUnixSocketTests.test_reprWithClassicProtocol.<locals>.ClassicProtocol.makeConnectionc S s d S r r r\ r r r ro � s zTDatagramUnixSocketTests.test_reprWithClassicProtocol.<locals>.ClassicProtocol.doStopN�r r r r� ro r r r r �ClassicProtocol� s r� z&twisted.test.test_unix.ClassicProtocolrr )r r� r r r � test_reprWithClassicProtocolz s �z4DatagramUnixSocketTests.test_reprWithClassicProtocolrv c C s* G dd� dt �}| �|t� | �|� d�S )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 @ s e Zd Zdd� Zdd� ZdS )zODatagramUnixSocketTests.test_reprWithNewStyleProtocol.<locals>.NewStyleProtocolc S s d S r r r� r r r r� � s z^DatagramUnixSocketTests.test_reprWithNewStyleProtocol.<locals>.NewStyleProtocol.makeConnectionc S s d S r r r\ r r r ro � s zVDatagramUnixSocketTests.test_reprWithNewStyleProtocol.<locals>.NewStyleProtocol.doStopNr� r r r r �NewStyleProtocol� s r� z'twisted.test.test_unix.NewStyleProtocolrx )r r� r r r � test_reprWithNewStyleProtocol� s �z5DatagramUnixSocketTests.test_reprWithNewStyleProtocolN) r r r r| r� r� rm r� r r} r� r r r r r� 3 s �r� z1This reactor does not support UNIX domain socketsz3This reactor does not support UNIX datagram sockets)(r| Z __future__r r rR rU rt r- Ztwisted.internetr r r r r r r Ztwisted.pythonr Ztwisted.python.compatr r Ztwisted.python.filepathr Z twisted.trialr Ztwisted.test.test_tcpr r Z ClientFactoryr ZTestCaser ZConnectedDatagramProtocolr~ ZDatagramProtocolr� r� Z IReactorUNIXr} ZIReactorUNIXDatagramr r r r �<module> s, fv
Save File
Cancel