
    Vh5                        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ej0                        Z e e
j4                  ed       d       G d dej6                               Z G d dej:                        Z G d dej>                        Z  e e
jB                  ed       d       G d dej6                               Z"y)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 Zd Zy)FailedConnectionClientFactoryc                     || _         y N)onFail)selfr   s     F/home/dcms/DCMS/lib/python3.12/site-packages/twisted/test/test_unix.py__init__z&FailedConnectionClientFactory.__init__   s	        c                 :    | j                   j                  |       y r   )r   errback)r   	connectorreasons      r   clientConnectionFailedz4FailedConnectionClientFactory.clientConnectionFailed   s    F#r   N)__name__
__module____qualname__r   r    r   r   r   r      s    $r   r   1This reactor does not support UNIX domain socketsc                   p    e Zd ZdZ ej
                  ed      sdZd Zd Z	d Z
d Zd Zd	 Zd
 Zd Zd Zy)UnixSocketTestsz
    Test unix sockets.
    Nr#   c                      j                         } j                         t               t        j                         x}_        t        j                  |      } j                  |j                         t        j                  t        j                  t        j                        } j                  |j                         |j                         |j                  |        fd}|j                  |       |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                     t        j                        }j                  j                  |g       j                  | j                  j                         |       y r   )r   UNIXAddressassertEqualpeerAddresses	transportgetPeer)protoexpectedpeernamer   serverFactorys     r   
cbConnMadez1UnixSocketTests.test_peerBind.<locals>.cbConnMade;   sH    **84H]888*EU__446Ar   )mktempr   r   DeferredprotocolConnectionMader	   
listenUNIX
addCleanupstopListeningsocketAF_UNIXSOCK_STREAMclosebindconnectaddCallback)r   filenameconnMadeunixPort
unixSocketr1   r/   r0   s   `     @@r   test_peerBindzUnixSocketTests.test_peerBind*   s     ;;=;;='):?..:JJ=7%%h>../]]6>>63E3EF

(()!8$	B
 	Z(r   c                      j                         t               }t        j                         }||_        t        j                  |      } j                  |j                         t               t        j                         }|_        t        j                         t        j                  ||g      } fd}|j                  |       |S )z
        L{IReactorUNIX.connectUNIX} can be used to connect a client to a server
        started with L{IReactorUNIX.listenUNIX}.
        c                     | \  }}j                  j                  t        j                        g       |j                  j                          |j                  j                          y r   )r)   r*   r   r(   r+   loseConnection)argsserverProtocolclientProtocolclientFactoryr?   r   s      r   allConnectedz1UnixSocketTests.test_dumber.<locals>.allConnectedT   s\    -1*NN ++g.A.A(.K-L $$335$$335r   )r2   r   r   r3   r4   r	   r5   r6   r7   r   connectUNIXgatherResultsr>   )	r   r0   serverConnMaderA   clientConnMadedrK   rJ   r?   s	   `      @@r   test_dumberzUnixSocketTests.test_dumberC   s    
 ;;='))/=,%%h>../'))/=,Hm4 @A		6 	
