U
    
W[B                     @   sb  d Z ddlmZmZ eZddl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 dd	lmZmZ dd
lmZmZmZmZ ddlmZ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) dd Z*e* Z+dd Z,G dd de Z-G dd de.Z/G dd dee/e-Z0G dd dee/e-Z1e2 3e04  e2 3e14  dS )zU
Tests for implementations of L{IReactorUDP} and the UDP parts of
L{IReactorSocket}.
    )divisionabsolute_importN)implementer)verifyObject)context)ILogContexterr)ReactorBuilder)DeferredmaybeDeferred)ILoggingContextIListeningPortIReactorUDPIReactorSocket)IPv4AddressIPv6Address)DatagramProtocol)LogObserverMixinfindFreePort)defererror)Server
GoodClient)SkipTestc                  C   sN   d} d}zt  t j} | d d}W n t jk
r<   Y nX | rJ|   |S )z5 Returns True if the system can bind an IPv6 address.NF)::1r   T)socketAF_INET6bindr   close)ZsockZ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    skipWithoutIPv66   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    	logPrefixK   s    z[DatagramTransportTestsMixin.test_startedListeningLogMessage.<locals>.SomeProtocol.logPrefixN)__name__
__module____qualname__r)   r   r   r   r    SomeProtocolI   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_startedListeningLogMessageA   s    z;DatagramTransportTestsMixin.test_startedListeningLogMessagec                    sx   |    |  | t d jf }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 %s Closed)c                    s       d S Nstopignoredr6   r   r    stopReactor^   s    zNDatagramTransportTestsMixin.test_connectionLostLogMessage.<locals>.stopReactorc                      s    d d = t j d S r;   )r   stopListeningaddCallbackr   )r5   r8   rA   r   r    doStopListeninga   s    
zRDatagramTransportTestsMixin.test_connectionLostLogMessage.<locals>.doStopListeningr   r.   N)	r/   r0   r1   r   r2   r3   ZcallWhenRunning
runReactorr4   )r(   r9   rD   r   )r5   r8   r6   rA   r    test_connectionLostLogMessageT   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	transportrB   r'   r   r   r    startProtocolx   s    
zdDatagramTransportTestsMixin.test_stopProtocolScheduling.<locals>.DisconnectingProtocol.startProtocolc                    s   d| _ | j| _   d S )NT)stoppedrH   stoppedInStartr=   r'   r@   r   r    stopProtocol~   s    zcDatagramTransportTestsMixin.test_stopProtocolScheduling.<locals>.DisconnectingProtocol.stopProtocolN)	r*   r+   r,   rG   rK   rH   rL   rJ   rM   r   r@   r   r    DisconnectingProtocolq   s   rN   N)	r   r0   r1   rE   
assertTruerG   rK   ZassertFalserL   )r(   rN   r7   r   r@   r    test_stopProtocolSchedulingk   s    
z7DatagramTransportTestsMixin.test_stopProtocolSchedulingN)r*   r+   r,   __doc__r:   rF   rP   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)r0   r1   r   rO   r   r   r(   r6   r3   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)r3   	interfaceZUDPN)	r   r   
SOCK_DGRAMr0   r1   r   r4   r2   r   )r(   hostZ
portNumberr6   r3   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   rV   N)r0   r1   r   r2   r4   rX   ZassertIsInstancer   )r(   r6   r3   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.comrZ   N)r0   assertRaisesr   InvalidAddressError	listenUDPr   )r(   r6   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;   )ra   r'   r   r   r    r)      s    zSUDPPortTestsMixin.test_logPrefix.<locals>.CustomLogPrefixDatagramProtocol.logPrefixc                 S   s.   | j d k	r*| j }d | _ |ttd  d S )Nrb   )rb   callbackr   getr   )r(   bytesr[   rb   r   r   r    datagramReceived   s    
zZUDPPortTestsMixin.test_logPrefix.<locals>.CustomLogPrefixDatagramProtocol.datagramReceivedN)r*   r+   r,   rd   r)   rh   r   r   r   r    CustomLogPrefixDatagramProtocol   s   ri   zCustom Datagramsc                    s     d|  d S )NzCustom Datagrams (UDP))r4   )rb   r'   r   r    	gotSystem   s    z3UDPPortTestsMixin.test_logPrefix.<locals>.gotSystemc                    s      S r;   r<   r>   r@   r   r    <lambda>       z2UDPPortTestsMixin.test_logPrefix.<locals>.<lambda>s
   some bytes	127.0.0.1N)r   r0   rb   r1   r2   rC   
