o
    >hO@                     @   sB  d Z ddlZddlmZ ddlmZ ddlmZmZ ddl	m
Z
mZ ddlmZmZ ddlmZmZmZmZ dd	lmZ dd
lmZmZ ddlmZ ddlmZ ddlmZmZ ddl m!Z!m"Z" ddl#m$Z$ ddl%m&Z& G dd deZ'G dd dZ(G dd dee(e'Z)G dd dee(e'Z*e+ ,e)-  e+ ,e*-  dS )zU
Tests for implementations of L{IReactorUDP} and the UDP parts of
L{IReactorSocket}.
    N)implementer)verifyObject)defererror)IPv4AddressIPv6Address)DeferredmaybeDeferred)IListeningPortILoggingContextIReactorSocketIReactorUDP)DatagramProtocol)LogObserverMixinfindFreePort)ReactorBuilder)context)ILogContexterr)
GoodClientServer)skipWithoutIPv6)SkipTestc                   @   s(   e Zd ZdZdd Zdd Zdd ZdS )	DatagramTransportTestsMixinzP
    Mixin defining tests which apply to any port/datagram based transport.
    c                 C   sd   |   }|  }ttG dd dt}| }| ||}d| jf }| |f|d d  dS )zu
        When a port starts, a message including a description of the associated
        protocol is logged.
        c                   @   s   e Zd Zdd ZdS )zQDatagramTransportTestsMixin.test_startedListeningLogMessage.<locals>.SomeProtocolc                 S   s   dS )NzCrazy Protocol selfr   r   ~/var/www/vedio/testing/chatpythonscript.ninositsolution.com/env/lib/python3.10/site-packages/twisted/internet/test/test_udp.py	logPrefix1   s   z[DatagramTransportTestsMixin.test_startedListeningLogMessage.<locals>.SomeProtocol.logPrefixN)__name__
__module____qualname__r   r   r   r   r   SomeProtocol/   s    r"   zCrazy Protocol starting on %dr   messageN)	observebuildReactorr   r   r   getListeningPortgetHostportassertEqual)r   loggedMessagesreactorr"   protocolpexpectedMessager   r   r   test_startedListeningLogMessage'   s   z;DatagramTransportTestsMixin.test_startedListeningLogMessagec                    sz   |    |  | t d j d}fdd fdd}| |  | |f d d  d	S )
z
        When a connection is lost a message is logged containing an
        address identifying the port and the fact that it was closed.
        z
(UDP Port z Closed)c                    s       d S Nstopignoredr+   r   r   stopReactorD      zNDatagramTransportTestsMixin.test_connectionLostLogMessage.<locals>.stopReactorc                      s    d d = t j d S r0   )r	   stopListeningaddCallbackr   )r*   r-   r6   r   r   doStopListeningG   s   
zRDatagramTransportTestsMixin.test_connectionLostLogMessage.<locals>.doStopListeningr   r#   N)	r$   r%   r&   r   r'   r(   callWhenRunning
runReactorr)   )r   r.   r:   r   )r*   r-   r+   r6   r   test_connectionLostLogMessage:   s   