l#r   c                    	  j                         t               }t        j                         }||_        t        j                  |d      	 j                  t        j                  dz                t               t        j                         }|_        t        j                  d       t        j                  ||g      } 	fd}|j                  |        fd}|j                  |       |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                     | \  }}j                  j                  t        j                        g       |j                  j                          |j                  j                          j                         S r   )r)   r*   r   r(   r+   rF   r7   )rG   rH   clientProtorJ   r?   r   rA   s      r   
_portStuffz0UnixSocketTests.test_pidFile.<locals>._portStuffy   sk    *.'NK ++g.A.A(.K-L !!002$$335))++r   c                 V    j                  t        j                  dz         d       y )NrU   locked)assertFalser   isLocked)ignoredr?   r   s    r   _checkz,UnixSocketTests.test_pidFile.<locals>._check   s#    X..x'/ABHMr   )r2   r   r   r3   r4   r	   r5   
assertTruer   r_   r   rL   rM   r>   )
r   r0   rN   rO   rP   r[   ra   rJ   r?   rA   s
   `      @@@r   test_pidFilezUnixSocketTests.test_pidFileb   s     ;;='))/=,%%htL))(W*<=>
 ())/=,Hma@ @A	, 	
j!	N 	
fr   c                    | j                         t               t        j                  d      }| j	                  t
        j                  t        j                  d       fd}|j                         j                  |      S )z
        L{IReactorUNIX.listenUNIX} raises L{error.CannotListenError} if passed
        the name of a file on which a server is already listening.
        TrS   c                 T    t        j                  d      }|j                         S NTrS   )r	   r5   r7   )ignrA   r?   r0   s     r   stoppedListeningz<UnixSocketTests.test_socketLocking.<locals>.stoppedListening   s&    ))(M4PH))++r   )	r2   r   r	   r5   assertRaisesr   CannotListenErrorr7   r>   )r   rA   rh   r?   r0   s      @@r   test_socketLockingz"UnixSocketTests.test_socketLocking   s|    
 ;;=')%%htL## 	 	
	, %%'334DEEr   c                    | j                         | _        t        d| j                  d      }dt        t        j
                  j                  t        j                              j                         j                  i}t        t        j                        j                         j                  }t        j                  |dd|f|      }|j                  |       |S )NzBfrom twisted.internet import protocol, reactor
reactor.listenUNIX(z), protocol.ServerFactory(),wantPID=True)
s
   PYTHONPATHs   -us   -c)env)r2   r?   r   r   ospathsepjoinsyspathasBytesMode
executabler
   getProcessValuer>   )r   callbacksourcerm   pyExerP   s         r   _uncleanSocketTestz"UnixSocketTests._uncleanSocketTest   s     }}
 hrzzsxx'@AMMOTTU(446;;!!%%)?SI	hr   c                 0      fd} j                  |      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                 x    t        j                  j                  t               d      }|j	                         S rf   )r	   r5   r?   r   r7   )rg   pr   s     r   ranStupidChildzGUnixSocketTests.test_uncleanServerSocketLocking.<locals>.ranStupidChild   s+    ""4==/2CTRA??$$r   ry   r   r}   s   ` r   test_uncleanServerSocketLockingz/UnixSocketTests.test_uncleanServerSocketLocking   s    	%
 &&~66r   c                 0      fd} j                  |      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                     t        j                         }t        |      }t        j                  j
                  |d       j                  |t        j                        S )NTrW   )	r   r3   r   r	   rL   r?   assertFailurer   BadFileError)rg   rP   fr   s      r   r}   zCUnixSocketTests.test_connectToUncleanServer.<locals>.ranStupidChild   sH     A-a0Aq4@%%a););<<r   r~   r   s   ` r   test_connectToUncleanServerz+UnixSocketTests.test_connectToUncleanServer   s    	= &&~66r   c                 D     j                         }t        j                  ||      d d|d} j                  t	              |        j                  t              |       t        j                  j                        } fd}|j                  |       |S )z~
        Test the C{__str__} and C{__repr__} implementations of a UNIX port when
        used with the given factory.
        < on >c                 ~    d d}j                  t              |       j                  t              |       y Nr   z (not listening)>r)   reprstr)rg   unconnectedStringfactoryNamer   rA   s     r   rh   z3UnixSocketTests._reprTest.<locals>.stoppedListening   s=    "#K=0A BT(^->?S],=>r   )
r2   r	   r5   r)   r   r   r   maybeDeferredr7   r>   )r   r0   r   r?   connectedStringrP   rh   rA   s   ` `    @r   	_reprTestzUnixSocketTests._reprTest   s    
 ;;=%%h>k]$xl!<h9X8 6 67	?
 	
&'r   c                 p     G d d      }| j                  |t               | j                   |       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                       e Zd Zd Zd Zy)EUnixSocketTests.test_reprWithNewStyleFactory.<locals>.NewStyleFactoryc                      y r   r"   r   s    r   doStartzMUnixSocketTests.test_reprWithNewStyleFactory.<locals>.NewStyleFactory.doStart       r   c                      y r   r"   r   s    r   doStopzLUnixSocketTests.test_reprWithNewStyleFactory.<locals>.NewStyleFactory.doStop   r   r   N)r   r    r!   r   r   r"   r   r   NewStyleFactoryr          r   r   z&twisted.test.test_unix.NewStyleFactoryassertIsInstancetyper   )r   r   s     r   test_reprWithNewStyleFactoryz,UnixSocketTests.test_reprWithNewStyleFactory   s8    	 	 	ot4~~G
 	
r   )r   r    r!   __doc__r   IReactorUNIXr	   skiprC   rQ   rc   rk   ry   r   r   r   r   r"   r   r   r%   r%      sO    
 #:""7D1B2>*XF."77,
r   r%   c                   0    e Zd ZdxZZdZd Zd Zd Zd Z	y)ClientProtoFNc                 h    t        j                         | _        t        j                         | _        y r   )r   r3   deferredStarteddeferredGotBackr   s    r   r   zClientProto.__init__       $~~/$~~/r   c                     d| _         y NTstoppedr   s    r   stopProtocolzClientProto.stopProtocol	  	    r   c                 H    d| _         | j                  j                  d        y r   startedr   rv   r   s    r   startProtocolzClientProto.startProtocol      %%d+r   c                 H    || _         | j                  j                  d        y r   )gotbackr   rv   )r   datas     r   datagramReceivedzClientProto.datagramReceived  r   r   )
r   r    r!   r   r   r   r   r   r   r   r"   r   r   r   r     s%    GgG0,,r   r   c                   4    e Zd ZdxZZdxZZd Zd Zd Z	d Z
y)ServerProtoFNc                 h    t        j                         | _        t        j                         | _        y r   )r   r3   r   deferredGotWhatr   s    r   r   zServerProto.__init__  r   r   c                     d| _         y r   r   r   s    r   r   zServerProto.stopProtocol  r   r   c                 H    d| _         | j                  j                  d        y r   r   r   s    r   r   zServerProto.startProtocol   r   r   c                     || _         | j                  j                  d|       || _        | j                  j                  d        y )N   hi back)gotfromr+   writegotwhatr   rv   )r   r   addrs      r   r   zServerProto.datagramReceived$  s8    Z.%%d+r   )r   r    r!   r   r   r   r   r   r   r   r   r"   r   r   r   r     s*    GgGg0,,r   r   z3This reactor does not support UNIX datagram socketsc                   (    e Zd ZdZd Zd Zd Zd Zy)DatagramUnixSocketTestsz%
    Test datagram UNIX sockets.
    c                    	  j                          j                         }t               	t               t        j                  |	      } j                  |j                         t        j                  |      } j                  |j                         t        j                  	j                  j                  g      }	fd} 	fd}|j                  |       |j                  |       |S )zf
        Test that a datagram can be sent to and received by a server and vice
        versa.
        )bindAddressc                     j                   j                  d       t        j                  j                  j
                  g      S )N   hi)r+   r   r   rM   r   r   )r`   cpsps    r   r   z4DatagramUnixSocketTests.test_exchange.<locals>.writeD  s7    LLu%&&(:(:B<N<N'OPPr   c                     j                  dj                         j                  j                         j                  dj                         y )Nr   r   )r)   r   r   r   )r`   
clientaddrr   r   r   s    r   _cbTestExchangez>DatagramUnixSocketTests.test_exchange.<locals>._cbTestExchangeH  s?    UBJJ/Z4Z4r   )r2   r   r   r	   listenUNIXDatagramr6   r7   connectUNIXDatagramr   rM   r   r>   )
r   
serveraddrscrP   r   r   r   r   r   s
   `      @@@r   test_exchangez%DatagramUnixSocketTests.test_exchange4  s    
 [[]
[[]
]]&&z26(''
BJO(!3!3R5G5G HI	Q	5
 	
e	o&r   c                    | j                         }t               }t        j                  ||      }| j	                  t
        j                  t        j                  ||       |j                          t        j                  |       y)z
        L{IReactorUNIXDatagram.listenUNIXDatagram} raises
        L{error.CannotListenError} if the unix socket specified is already in
        use.
        N)
r2   r   r	   r   ri   r   rj   r7   rn   unlink)r   r   r|   r   s       r   test_cannotListenz)DatagramUnixSocketTests.test_cannotListenQ  s^     {{}M&&tQ/%1173M3MtUVW	
		$r   c                 D     j                         }t        j                  ||      d d|d} j                  t	              |        j                  t              |       t        j                  j                        } fd}|j                  |       |S )z
        Test the C{__str__} and C{__repr__} implementations of a UNIX datagram
        port when used with the given protocol.
        r   r   r   c                 ~    d d}j                  t              |       j                  t              |       y r   r   )rg   r   protocolNamer   rA   s     r   rh   z;DatagramUnixSocketTests._reprTest.<locals>.stoppedListeningn  s=    "#L>1B CT(^->?S],=>r   )
r2   r	   r   r)   r   r   r   r   r7   r>   )r   serverProtor   r?   r   stopDeferredrh   rA   s   ` `    @r   r   z!DatagramUnixSocketTests._reprTest`  s    
 ;;=--hDl^4|1=h9X8**8+A+AB	?
 	  !12r   c                 p     G d d      }| j                  |t               | j                   |       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                       e Zd Zd Zd Zy)ODatagramUnixSocketTests.test_reprWithNewStyleProtocol.<locals>.NewStyleProtocolc                      y r   r"   )r   r+   s     r   makeConnectionz^DatagramUnixSocketTests.test_reprWithNewStyleProtocol.<locals>.NewStyleProtocol.makeConnection  r   r   c                      y r   r"   r   s    r   r   zVDatagramUnixSocketTests.test_reprWithNewStyleProtocol.<locals>.NewStyleProtocol.doStop  r   r   N)r   r    r!   r   r   r"   r   r   NewStyleProtocolr   ~  r   r   r   z'twisted.test.test_unix.NewStyleProtocolr   )r   r   s     r   test_reprWithNewStyleProtocolz5DatagramUnixSocketTests.test_reprWithNewStyleProtocolv  s9    	 	 	.5~~ I
 	
r   N)r   r    r!   r   r   r   r   r   r"   r   r   r   r   +  s    
:,
r   r   )#r   rn   r8   rq   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   <module>r      s   
 
  
  X X X # / , B "$H$:$: $ 
..7\
h'' \
	\
~,(44 ,(,(++ ,, '
''669[
h// [
	[
r   