o
    Óó·i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   ú\C:\Users\psant\Downloads\PAF backend\school\venv\Lib\site-packages\twisted/test/test_unix.pyÚ__init__   ó   
z&FailedConnectionClientFactory.__init__c                 C   s   | j  |¡ d S r   )r   Zerrback)r   Z	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Ú	transportZgetPeer)ÚprotoÚexpected©Úpeernamer   ÚserverFactoryr   r   Ú
cbConnMade;   s   
z1UnixSocketTests.test_peerBind.<locals>.cbConnMadeN)Úmktempr   r   ÚDeferredÚprotocolConnectionMader   Ú
listenUNIXÚ
addCleanupÚstopListeningÚsocketÚAF_UNIXÚSOCK_STREAMÚcloseÚbindÚconnectÚaddCallback)r   ÚfilenameZconnMadeÚunixPortZ
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ÚserverProtocolZclientProtocol©ÚclientFactoryr8   r   r   r   ÚallConnectedT   s   ÿ
z1UnixSocketTests.test_dumber.<locals>.allConnectedN)r+   r   r   r,   r-   r   r.   r/   r0   r   ÚconnectUNIXÚgatherResultsr7   )r   r)   ÚserverConnMader9   ÚclientConnMadeÚdr@   r   r>   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©ZwantPIDú.locké   ©ZcheckPIDc                    s:   | \}}ˆ  ˆ jt ˆ¡g¡ |j ¡  |j ¡  ˆ ¡ S r   )r"   r#   r   r!   r$   r;   r0   )r<   r=   ZclientProto©r?   r8   r   r9   r   r   Ú
_portStuffy   s   ÿ

z0UnixSocketTests.test_pidFile.<locals>._portStuffc                    s   ˆ  t ˆ d ¡d¡ d S )NrH   Úlocked)ZassertFalser
   ÚisLocked©Zignored)r8   r   r   r   Ú_checkˆ   s   z,UnixSocketTests.test_pidFile.<locals>._checkN)r+   r   r   r,   r-   r   r.   Ú
assertTruer
   rN   r   rA   rB   r7   )r   r)   rC   rD   rE   rL   rP   r   rK   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.
        TrG   c                    s   t jˆ ˆdd�}| ¡ S ©NTrG   )r   r.   r0   )Úignr9   ©r8   r)   r   r   ÚstoppedListeningŸ   s   z<UnixSocketTests.test_socketLocking.<locals>.stoppedListeningN)	r+   r   r   r.   ÚassertRaisesr   ÚCannotListenErrorr0   r7   )r   r9   rV   r   rU   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+   r8   r   r   ÚosÚpathsepÚjoinÚsysÚpathZasBytesModeÚ
executabler	   ZgetProcessValuer7   )r   ÚcallbackÚsourcerZ   ZpyExerE   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 rS   )r   r.   r8   r   r0   )rT   Úp©r   r   r   ÚranStupidChild¾   s   zGUnixSocketTests.test_uncleanServerSocketLocking.<locals>.ranStupidChildN©rc   ©r   rg   r   rf   r   Útest_uncleanServerSocketLocking¶   s   
z/UnixSocketTests.test_uncleanServerSocketLockingc                    rd   )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 )NTrJ   )	r   r,   r   r   rA   r8   ZassertFailurer   ZBadFileError)rT   rE   Úfrf   r   r   rg   Ì   s   zCUnixSocketTests.test_connectToUncleanServer.<locals>.ranStupidChildNrh   ri   r   rf   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 ©Nrn   z (not listening)>©r"   ÚreprÚstr©rT   ZunconnectedString©ÚfactoryNamer   r9   r   r   rV   â   ó   z3UnixSocketTests._reprTest.<locals>.stoppedListeningN)
r+   r   r.   r"   rt   ru   r   ÚmaybeDeferredr0   r7   )r   r)   rx   r8   ÚconnectedStringrE   rV   r   rw   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   rf   r   r   r   ÚdoStartó   ó   zMUnixSocketTests.test_reprWithNewStyleFactory.<locals>.NewStyleFactory.doStartc                 S   r   r   r   rf   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.NewStyleFactoryN©ZassertIsInstanceÚtyper|   )r   rƒ   r   r   r   Útest_reprWithNewStyleFactoryê   ó
   ÿz,UnixSocketTests.test_reprWithNewStyleFactory)r   r   r   Ú__doc__r   ÚIReactorUNIXr   Úskipr:   rF   rR   rY   rc   rj   rl   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ÚdeferredGotBackrf   r   r   r   r     ó   
zClientProto.__init__c                 C   ó
   d| _ d S ©NT©Ústoppedrf   r   r   r   ÚstopProtocol	  r   zClientProto.stopProtocolc                 C   ó   d| _ | j d ¡ d S r‘   ©Ústartedr�   ra   rf   r   r   r   ÚstartProtocol  ó   zClientProto.startProtocolc                 C   s   || _ | j d ¡ d S r   )ÚgotbackrŽ   ra   )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�   ÚdeferredGotWhatrf   r   r   r   r     r�   zServerProto.__init__c                 C   r�   r‘   r’   rf   r   r   r   r”     r   zServerProto.stopProtocolc                 C   r•   r‘   r–   rf   r   r   r   r˜      r™   zServerProto.startProtocolc                 C   s*   || _ | j d|¡ || _| j d ¡ d S )Nó   hi back)Úgotfromr$   ÚwriteÚgotwhatrž   ra   )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.
        )ZbindAddressc                    s   ˆ j  d¡ t ˆjˆ jg¡S )Nó   hi)r$   r¡   r   rB   rž   rŽ   rO   )Ú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š   rO   ©Z
clientaddrr¦   r   r§   r   r   Ú_cbTestExchangeH  s   z>DatagramUnixSocketTests.test_exchange.<locals>._cbTestExchangeN)r+   r�   r‹   r   ÚlistenUNIXDatagramr/   r0   ZconnectUNIXDatagramr   rB   r�   r7   )r   Z
serveraddrÚsÚcrE   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ª   rW   r   rX   r0   r[   Úunlink)r   r£   re   r«   r   r   r   Útest_cannotListenQ  s   z)DatagramUnixSocketTests.test_cannotListenc                    rm   )zˆ
        Test the C{__str__} and C{__repr__} implementations of a UNIX datagram
        port when used with the given protocol.
        rn   ro   rp   c                    rq   rr   rs   rv   ©ÚprotocolNamer   r9   r   r   rV   n  ry   z;DatagramUnixSocketTests._reprTest.<locals>.stoppedListeningN)
r+   r   rª   r"   rt   ru   r   rz   r0   r7   )r   ZserverProtor±   r8   r{   ZstopDeferredrV   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   rf   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.NewStyleProtocolNr„   )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ˆ   r[   r1   r^   r   r   Ztwisted.internetr   r   r   r   r   r   r	   Ztwisted.pythonr
   Ztwisted.python.compatr   Ztwisted.python.filepathr   Ztwisted.test.test_tcpr   r   Ztwisted.trialZClientFactoryr   r‰   ZTestCaser    ZConnectedDatagramProtocolr‹   ÚDatagramProtocolr�   ZIReactorUNIXDatagramr¤   r   r   r   r   Ú<module>   s2   $þ `þ