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   $ `