
    Vh                        d Z ddlm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 e	dk(  rdd	lmZ ndd
lmZ  G d d      Z G d de      Z G d dej"                        Zy)zK
Tests for the internal implementation details of L{twisted.internet.udp}.
    )annotationsN)udp)DatagramProtocol)platformType)unittestwin32)WSAEWOULDBLOCK)EWOULDBLOCKc                  (    e Zd ZdZddZddZddZy)	StringUDPSocketa  
    A fake UDP socket object, which returns a fixed sequence of strings and/or
    socket errors.  Useful for testing.

    @ivar retvals: A C{list} containing either strings or C{socket.error}s.

    @ivar connectedAddr: The address the socket is connected to.
    c                     || _         d | _        y N)retvalsconnectedAddr)selfr   s     X/home/dcms/DCMS/lib/python3.12/site-packages/twisted/internet/test/test_udp_internals.py__init__zStringUDPSocket.__init__    s    ,0    c                    || _         y r   )r   )r   addrs     r   connectzStringUDPSocket.connect$   s
    !r   c                x    | j                   j                  d      }t        |t        j                        r||dfS )zH
        Return (or raise) the next value from C{self.retvals}.
        r   N)r   pop
isinstancesocketerror)r   sizerets      r   recvfromzStringUDPSocket.recvfrom'   s5     llq!c6<<(IDyr   N)r   zlist[bytes | socket.error]returnNone)r   objectr    r!   )r   intr    ztuple[bytes, None])__name__
__module____qualname____doc__r   r   r    r   r   r   r      s    1"r   r   c                       e Zd ZdZddZddZy)	KeepReadsz%
    Accumulate reads in a list.
    c                    g | _         y r   )reads)r   s    r   r   zKeepReads.__init__6   s	    "$
r   c                :    | j                   j                  |       y r   )r,   append)r   datar   s      r   datagramReceivedzKeepReads.datagramReceived9   s    

$r   Nr    r!   )r/   bytesr   r"   r    r!   )r$   r%   r&   r'   r   r0   r(   r   r   r*   r*   1   s    % r   r*   c                  0    e Zd ZdZddZddZddZddZy)ErrorsTestsz/
    Error handling tests for C{udp.Port}.
    c           	        t         j                  j                  d       | j                  t         j                  j                  d       t               }t        j                  d|      }t        ddt        j                  d      dt        j                  d      g      |_        |j                          | j                  |j                  ddg       |j                          | j                  |j                  g d       y)z
        Socket reads with some good data followed by a socket error which can
        be ignored causes reading to stop, and no log messages to be logged.
        iN   result   123   456)r6   r7   r8   )r   _sockErrReadIgnorer.   
addCleanupremover*   Portr   r   r   doReadassertEqualr,   r   protocolports      r   test_socketReadNormalz!ErrorsTests.test_socketReadNormalB   s     	%%e,..55u=;xxh' &U 3VV\\%=PQ
 	)V)<=)DEr   c                   t         j                  j                  d       | j                  t         j                  j                  d       t               }d |_        t        j                  d|      }t        dt        j                  d      dt        j                  t              g      |_	        |j                          | j                  |j                  dg       |j                          | j                  |j                  ddg       y)z
        If the socket is unconnected, socket reads with an immediate
        connection refusal are ignored, and reading stops. The protocol's
        C{connectionRefused} method is not called.
        c                     ddz  S )N   r   r(   r(   r   r   <lambda>z5ErrorsTests.test_readImmediateError.<locals>.<lambda>e   s
    QU r   N   a   b)r   _sockErrReadRefuser.   r:   r;   r*   connectionRefusedr<   r   r   r   r
   r=   r>   r,   r?   s      r   test_readImmediateErrorz#ErrorsTests.test_readImmediateErrorX   s     	%%e,..55u=;%2"xxh' &6<<&fll;.GH
 	$0$6r   c                   t         j                  j                  d       | j                  t         j                  j                  d       t               }g fd|_        t        j                  d|      }t        dt        j                  d      dt        j                  t              g      |_	        |j                  dd       |j                          | j                  |j                  dg       | j                  dg       |j                          | j                  |j                  ddg       | j                  dg       y)	z
        If the socket connected, socket reads with an immediate
        connection refusal are ignored, and reading stops. The protocol's
        C{connectionRefused} method is called.
        rD   c                 &     j                  d      S )NT)r.   )refuseds   r   rG   z>ErrorsTests.test_connectedReadImmediateError.<locals>.<lambda>   s    W^^D-A r   NrH   rI   z	127.0.0.1i'  T)r   rJ   r.   r:   r;   r*   rK   r<   r   r   r   r
   r   r=   r>   r,   )r   r@   rA   rO   s      @r    test_connectedReadImmediateErrorz,ErrorsTests.test_connectedReadImmediateErrort   s     	%%e,..55u=;%A"xxh'%6<<&fll;.GH
 	[$' 	$04&) 	$64&)r   c                   t               }t        j                  d|      }t        dt	        j
                  d      g      |_        | j                  t        j
                  |j                         | j                  |j                  dg       y)zG
        Socket reads with an unknown socket error are raised.
        Ns   goodi)
r*   r   r<   r   r   r   assertRaisesr=   r>   r,   r?   s      r   test_readUnknownErrorz!ErrorsTests.test_readUnknownError   se     ;xxh' &wU0C&DE&,,4'3r   Nr1   )r$   r%   r&   r'   rB   rL   rP   rS   r(   r   r   r4   r4   =   s    F,78*>
4r   r4   )r'   
__future__r   r   twisted.internetr   twisted.internet.protocolr   twisted.python.runtimer   twisted.trialr   errnor	   r
   r   r*   SynchronousTestCaser4   r(   r   r   <module>r[      sW    #    6 / "73! 6	   	 `4(.. `4r   