o
    ¯b5  ã                   @   s  d Z ddlZddlZddlZddlmZ ddlmZmZm	Z	m
Z
mZmZmZ ddlmZ ddlmZ ddlmZ ddlmZmZ dd	lmZ G d
d„ dejƒZee
 ed¡ dƒG dd„ dejƒƒZG dd„ dejƒZG dd„ dejƒZ ee
 !ed¡ dƒG dd„ dejƒƒZ"dS )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 Zd Zdd„ Zdd„ ZdS )Ú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 Zd ZdZe ed¡sdZdd„ Zdd„ Z	dd	„ Z
d
d„ Zdd„ Zdd„ Zdd„ Zdd„ Zdd„ ZdS )ÚUnixSocketTestsz
    Test unix sockets.
    Nr!   c                    sŠ   ˆ  ¡ }ˆ  ¡ ‰ tƒ ‰t ¡  }ˆ_t |ˆ¡}ˆ |j¡ t	 	t	j
t	j¡}ˆ |j¡ | ˆ ¡ | |¡ ‡ ‡‡fdd„}| |¡ |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¡}‡ ‡‡fdd„}| |¡ |S )zŠ
        L{IReactorUNIX.connectUNIX} can be used to connect a client to a server
        started with L{IReactorUNIX.listenUNIX}.
        c                    s6   | \}}ˆ  ˆ jt ˆ¡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   
zUnixSocketTests.test_dumberc                    sª   ˆ  ¡ ‰tƒ }t ¡ }||_tjˆ|dd�‰ˆ t 	ˆd ¡¡ t
ƒ ‰ t ¡ }|ˆ _tjˆˆ dd� t ||g¡}‡ ‡‡‡fdd„}| |¡ ‡‡fdd	„}| |¡ |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:   | \}}ˆ  ˆ jt ˆ¡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ƒ ‰tjˆ ˆdd�}| jtjtjˆ ˆdd� ‡ ‡fdd„}| ¡  |¡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ˆ ˆdd�}| ¡ 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   |   ¡ | _td| jf ƒ}dttj tj¡ƒ 	¡ ji}ttj
ƒ 	¡ j}tj|dd|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                    ó   ‡ fdd„}ˆ   |¡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ˆ jtƒ dd�}| ¡ 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|ƒ}tjˆ j|dd� ˆ  |tj¡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¡}‡ ‡‡fdd„}| 	|¡ |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Ô   ó   
zUnixSocketTests._reprTestc                 C   ó(   G dd„ 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 Zd Zd ZZdZdd„ Zdd„ Zdd„ Zd	d
„ Z	dS )ÚClientProtoFNc                 C   ó   t  ¡ | _t  ¡ | _d S r   )r   r/   ÚdeferredStartedÚdeferredGotBackrt   r   r   r   r     ó   
zClientProto.__init__c                 C   ó
   d| _ d S ©NT©Ústoppedrt   r   r   r   ÚstopProtocol	  r   zClientProto.stopProtocolc                 C   ó   d| _ | j d ¡ d S r£   ©ÚstartedrŸ   rn   rt   r   r   r   ÚstartProtocol  ó   zClientProto.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 Zd Zd ZZd ZZdd„ Zdd„ Zdd„ Z	d	d
„ Z
dS )ÚServerProtoFNc                 C   rž   r   )r   r/   rŸ   ÚdeferredGotWhatrt   r   r   r   r     r¡   zServerProto.__init__c                 C   r¢   r£   r¤   rt   r   r   r   r¦     r   zServerProto.stopProtocolc                 C   r§   r£   r¨   rt   r   r   r   rª      r«   zServerProto.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 Zd ZdZdd„ Zdd„ Zdd„ Zdd	„ Zd
S )ÚDatagramUnixSocketTestsz%
    Test datagram UNIX sockets.
    c                    sš   ˆ  ¡ ‰ ˆ  ¡ }tƒ ‰tƒ ‰t |ˆ¡}ˆ |j¡ tj|ˆˆ d�}ˆ |j¡ t 	ˆj
ˆj
g¡}‡‡fdd„}‡ ‡‡‡fdd„}| |¡ | |¡ |S )zf
        Test that a datagram can be sent to and received by a server and vice
        versa.
        )ÚbindAddressc                    s   ˆ j  d¡ t ˆjˆ jg¡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 ||¡}|  tjtj||¡ | ¡  t 	|¡ dS )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   $þ `þ