z9DatagramTransportTestsMixin.test_connectionLostLogMessagec                    s`   G  fdddt }|   | }|  | |   | |j | |j | |j dS )z
        L{DatagramProtocol.stopProtocol} is called asynchronously (ie, not
        re-entrantly) when C{stopListening} is used to stop the datagram
        transport.
        c                       s0   e Zd ZdZdZdZdZdd Z fddZdS )zVDatagramTransportTestsMixin.test_stopProtocolScheduling.<locals>.DisconnectingProtocolFc                 S   s    d| _ d| _| j  d| _d S )NTF)startedinStartProtocol	transportr8   r   r   r   r   startProtocol]   s   

zdDatagramTransportTestsMixin.test_stopProtocolScheduling.<locals>.DisconnectingProtocol.startProtocolc                    s   d| _ | j| _   d S )NT)stoppedr?   stoppedInStartr2   r   r5   r   r   stopProtocolc   s   zcDatagramTransportTestsMixin.test_stopProtocolScheduling.<locals>.DisconnectingProtocol.stopProtocolN)	r   r    r!   r>   rB   r?   rC   rA   rD   r   r5   r   r   DisconnectingProtocolW   s    rE   N)	r   r%   r&   r<   
assertTruer>   rB   assertFalserC   )r   rE   r,   r   r5   r   test_stopProtocolSchedulingP   s   
z7DatagramTransportTestsMixin.test_stopProtocolSchedulingN)r   r    r!   __doc__r/   r=   rH   r   r   r   r   r   "   s
    r   c                   @   s   e Zd ZdZdd Zdd Zedd Zdd	 Zd
d Z	dd Z
dd Zdd Zedd Zedd Zdd Zedd Zedd Zdd Zdd Zd S )!UDPPortTestsMixinzY
    Tests for L{IReactorUDP.listenUDP} and
    L{IReactorSocket.adoptDatagramPort}.
    c                 C   s*   |   }| |t }| tt| dS )zY
        L{IReactorUDP.listenUDP} returns an object providing L{IListeningPort}.
        N)r%   r&   r   rF   r   r
   r   r+   r(   r   r   r   test_interfacex   s   z UDPPortTestsMixin.test_interfacec                 C   sH   t tjd\}}|  }| j|t ||d}| | td|| dS )z
        L{IListeningPort.getHost} returns an L{IPv4Address} giving a
        dotted-quad of the IPv4 address the port is listening on as well as
        the port number.
        )type)r(   	interfaceUDPN)	r   socket
SOCK_DGRAMr%   r&   r   r)   r'   r   )r   host
portNumberr+   r(   r   r   r   test_getHost   s   
zUDPPortTestsMixin.test_getHostc                 C   s@   |   }| j|t dd}| }| |jd | |t dS )zr
        L{IListeningPort.getHost} returns an L{IPv6Address} when listening on
        an IPv6 interface.
        ::1rN   N)r%   r&   r   r'   r)   rR   assertIsInstancer   )r   r+   r(   addrr   r   r   test_getHostIPv6   s
   z"UDPPortTestsMixin.test_getHostIPv6c                 C   s&   |   }| jtj|jt ddd dS )z
        An L{InvalidAddressError} is raised when trying to listen on an address
        that isn't a valid IPv4 or IPv6 address.
        r   zexample.comrV   N)r%   assertRaisesr   InvalidAddressError	listenUDPr   )r   r+   r   r   r   test_invalidInterface   s   
z'UDPPortTestsMixin.test_invalidInterfacec                    s   G dd dt }  |d}|j} |}| }fdd}|| |t | fdd |dd	|j	f 
  d
S )z
        Datagram transports implement L{ILoggingContext.logPrefix} to return a
        message reflecting the protocol they are running.
        c                   @   s$   e Zd Zdd Zdd Zdd ZdS )zIUDPPortTestsMixin.test_logPrefix.<locals>.CustomLogPrefixDatagramProtocolc                 S   s   || _ t | _d S r0   )_prefixr   system)r   prefixr   r   r   __init__   s   zRUDPPortTestsMixin.test_logPrefix.<locals>.CustomLogPrefixDatagramProtocol.__init__c                 S   s   | j S r0   )r^   r   r   r   r   r      s   zSUDPPortTestsMixin.test_logPrefix.<locals>.CustomLogPrefixDatagramProtocol.logPrefixc                 S   s2   | j d ur| j }d | _ |ttd  d S d S )Nr_   )r_   callbackr   getr   )r   bytesrX   r_   r   r   r   datagramReceived   s
   
zZUDPPortTestsMixin.test_logPrefix.<locals>.CustomLogPrefixDatagramProtocol.datagramReceivedN)r   r    r!   ra   r   re   r   r   r   r   CustomLogPrefixDatagramProtocol   s    rf   zCustom Datagramsc                    s     d|  d S )NzCustom Datagrams (UDP))r)   )r_   r   r   r   	gotSystem      z3UDPPortTestsMixin.test_logPrefix.<locals>.gotSystemc                          S r0   r1   r3   r5   r   r   <lambda>       z2UDPPortTestsMixin.test_logPrefix.<locals>.<lambda>s
   some bytes	127.0.0.1N)r   r%   r_   r&   r'   r9   
addErrbackr   writer(   r<   )r   rf   r,   dr(   addressrg   r   )r+   r   r   test_logPrefix   s   

z UDPPortTestsMixin.test_logPrefixc                    s   G dd dt } | }|j}|}| }d  fdd}|| |t |fdd | d|j	f 
 d	S )
zH
        Write a sequence of L{bytes} to a L{DatagramProtocol}.
        c                   @   s   e Zd Zdd Zdd ZdS )zDUDPPortTestsMixin.test_writeSequence.<locals>.SimpleDatagramProtocolc                 S   s   t  | _d S r0   )r   r   r   r   r   r   ra      r7   zMUDPPortTestsMixin.test_writeSequence.<locals>.SimpleDatagramProtocol.__init__c                 S   s   | j | d S r0   )r   rb   )r   datarX   r   r   r   re      rh   zUUDPPortTestsMixin.test_writeSequence.<locals>.SimpleDatagramProtocol.datagramReceivedN)r   r    r!   ra   re   r   r   r   r   SimpleDatagramProtocol   s    rs   )s   somes   bytess   tos   writec                    s    d |  d S )N    )r)   join)rr   )dataToWriter   r   r   gotData   s   z5UDPPortTestsMixin.test_writeSequence.<locals>.gotDatac                    ri   r0   r1   r3   r5   r   r   rj      rk   z6UDPPortTestsMixin.test_writeSequence.<locals>.<lambda>rl   N)r   r%   r   r&   r'   r9   rm   r   writeSequencer(   r<   )r   rs   r,   r   r(   rp   rw   r   )rv   r+   r   r   test_writeSequence   s   

z$UDPPortTestsMixin.test_writeSequencec                 C   s4   |   }| |t }| t| jt| dS )zQ
        C{str()} on the listening port object includes the port number.
        N)r%   r&   r   assertInstrr'   r(   rK   r   r   r   test_str      zUDPPortTestsMixin.test_strc                 C   s4   |   }| |t }| t| jt| dS )zR
        C{repr()} on the listening port object includes the port number.
        N)r%   r&   r   rz   reprr'   r(   r{   rK   r   r   r   	test_repr   r}   zUDPPortTestsMixin.test_reprc                       |   t t  }_| jdd t  t  } _| j dd  j } fdd}fdd}t	||g}|
| |
| |t |  jd }| |d|j|jff d	S )
zS
        Writing to an IPv6 UDP socket on the loopback interface succeeds.
        rU   rV   c                    s,    j ddj  jf t  }_|S )
            Send a datagram from the client once it's started.

            @param ignored: a list of C{[None, None]}, which is ignored
            @returns: a deferred which fires when the server has received a
                datagram.
               spamrU   )r@   rn   r'   r(   r   r   packetReceivedr4   serverReceivedclientserverr   r   cbClientStarted  s   zDUDPPortTestsMixin.test_writeToIPv6Interface.<locals>.cbClientStartedc                           dS z
            Stop the reactor after a datagram is received.

            @param ignored: L{None}, which is ignored
            @returns: L{None}
            Nr1   r3   r5   r   r   cbServerReceived  s   zEUDPPortTestsMixin.test_writeToIPv6Interface.<locals>.cbServerReceivedr   r   Nr%   r   r   r   startedDeferredr&   r   r@   r'   gatherResultsr9   rm   r   r<   packetsr)   rR   r(   r   serverStartedclientStartedcAddrr   r   ro   packetr   r   r+   r   r   test_writeToIPv6Interface   s"   
	




z+UDPPortTestsMixin.test_writeToIPv6Interfacec                    r   )
z
        An IPv6 address can be passed as the C{interface} argument to
        L{listenUDP}. The resulting Port accepts IPv6 datagrams.
        rU   rV   c                    s4    j dj  j  j d t  }_|S )r   rU   r   )r@   connectr'   r(   rn   r   r   r   r   r   r   r   r   4  s   	zMUDPPortTestsMixin.test_connectedWriteToIPv6Interface.<locals>.cbClientStartedc                    r   r   r1   r3   r5   r   r   r   B  s   zNUDPPortTestsMixin.test_connectedWriteToIPv6Interface.<locals>.cbServerReceivedr   r   Nr   r   r   r   r   "test_connectedWriteToIPv6Interface$  s"   






z4UDPPortTestsMixin.test_connectedWriteToIPv6Interfacec                 C   .   |   }| |t }| tj|jdd dS )zn
        Writing to a hostname instead of an IP address will raise an
        L{InvalidAddressError}.
        spam)example.invalid   Nr%   r&   r   rZ   r   r[   rn   rK   r   r   r   /test_writingToHostnameRaisesInvalidAddressErrorU  s
   zAUDPPortTestsMixin.test_writingToHostnameRaisesInvalidAddressErrorc                 C   2   |   }| j|t dd}| tj|jdd dS )l
        Writing to an IPv6 address on an IPv4 socket will raise an
        L{InvalidAddressError}.
        rl   rV   r   )rU   r   Nr   rK   r   r   r   1test_writingToIPv6OnIPv4RaisesInvalidAddressError`  s   zCUDPPortTestsMixin.test_writingToIPv6OnIPv4RaisesInvalidAddressErrorc                 C   r   )r   rU   rV   r   )rl   r   Nr   rK   r   r   r   1test_writingToIPv4OnIPv6RaisesInvalidAddressErrorj  s
   zCUDPPortTestsMixin.test_writingToIPv4OnIPv6RaisesInvalidAddressErrorc                 C   r   )zq
        Connecting to a hostname instead of an IP address will raise an
        L{InvalidAddressError}.
        r   r   N)r%   r&   r   rZ   r   r[   r   rK   r   r   r   2test_connectingToHostnameRaisesInvalidAddressErrorv  s   zDUDPPortTestsMixin.test_connectingToHostnameRaisesInvalidAddressErrorc                 C   s2   |   }| |t }|d | |  dS )zk
        L{IListeningPort.setBroadcastAllowed} sets broadcast to be allowed
        on the socket.
        TN)r%   r&   r   setBroadcastAllowedrF   getBroadcastAllowedrK   r   r   r   test_allowBroadcast  s   
