o i@sdZddlmZddlZddlmZddlmZddlm Z ddl m Z e dkr/dd l m Zndd l mZGd d d ZGd ddeZGddde jZdS)zK Tests for the internal implementation details of L{twisted.internet.udp}. ) annotationsN)udp)DatagramProtocol) platformType)unittestwin32)WSAEWOULDBLOCK) EWOULDBLOCKc@s.eZdZdZdddZdd d ZdddZdS)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. retvalslist[bytes | socket.error]returnNonecCs||_d|_dSN)r connectedAddr)selfr rnC:\Users\psant\Downloads\PAF backend\school\venv\Lib\site-packages\twisted/internet/test/test_udp_internals.py__init__ s zStringUDPSocket.__init__addrobjectcCs ||_dSr)r)rrrrrconnect$ zStringUDPSocket.connectsizeinttuple[bytes, None]cCs$|jd}t|tjr||dfS)zH Return (or raise) the next value from C{self.retvals}. rN)r pop isinstancesocketerror)rrretrrrrecvfrom's  zStringUDPSocket.recvfromN)r r r r)rrr r)rrr r)__name__ __module__ __qualname____doc__rrr!rrrrr s  r c@s$eZdZdZd ddZdd d Zd S) KeepReadsz% Accumulate reads in a list. r rcCs g|_dSr)reads)rrrrr6rzKeepReads.__init__databytesrrcCs|j|dSr)r'append)rr(rrrrdatagramReceived9szKeepReads.datagramReceivedNr r)r(r)rrr r)r"r#r$r%rr+rrrrr&1s r&c@s8eZdZdZd ddZd ddZd dd Zd d d Zd S) ErrorsTestsz/ Error handling tests for C{udp.Port}. r rcCstjd|tjjdt}td|}tddt ddt dg|_| | |j ddg| | |j gddS)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. iNresult123456)r.r/r0) rZ_sockErrReadIgnorer* addCleanupremover&Portr rrdoRead assertEqualr'rprotocolportrrrtest_socketReadNormalBs  z!ErrorsTests.test_socketReadNormalcCstjd|tjjdt}dd|_td|}tdt ddt t g|_ | | |jdg| | |jddgdS)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. cSsddS)Nrrrrrresz5ErrorsTests.test_readImmediateError..Nab)r_sockErrReadRefuser*r1r2r&connectionRefusedr3r rrr r4r5r'r6rrrtest_readImmediateErrorXs   z#ErrorsTests.test_readImmediateErrorcstjd|tjjdt}gfdd|_td|}tdt ddt t g|_ | dd| ||jdg|d g| ||jddg|d gdS) 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. r:cs dS)NT)r*rZrefusedrrr<s z>ErrorsTests.test_connectedReadImmediateError..Nr=r>z 127.0.0.1i'T)rr?r*r1r2r&r@r3r rrr rr4r5r'r6rrBr test_connectedReadImmediateErrorts   z,ErrorsTests.test_connectedReadImmediateErrorcCsJt}td|}tdtdg|_|tj|j||j dgdS)zG Socket reads with an unknown socket error are raised. Nsgoodi) r&rr3r rr assertRaisesr4r5r'r6rrrtest_readUnknownErrors  z!ErrorsTests.test_readUnknownErrorNr,)r"r#r$r%r9rArCrErrrrr-=s    r-)r% __future__rrZtwisted.internetrZtwisted.internet.protocolrZtwisted.python.runtimerZ twisted.trialrerrnorr r r&ZSynchronousTestCaser-rrrrs