addErrbackr   writer3   rE   )r(   ri   r7   dr3   addressrj   r   )r6   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    rd      s    zMUDPPortTestsMixin.test_writeSequence.<locals>.SimpleDatagramProtocol.__init__c                 S   s   | j | d S r;   )r   re   )r(   datar[   r   r   r    rh      s    zUUDPPortTestsMixin.test_writeSequence.<locals>.SimpleDatagramProtocol.datagramReceivedN)r*   r+   r,   rd   rh   r   r   r   r    SimpleDatagramProtocol   s   rt   )s   somes   bytess   tos   writec                    s    d |  d S )Nrl   )r4   join)rs   )dataToWriter(   r   r    gotData   s    z5UDPPortTestsMixin.test_writeSequence.<locals>.gotDatac                    s      S r;   r<   r>   r@   r   r    rk      rl   z6UDPPortTestsMixin.test_writeSequence.<locals>.<lambda>rm   N)r   r0   r   r1   r2   rC   rn   r   ZwriteSequencer3   rE   )r(   rt   r7   r   r3   rq   rw   r   )rv   r6   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)r0   r1   r   assertInstrr2   r3   rS   r   r   r    test_str  s    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)r0   r1   r   ry   reprr2   r3   rz   rS   r   r   r    	test_repr  s    zUDPPortTestsMixin.test_reprc                    s   |   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   rZ   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   )rI   ro   r2   r3   r   r
   packetReceivedr?   ZserverReceivedclientserverr   r    cbClientStarted#  s     zDUDPPortTestsMixin.test_writeToIPv6Interface.<locals>.cbClientStartedc                    s       dS z
            Stop the reactor after a datagram is received.

            @param ignored: L{None}, which is ignored
            @returns: L{None}
            Nr<   r>   r@   r   r    cbServerReceived0  s    zEUDPPortTestsMixin.test_writeToIPv6Interface.<locals>.cbServerReceivedr   r   Nr0   r   r   r
   ZstartedDeferredr1   r   rI   r2   ZgatherResultsrC   rn   r   rE   Zpacketsr4   rX   r3   r(   ZserverStartedZclientStartedZcAddrr   r   rp   Zpacketr   r   r6   r   r    test_writeToIPv6Interface  s"    
	




z+UDPPortTestsMixin.test_writeToIPv6Interfacec                    s   |   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 )
z
        An IPv6 address can be passed as the C{interface} argument to
        L{listenUDP}. The resulting Port accepts IPv6 datagrams.
        r   rZ   c                    s4    j dj  j  j d t  }_|S )r~   r   r   )rI   connectr2   r3   ro   r   r
   r   r   r   r   r    r   S  s    	zMUDPPortTestsMixin.test_connectedWriteToIPv6Interface.<locals>.cbClientStartedc                    s       dS r   r<   r>   r@   r   r    r   a  s    zNUDPPortTestsMixin.test_connectedWriteToIPv6Interface.<locals>.cbServerReceivedr   r   Nr   r   r   r   r    "test_connectedWriteToIPv6InterfaceC  s"    






z4UDPPortTestsMixin.test_connectedWriteToIPv6Interfacec                 C   s.   |   }| |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r0   r1   r   r]   r   r^   ro   rS   r   r   r    /test_writingToHostnameRaisesInvalidAddressErroru  s      zAUDPPortTestsMixin.test_writingToHostnameRaisesInvalidAddressErrorc                 C   s2   |   }| 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}.
        rm   rZ   r   )r   r   Nr   rS   r   r   r    1test_writingToIPv6OnIPv4RaisesInvalidAddressError  s         zCUDPPortTestsMixin.test_writingToIPv6OnIPv4RaisesInvalidAddressErrorc                 C   s2   |   }| j|t dd}| tj|jdd dS )r   r   rZ   r   )rm   r   Nr   rS   r   r   r    1test_writingToIPv4OnIPv6RaisesInvalidAddressError  s         zCUDPPortTestsMixin.test_writingToIPv4OnIPv6RaisesInvalidAddressErrorc                 C   s.   |   }| |t }| tj|jdd dS )zq
        Connecting to a hostname instead of an IP address will raise an
        L{InvalidAddressError}.
        r   r   N)r0   r1   r   r]   r   r^   r   rS   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)r0   r1   r   ZsetBroadcastAllowedrO   ZgetBroadcastAllowedrS   r   r   r    test_allowBroadcast  s    
z%UDPPortTestsMixin.test_allowBroadcastN)r*   r+   r,   rQ   rT   rY   r%   r\   r`   rr   rx   r{   r}   r   r   r   r   r   r   r   r   r   r   r    rR      s*   	
#		
.
1

rR   c                   @   s    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.
        )rV   maxPacketSize)r_   )r(   r6   r7   r3   rV   r   r   r   r    r1     s    
z&UDPServerTestsBuilder.getListeningPortN)r   r   r   )r*   r+   r,   rQ   r   requiredInterfacesr1   r   r   r   r    r     s
     r   c                   @   s    e Zd ZdZefZdddZdS )	UDPFDServerTestsBuilderzC
    Run L{UDPPortTestsMixin} tests using adopted UDP sockets.
    r   r   r   c           	      C   s   t |rd|kr.tj}t||d d }ntj}||f}t|tj}|| |d z||	 |j||W S |	  |
  X ntddS )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
providedByr   r   ZgetaddrinfoZAF_INETrW   r   Zsetblockingfilenor   ZadoptDatagramPortZfamilyr   )	r(   r6   r7   r3   rV   r   Zdomainrq   ZportSockr   r   r    r1     s&    


  z(UDPFDServerTestsBuilder.getListeningPortN)r   r   r   )r*   r+   r,   rQ   r   r   r1   r   r   r   r    r     s
     r   )5rQ   Z
__future__r   r   rU   Z__metaclass__r   Zzope.interfacer   Zzope.interface.verifyr   Ztwisted.pythonr   Ztwisted.python.logr   r   Z#twisted.internet.test.reactormixinsr	   Ztwisted.internet.deferr
   r   Ztwisted.internet.interfacesr   r   r   r   Ztwisted.internet.addressr   r   Ztwisted.internet.protocolr   Z&twisted.internet.test.connectionmixinsr   r   Ztwisted.internetr   r   Ztwisted.test.test_udpr   r   Ztwisted.trial.unittestr   r!   r"   r%   r&   objectrR   r   r   globalsupdateZmakeTestCaseClassesr   r   r   r    <module>   sB   Q  &  7