z%UDPPortTestsMixin.test_allowBroadcastN)r   r    r!   rI   rL   rT   r   rY   r]   rq   ry   r|   r   r   r   r   r   r   r   r   r   r   r   r   rJ   r   s,    
$
,
0
	
	rJ   c                   @   "   e Zd ZdZefZ	dddZdS )	UDPServerTestsBuilderzM
    Run L{UDPPortTestsMixin} tests using newly created UDP
    sockets.
    r        c                 C   s   |j ||||dS )aB  
        Get a UDP port from a reactor.

        @param reactor: A reactor used to build the returned
            L{IListeningPort} provider.
        @type reactor: L{twisted.internet.interfaces.IReactorUDP}

        @see: L{twisted.internet.IReactorUDP.listenUDP} for other
            argument and return types.
        )rN   maxPacketSize)r\   )r   r+   r,   r(   rN   r   r   r   r   r&     s   z&UDPServerTestsBuilder.getListeningPortNr   r   r   )r   r    r!   rI   r   requiredInterfacesr&   r   r   r   r   r     s
    r   c                   @   r   )	UDPFDServerTestsBuilderzC
    Run L{UDPPortTestsMixin} tests using adopted UDP sockets.
    r   r   r   c           	      C   s   t |rMd|v rtj}t||d d }ntj}||f}t|tj}|| |d z|	|
 |j||W |
  |  S |
  |  w td)a  
        Get a UDP port from a reactor, wrapping an already-initialized file
        descriptor.

        @param reactor: A reactor used to build the returned
            L{IListeningPort} provider.
        @type reactor: L{twisted.internet.interfaces.IReactorSocket}

        @param port: A port number to which the adopted socket will be
            bound.
        @type port: C{int}

        @param interface: The local IPv4 or IPv6 address to which the
            adopted socket will be bound.  defaults to '', ie all IPv4
            addresses.
        @type interface: C{str}

        @see: L{twisted.internet.IReactorSocket.adoptDatagramPort} for other
            argument and return types.
        :r      Fz'Reactor does not provide IReactorSocket)r   
providedByrP   AF_INET6getaddrinfoAF_INETrQ   bindsetblockingadoptDatagramPortfilenofamilycloser   )	r   r+   r,   r(   rN   r   domainrp   portSockr   r   r   r&     s$   




z(UDPFDServerTestsBuilder.getListeningPortNr   )r   r    r!   rI   r   r   r&   r   r   r   r   r     s
    r   ).rI   rP   zope.interfacer   zope.interface.verifyr   twisted.internetr   r   twisted.internet.addressr   r   twisted.internet.deferr   r	   twisted.internet.interfacesr
   r   r   r   twisted.internet.protocolr   &twisted.internet.test.connectionmixinsr   r   #twisted.internet.test.reactormixinsr   twisted.pythonr   twisted.python.logr   r   twisted.test.test_udpr   r   twisted.test.testutilsr   twisted.trial.unittestr   r   rJ   r   r   globalsupdatemakeTestCaseClassesr   r   r   r   <module>   s8   P  

9