o
    Òó·iO@  ã                   @   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   údC:\Users\psant\Downloads\PAF backend\school\venv\Lib\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(   Z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=   rA   r>   rB   r@   rC   r   r5   r   r   ÚDisconnectingProtocolW   s    rD   N)	r   r%   r&   r;   Ú
assertTruer=   rA   ZassertFalserB   )r   rD   r,   r   r5   r   Útest_stopProtocolSchedulingP   s   
z7DatagramTransportTestsMixin.test_stopProtocolSchedulingN)r   r    r!   Ú__doc__r/   r<   rF   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   rE   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(   Ú	interfaceZUDPN)	r   ÚsocketÚ
SOCK_DGRAMr%   r&   r   r)   r'   r   )r   ÚhostZ
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©rL   N)r%   r&   r   r'   r)   rO   Z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.comrR   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   )rY   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 )NrZ   )rZ   Úcallbackr   Úgetr   )r   ÚbytesrS   rZ   r   r   r   ÚdatagramReceivedµ   s
   
ýzZUDPPortTestsMixin.test_logPrefix.<locals>.CustomLogPrefixDatagramProtocol.datagramReceivedN)r   r    r!   r\   r   r`   r   r   r   r   ÚCustomLogPrefixDatagramProtocol­   s    ra   zCustom Datagramsc                    s   ˆ   d| ¡ d S )NzCustom Datagrams (UDP))r)   )rZ   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%   rZ   r&   r'   r9   Ú
addErrbackr   Úwriter(   r;   )r   ra   r,   Údr(   Úaddressrb   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   r\   Ñ   r7   zMUDPPortTestsMixin.test_writeSequence.<locals>.SimpleDatagramProtocol.__init__c                 S   s   | j  |¡ d S r0   )r   r]   )r   ÚdatarS   r   r   r   r`   Ô   rc   zUUDPPortTestsMixin.test_writeSequence.<locals>.SimpleDatagramProtocol.datagramReceivedN)r   r    r!   r\   r`   r   r   r   r   ÚSimpleDatagramProtocolÐ   s    rn   )s   somes   bytess   tos   writec                    s   ˆ  d ˆ ¡| ¡ d S )Nó    )r)   Újoin)rm   )ÚdataToWriter   r   r   ÚgotDataÞ   s   z5UDPPortTestsMixin.test_writeSequence.<locals>.gotDatac                    rd   r0   r1   r3   r5   r   r   re   ã   rf   z6UDPPortTestsMixin.test_writeSequence.<locals>.<lambda>rg   N)r   r%   r   r&   r'   r9   rh   r   ZwriteSequencer(   r;   )r   rn   r,   r   r(   rk   rr   r   )rq   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(   rI   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   rt   Úreprr'   r(   ru   rI   r   r   r   Ú	test_reprï   rw   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.
        rQ   rR   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.
            ó   spamrQ   N)r?   ri   r'   r(   r   r   ÚpacketReceived©r4   Z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   ZstartedDeferredr&   r   r?   r'   ZgatherResultsr9   rh   r   r;   Zpacketsr)   rO   r(   ©r   ZserverStartedZclientStartedZcAddrr‚   r…   rj   Úpacketr   ©r€   r+   r�   r   Útest_writeToIPv6Interface÷   s"   
	




z+UDPPortTestsMixin.test_writeToIPv6Interfacec                    rz   )
z�
        An IPv6 address can be passed as the C{interface} argument to
        L{listenUDP}. The resulting Port accepts IPv6 datagrams.
        rQ   rR   c                    s4   ˆ j  dˆj  ¡ j¡ ˆ j  d¡ t ¡  }ˆ_|S )r{   rQ   r|   N)r?   Úconnectr'   r(   ri   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   rU   r   rV   ri   rI   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}.
        rg   rR   rŽ   )rQ   r�   Nr‘   rI   r   r   r   Ú1test_writingToIPv6OnIPv4RaisesInvalidAddressError`  s   zCUDPPortTestsMixin.test_writingToIPv6OnIPv4RaisesInvalidAddressErrorc                 C   r“   )r”   rQ   rR   rŽ   )rg   r�   Nr‘   rI   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   rU   r   rV   r‹   rI   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   ZsetBroadcastAllowedrE   ZgetBroadcastAllowedrI   r   r   r   Útest_allowBroadcast  s   
z%UDPPortTestsMixin.test_allowBroadcastN)r   r    r!   rG   rJ   rP   r   rT   rX   rl   rs   rv   ry   rŠ   rŒ   r’   r•   r–   r—   r˜   r   r   r   r   rH   r   s,    
$
,
0
	
	rH   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.
        )rL   ÚmaxPacketSizeN)rW   )r   r+   r,   r(   rL   r�   r   r   r   r&   ”  s   ÿz&UDPServerTestsBuilder.getListeningPortN©r   r›   rœ   )r   r    r!   rG   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 IReactorSocketN)r   Z
providedByrM   ÚAF_INET6ÚgetaddrinfoÚAF_INETrN   ÚbindÚsetblockingZadoptDatagramPortÚfilenoÚfamilyÚcloser   )	r   r+   r,   r(   rL   r�   Údomainrk   ZportSockr   r   r   r&   ¯  s$   


ÿ
ý
z(UDPFDServerTestsBuilder.getListeningPortNrž   )r   r    r!   rG   r   rŸ   r&   r   r   r   r   r    ¦  s
    ÿr    ).rG   rM   Zzope.interfacer   Zzope.interface.verifyr   Ztwisted.internetr   r   Ztwisted.internet.addressr   r   Ztwisted.internet.deferr   r	   Ztwisted.internet.interfacesr
   r   r   r   Ztwisted.internet.protocolr   Z&twisted.internet.test.connectionmixinsr   r   Z#twisted.internet.test.reactormixinsr   Ztwisted.pythonr   Ztwisted.python.logr   r   Ztwisted.test.test_udpr   r   Ztwisted.test.testutilsr   Ztwisted.trial.unittestr   r   rH   rš   r    ÚglobalsÚupdateZmakeTestCaseClassesr   r   r   r   Ú<module>   s8   P  
ÿ
ÿ9