o
    ¯bñA  ã                   @   sL  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„ Z%e%ƒ Z&dd„ 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)ÚSkipTestc                  C   sJ   d} d}zt   t j¡} |  d¡ d}W n	 ty   Y nw | r#|  ¡  |S )z4Returns True if the system can bind an IPv6 address.NF)ú::1r   T)ÚsocketÚAF_INET6ÚbindÚOSErrorÚclose)ÚsockÚhas_ipv6© r    ú@/usr/lib/python3/dist-packages/twisted/internet/test/test_udp.pyÚ	_has_ipv6!   s   
ÿr"   c                 C   s   t sd| _| S )Nz.Does not work on systems without IPv6 support.)ÚHAS_IPV6Úskip)Úfr    r    r!   ÚskipWithoutIPv65   s   r&   c                   @   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 Protocolr    ©Úselfr    r    r!   Ú	logPrefixJ   s   z[DatagramTransportTestsMixin.test_startedListeningLogMessage.<locals>.SomeProtocol.logPrefixN)Ú__name__Ú
__module__Ú__qualname__r*   r    r    r    r!   ÚSomeProtocolH   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©r7   r    r!   ÚstopReactor]   ó   zNDatagramTransportTestsMixin.test_connectionLostLogMessage.<locals>.stopReactorc                      s   ˆ d d …= t ˆjƒ ˆ¡ d S r<   )r	   ÚstopListeningÚaddCallbackr    )r6   r9   rB   r    r!   ÚdoStopListening`   s   
zRDatagramTransportTestsMixin.test_connectionLostLogMessage.<locals>.doStopListeningr   r/   N)	r0   r1   r2   r   r3   r4   ÚcallWhenRunningÚ
runReactorr5   )r)   r:   rF   r    )r6   r9   r7   rB   r!   Útest_connectionLostLogMessageS   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Ú	transportrD   r(   r    r    r!   ÚstartProtocolw   s   

zdDatagramTransportTestsMixin.test_stopProtocolScheduling.<locals>.DisconnectingProtocol.startProtocolc                    s   d| _ | j| _ˆ  ¡  d S )NT)ÚstoppedrK   ÚstoppedInStartr>   r(   rA   r    r!   ÚstopProtocol}   s   zcDatagramTransportTestsMixin.test_stopProtocolScheduling.<locals>.DisconnectingProtocol.stopProtocolN)	r+   r,   r-   rJ   rN   rK   rO   rM   rP   r    rA   r    r!   ÚDisconnectingProtocolp   s    rQ   N)	r   r1   r2   rH   Ú
assertTruerJ   rN   ÚassertFalserO   )r)   rQ   r8   r    rA   r!   Útest_stopProtocolSchedulingi   s   
z7DatagramTransportTestsMixin.test_stopProtocolSchedulingN)r+   r,   r-   Ú__doc__r;   rI   rT   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)r1   r2   r   rR   r   r
   ©r)   r7   r4   r    r    r!   Útest_interface’   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)r4   Ú	interfaceÚUDPN)	r   r   Ú
SOCK_DGRAMr1   r2   r   r5   r3   r   )r)   ÚhostÚ
portNumberr7   r4   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.
        r   ©rZ   N)r1   r2   r   r3   r5   r]   ÚassertIsInstancer   )r)   r7   r4   Ú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.comr`   N)r1   ÚassertRaisesr   ÚInvalidAddressErrorÚ	listenUDPr   )r)   r7   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 r<   )Ú_prefixr   Úsystem)r)   Úprefixr    r    r!   Ú__init__È   s   zRUDPPortTestsMixin.test_logPrefix.<locals>.CustomLogPrefixDatagramProtocol.__init__c                 S   s   | j S r<   )rh   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 )Nri   )ri   Úcallbackr   Úgetr   )r)   Úbytesrb   ri   r    r    r!   ÚdatagramReceivedÏ   s
   
ýzZUDPPortTestsMixin.test_logPrefix.<locals>.CustomLogPrefixDatagramProtocol.datagramReceivedN)r+   r,   r-   rk   r*   ro   r    r    r    r!   ÚCustomLogPrefixDatagramProtocolÇ   s    rp   zCustom Datagramsc                    s   ˆ   d| ¡ d S )NzCustom Datagrams (UDP))r5   )ri   r(   r    r!   Ú	gotSystemÛ   ó   z3UDPPortTestsMixin.test_logPrefix.<locals>.gotSystemc                    ó   ˆ   ¡ S r<   r=   r?   rA   r    r!   Ú<lambda>à   ó    z2UDPPortTestsMixin.test_logPrefix.<locals>.<lambda>s
   some bytesú	127.0.0.1N)r   r1   ri   r2   r3   rE   Ú
