o
    bA                     @   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