addErrbackr   Úwriter4   rH   )r)   rp   r8   Údr4   Úaddressrq   r    )r7   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 r<   )r   r   r(   r    r    r!   rk   ë   rC   zMUDPPortTestsMixin.test_writeSequence.<locals>.SimpleDatagramProtocol.__init__c                 S   s   | j  |¡ d S r<   )r   rl   )r)   Údatarb   r    r    r!   ro   î   rr   zUUDPPortTestsMixin.test_writeSequence.<locals>.SimpleDatagramProtocol.datagramReceivedN)r+   r,   r-   rk   ro   r    r    r    r!   ÚSimpleDatagramProtocolê   s    r}   )s   somes   bytess   tos   writec                    s   ˆ  d ˆ ¡| ¡ d S )Nó    )r5   Újoin)r|   )ÚdataToWriter)   r    r!   ÚgotDataø   s   z5UDPPortTestsMixin.test_writeSequence.<locals>.gotDatac                    rs   r<   r=   r?   rA   r    r!   rt   ý   ru   z6UDPPortTestsMixin.test_writeSequence.<locals>.<lambda>rv   N)r   r1   r   r2   r3   rE   rw   r   ÚwriteSequencer4   rH   )r)   r}   r8   r   r4   rz   r�   r    )r€   r7   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)r1   r2   r   ÚassertInÚstrr3   r4   rW   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)r1   r2   r   r„   Úreprr3   r4   r…   rW   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.
        r   r`   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.
            ó   spamr   )rL   rx   r3   r4   r   r   ÚpacketReceived©r@   Ú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}
            Nr=   r?   rA   r    r!   ÚcbServerReceived,  s   zEUDPPortTestsMixin.test_writeToIPv6Interface.<locals>.cbServerReceivedr   rŒ   N©r1   r   r   r   ÚstartedDeferredr2   r   rL   r3   ÚgatherResultsrE   rw   r   rH   Úpacketsr5   r]   r4   ©r)   ÚserverStartedÚclientStartedÚcAddrr“   r–   ry   Úpacketr    ©r‘   r7   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.
        r   r`   c                    s4   ˆ j  dˆj  ¡ j¡ ˆ j  d¡ t ¡  }ˆ_|S )r‹   r   rŒ   )rL   Úconnectr3   r4   rx   r   r   r�   rŽ   r�   r    r!   r“   N  s   	zMUDPPortTestsMixin.test_connectedWriteToIPv6Interface.<locals>.cbClientStartedc                    r”   r•   r=   r?   rA   r    r!   r–   \  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©r1   r2   r   rd   r   re   rx   rW   r    r    r!   Ú/test_writingToHostnameRaisesInvalidAddressErroro  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}.
        rv   r`   r¥   )r   r§   Nr¨   rW   r    r    r!   Ú1test_writingToIPv6OnIPv4RaisesInvalidAddressErrorz  s   zCUDPPortTestsMixin.test_writingToIPv6OnIPv4RaisesInvalidAddressErrorc                 C   rª   )r«   r   r`   r¥   )rv   r§   Nr¨   rW   r    r    r!   Ú1test_writingToIPv4OnIPv6RaisesInvalidAddressError„  s
   ÿzCUDPPortTestsMixin.test_writingToIPv4OnIPv6RaisesInvalidAddressErrorc                 C   r¤   )zq
        Connecting to a hostname instead of an IP address will raise an
        L{InvalidAddressError}.
        r¦   r§   N)r1   r2   r   rd   r   re   r¢   rW   r    r    r!   Ú2test_connectingToHostnameRaisesInvalidAddressError�  s   zDUDPPortTestsMixin.test_connectingToHostnameRaisesInvalidAddressErrorc                 C   s2   |   ¡ }|  |tƒ ¡}| d¡ |  | ¡ ¡ dS )zk
        L{IListeningPort.setBroadcastAllowed} sets broadcast to be allowed
        on the socket.
        TN)r1   r2   r   ÚsetBroadcastAllowedrR   ÚgetBroadcastAllowedrW   r    r    r!   Útest_allowBroadcast™  s   
z%UDPPortTestsMixin.test_allowBroadcastN)r+   r,   r-   rU   rX   r_   r&   rc   rg   r{   rƒ   r†   r‰   r¡   r£   r©   r¬   r­   r®   r±   r    r    r    r!   rV   Œ   s,    
$
,
0
	
	rV   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.
        )rZ   ÚmaxPacketSize)rf   )r)   r7   r8   r4   rZ   r¶   r    r    r!   r2   ®  s   ÿz&UDPServerTestsBuilder.getListeningPortN©r   r´   rµ   )r+   r,   r-   rU   r   ÚrequiredInterfacesr2   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   Ú
providedByr   r   ÚgetaddrinfoÚAF_INETr\   r   ÚsetblockingÚadoptDatagramPortÚfilenoÚfamilyr   r   )	r)   r7   r8   r4   rZ   r¶   Údomainrz   ÚportSockr    r    r!   r2   É  s$   


ÿ
ý
z(UDPFDServerTestsBuilder.getListeningPortNr·   )r+   r,   r-   rU   r   r¸   r2   r    r    r    r!   r¹   À  s
    ÿr¹   )/rU   r   Ú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.trial.unittestr   r"   r#   r&   r'   rV   r³   r¹   ÚglobalsÚupdateÚmakeTestCaseClassesr    r    r    r!   Ú<module>   s<   Q  
ÿ
ÿ9