/JfA$dZddlZddlmZddlmZddlmZmZddl m Z m Z ddl m Z mZddlmZmZmZmZdd lmZdd lmZmZdd lmZdd lmZdd lmZmZddl m!Z!m"Z"ddl#m$Z$dZ%e%Z&dZ'GddeZ(GddZ)Gddee)e(Z*Gddee)e(Z+e,-e*.e,-e+.dS)zU Tests for implementations of L{IReactorUDP} and the UDP parts of L{IReactorSocket}. N) implementer) verifyObject)defererror) IPv4Address IPv6Address)Deferred maybeDeferred)IListeningPortILoggingContextIReactorSocket IReactorUDP)DatagramProtocol)LogObserverMixin findFreePort)ReactorBuilder)context) ILogContexterr) GoodClientServer)SkipTestcd}d} tjtj}|dd}n#t$rYnwxYw|r||S)z4Returns True if the system can bind an IPv6 address.NF)::1rT)socketAF_INET6bindOSErrorclose)sockhas_ipv6s a/home/alex/cs2snipeproduction/venv/lib/python3.11/site-packages/twisted/internet/test/test_udp.py _has_ipv6r#!sw DH }V_-- *        Os5< A A c"tsd|_|S)Nz.Does not work on systems without IPv6 support.)HAS_IPV6skip)fs r"skipWithoutIPv6r(5s BA Hc$eZdZdZdZdZdZdS)DatagramTransportTestsMixinzP Mixin defining tests which apply to any port/datagram based transport. cp|}|}ttGddt}|}|||}d|jfz}||f|dddS)zu When a port starts, a message including a description of the associated protocol is logged. ceZdZdZdS)QDatagramTransportTestsMixin.test_startedListeningLogMessage..SomeProtocolcdS)NzCrazy Protocolselfs r" logPrefixz[DatagramTransportTestsMixin.test_startedListeningLogMessage..SomeProtocol.logPrefixJs''r)N)__name__ __module__ __qualname__r3r0r)r" SomeProtocolr.Hs# ( ( ( ( (r)r7zCrazy Protocol starting on %drmessageN) observe buildReactorrr rgetListeningPortgetHostport assertEqual)r2loggedMessagesreactorr7protocolpexpectedMessages r"test_startedListeningLogMessagez;DatagramTransportTestsMixin.test_startedListeningLogMessage@s ##%% _ % % ( ( ( ( (+ ( ( & % ( <>>  ! !'8 4 49QYY[[=My-IJJJJJr)c|||tdjd}fdfd}||||fdddS)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)c0dSNstopignoredr@s r" stopReactorzNDatagramTransportTestsMixin.test_connectionLostLogMessage..stopReactor]s LLNNNNNr)c`dd=tjdSrG)r stopListening addCallback)r?rBrLsr"doStopListeningzRDatagramTransportTestsMixin.test_connectionLostLogMessage..doStopListening`s3qqq! !/ * * 6 6{ C C C C Cr)rr8N) r9r:r;rr<r=callWhenRunning runReactorr>)r2rCrPr?rBr@rLs @@@@r"test_connectionLostLogMessagez9DatagramTransportTestsMixin.test_connectionLostLogMessageSs ##%%  ! !'+;+=+= > >Aqyy{{'7AAA      D D D D D D D 000     /+^A->y-IJJJJJr)c`Gfddt}||}|||||j||j||jdS)z L{DatagramProtocol.stopProtocol} is called asynchronously (ie, not re-entrantly) when C{stopListening} is used to stop the datagram transport. c0eZdZdZdZdZdZdZfdZdS)VDatagramTransportTestsMixin.test_stopProtocolScheduling..DisconnectingProtocolFcbd|_d|_|jd|_dS)NTF)startedinStartProtocol transportrNr1s r" startProtocolzdDatagramTransportTestsMixin.test_stopProtocolScheduling..DisconnectingProtocol.startProtocolvs3# '+$,,...',$$$r)cVd|_|j|_dS)NT)stoppedrYstoppedInStartrIr2r@s r" stopProtocolzcDatagramTransportTestsMixin.test_stopProtocolScheduling..DisconnectingProtocol.stopProtocol|s&# &*&:# r)N) r4r5r6rXr]rYr^r[r`)r@sr"DisconnectingProtocolrVpsQGG#O"N - - -        r)raN) rr:r;rR assertTruerXr] assertFalser^)r2rarAr@s @r"test_stopProtocolSchedulingz7DatagramTransportTestsMixin.test_stopProtocolSchedulingis       $4   "##%%((** gx000     ())) ())) 011111r)N)r4r5r6__doc__rDrSrdr0r)r"r+r+;sQKKK&KKK,22222r)r+ceZdZdZdZdZedZdZdZ dZ dZ d Z ed Z ed Zd Zed ZedZdZdZdS)UDPPortTestsMixinzY Tests for L{IReactorUDP.listenUDP} and L{IReactorSocket.adoptDatagramPort}. c|}||t}|t t |dS)zY L{IReactorUDP.listenUDP} returns an object providing L{IListeningPort}. N)r:r;rrbrr r2r@r=s r"test_interfacez UDPPortTestsMixin.test_interfacesS##%%$$W.>.@.@AA  ^T::;;;;;r)c ttj\}}|}||t ||}||td||dS)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= interfaceUDPN) rr SOCK_DGRAMr:r;rr>r<r)r2host portNumberr@r=s r" test_getHostzUDPPortTestsMixin.test_getHosts (V->???j##%%$$ %''jD%   UD*)M)MNNNNNr)c |}||td}|}||jd||tdS)zr L{IListeningPort.getHost} returns an L{IPv6Address} when listening on an IPv6 interface. rrmN)r:r;rr<r>rpassertIsInstancer)r2r@r=addrs r"test_getHostIPv6z"UDPPortTestsMixin.test_getHostIPv6sw ##%%$$W.>.@.@E$RR||~~ E*** dK00000r)c|}|tj|jt dddS)z An L{InvalidAddressError} is raised when trying to listen on an address that isn't a valid IPv4 or IPv6 address. rz example.comrtN)r: assertRaisesrInvalidAddressError listenUDPrr_s r"test_invalidInterfacez'UDPPortTestsMixin.test_invalidInterfacesW ##%%   %      #      r)cGddt}|d}|j}|}|}fd}|||t|fd|dd|j f dS) z Datagram transports implement L{ILoggingContext.logPrefix} to return a message reflecting the protocol they are running. c eZdZdZdZdZdS)IUDPPortTestsMixin.test_logPrefix..CustomLogPrefixDatagramProtocolc:||_t|_dSrG)_prefixr system)r2prefixs r"__init__zRUDPPortTestsMixin.test_logPrefix..CustomLogPrefixDatagramProtocol.__init__s% &jj r)c|jSrG)rr1s r"r3zSUDPPortTestsMixin.test_logPrefix..CustomLogPrefixDatagramProtocol.logPrefixs |#r)c|jB|j}d|_|tjtddSdS)Nr)rcallbackrgetr)r2bytesrvrs r"datagramReceivedzZUDPPortTestsMixin.test_logPrefix..CustomLogPrefixDatagramProtocol.datagramReceivedsG;*![F"&DKOOGK $<$)rr2s r" gotSystemz3UDPPortTestsMixin.test_logPrefix..gotSystems   5v > > > > >r)c,SrGrHrJs r"z2UDPPortTestsMixin.test_logPrefix..sgllnnr)s some bytes 127.0.0.1N) rr:rr;r<rO addErrbackrwriter=rR)r2rrAdr=addressrr@s` @r"test_logPrefixz UDPPortTestsMixin.test_logPrefixs H H H H H.> H H H##%%223EFF O$$Wh77,,.. ? ? ? ? ? i    S 4444555 =; "=>>>      r)cGddt}|}|j}|}|}dfd}|j||jt|jfd|d|j f dS)zH Write a sequence of L{bytes} to a L{DatagramProtocol}. ceZdZdZdZdS)DUDPPortTestsMixin.test_writeSequence..SimpleDatagramProtocolc,t|_dSrG)r rr1s r"rzMUDPPortTestsMixin.test_writeSequence..SimpleDatagramProtocol.__init__s%ZZ r)c:|j|dSrG)rr)r2datarvs r"rzUUDPPortTestsMixin.test_writeSequence..SimpleDatagramProtocol.datagramReceiveds ##D)))))r)N)r4r5r6rrr0r)r"SimpleDatagramProtocolrs2 ( ( ( * * * * *r)r)ssomesbytesstoswritecZd|dS)Nr))r>join)r dataToWriter2s r"gotDataz5UDPPortTestsMixin.test_writeSequence..gotDatas+   SXXk22D 9 9 9 9 9r)c,SrGrHrJs r"rz6UDPPortTestsMixin.test_writeSequence..s',,..r)rN) rr:rr;r<rOrr writeSequencer=rR) r2rrArr=rrrr@s ` @@r"test_writeSequencez$UDPPortTestsMixin.test_writeSequences  * * * * *%5 * * *##%%))++$$Wh77,,..:  : : : : : : '"""8888999 ;gl(CDDD      r)c|}||t}|t |jt |dS)zQ C{str()} on the listening port object includes the port number. N)r:r;rassertInstrr<r=ris r"test_strzUDPPortTestsMixin.test_strsd##%%$$W.>.@.@AA c$,,..-..D :::::r)c|}||t}|t |jt|dS)zR C{repr()} on the listening port object includes the port number. N)r:r;rrreprr<r=rris r" test_reprzUDPPortTestsMixin.test_reprsd##%%$$W.>.@.@AA d4<<>>.//T;;;;;r)c | t tjx} _| dt tjx}_| dj} fd} fd}tj ||g}| || || t|  jd}||d|j|jffdS)zS Writing to an IPv6 UDP socket on the loopback interface succeeds. rrtcjddjjft jx}_|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)rZrr<r=rr packetReceivedrKserverReceivedclientservers r"cbClientStartedzDUDPPortTestsMixin.test_writeToIPv6Interface..cbClientStartedsP   " "7UF4D4L4L4N4N4S,T U U U5:^5E5E ENV2! !r)c0dSz Stop the reactor after a datagram is received. @param ignored: L{None}, which is ignored @returns: L{None} NrHrJs r"cbServerReceivedzEUDPPortTestsMixin.test_writeToIPv6Interface..cbServerReceived+s LLNNNNNr)rrNr:rrr startedDeferredr;rrZr< gatherResultsrOrrrRpacketsr>rpr= r2 serverStarted clientStartedcAddrrrrpacketrr@rs @@@r"test_writeToIPv6Interfacez+UDPPortTestsMixin.test_writeToIPv6Interfacesk ##%%161A1AA . gv???161A1AA . gv??? ((** " " " " " "        > ? ? o&&& &''' S    " 'EJ +C!DEEEEEr)c | t tjx} _| dt tjx}_| dj} fd} fd}tj ||g}| || || t|  jd}||d|j|jffdS)z An IPv6 address can be passed as the C{interface} argument to L{listenUDP}. The resulting Port accepts IPv6 datagrams. rrtcjdjjjdt jx}_|S)rrr)rZconnectr<r=rrr rrs r"rzMUDPPortTestsMixin.test_connectedWriteToIPv6Interface..cbClientStartedMsb   $ $UF,<,D,D,F,F,K L L L   " "7 + + +5:^5E5E ENV2! !r)c0dSrrHrJs r"rzNUDPPortTestsMixin.test_connectedWriteToIPv6Interface..cbServerReceived[s LLNNNNNr)rrNrrs @@@r""test_connectedWriteToIPv6Interfacez4UDPPortTestsMixin.test_connectedWriteToIPv6Interface=sk ##%%161A1AA . gv???161A1AA . gv??? ((** " " " " " "        > ? ? o&&& &''' S    " 'EJ +C!DEEEEEr)c|}||t}|tj|jdddS)zn Writing to a hostname instead of an IP address will raise an L{InvalidAddressError}. spam)example.invalidNr:r;rryrrzrris r"/test_writingToHostnameRaisesInvalidAddressErrorzAUDPPortTestsMixin.test_writingToHostnameRaisesInvalidAddressErrornsb ##%%$$W.>.@.@AA   %tz6;Q     r)c|}||td}|tj|jdddS)l Writing to an IPv6 address on an IPv4 socket will raise an L{InvalidAddressError}. rrtr)rrNrris r"1test_writingToIPv6OnIPv4RaisesInvalidAddressErrorzCUDPPortTestsMixin.test_writingToIPv6OnIPv4RaisesInvalidAddressErrorys[ ##%%$$W.>.@.@K$XX %3TZTTTTTr)c|}||td}|tj|jdddS)rrrtr)rrNrris r"1test_writingToIPv4OnIPv6RaisesInvalidAddressErrorzCUDPPortTestsMixin.test_writingToIPv4OnIPv6RaisesInvalidAddressErrorsg ##%%$$W.>.@.@E$RR   %tz6;K     r)c|}||t}|tj|jdddS)zq Connecting to a hostname instead of an IP address will raise an L{InvalidAddressError}. rrN)r:r;rryrrzrris r"2test_connectingToHostnameRaisesInvalidAddressErrorzDUDPPortTestsMixin.test_connectingToHostnameRaisesInvalidAddressErrorsX ##%%$$W.>.@.@AA %3T\CTVWXXXXXr)c|}||t}|d||dS)zk L{IListeningPort.setBroadcastAllowed} sets broadcast to be allowed on the socket. TN)r:r;rsetBroadcastAllowedrbgetBroadcastAllowedris r"test_allowBroadcastz%UDPPortTestsMixin.test_allowBroadcastsi ##%%$$W.>.@.@AA   &&& 002233333r)N)r4r5r6rerjrrr(rwr|rrrrrrrrrrrr0r)r"rgrgsM <<< O O O 1 1_ 1    "!"!"!H!!!8;;;<<<*F*F_*FX.F.F_.F`    UU_U   _  YYY44444r)rgc"eZdZdZefZ ddZdS)UDPServerTestsBuilderzM Run L{UDPPortTestsMixin} tests using newly created UDP sockets. r c4|||||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. )rm maxPacketSize)r{)r2r@rAr=rmrs r"r;z&UDPServerTestsBuilder.getListeningPorts*  (i}!   r)Nrrr)r4r5r6rerrequiredInterfacesr;r0r)r"rrsC &FJ      r)rc"eZdZdZefZ ddZdS)UDPFDServerTestsBuilderzC Run L{UDPPortTestsMixin} tests using adopted UDP sockets. rrrcjtj|rd|vr.tj}tj||dd}ntj}||f}tj|tj}|||d | | |j ||| | S#| | wxYwtd)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. :rFz'Reactor does not provide IReactorSocket)r providedByrr getaddrinfoAF_INETror setblockingadoptDatagramPortfilenofamilyrr) r2r@rAr=rmrdomainrportSocks r"r;z(UDPFDServerTestsBuilder.getListeningPorts!.  $W - - Fi ,Y==a@C$d+}VV->??H MM' " " "   ' ' ' !00OO%%x- !!!    !!!    DEE Es ".C99*D#Nr)r4r5r6rer rr;r0r)r"rrsJ)*FJ-F-F-F-F-F-Fr)r)/rerzope.interfacerzope.interface.verifyrtwisted.internetrrtwisted.internet.addressrrtwisted.internet.deferr r twisted.internet.interfacesr r r rtwisted.internet.protocolr&twisted.internet.test.connectionmixinsrr#twisted.internet.test.reactormixinsrtwisted.pythonrtwisted.python.logrrtwisted.test.test_udprrtwisted.trial.unittestrr#r%r(r+rgrrglobalsupdatemakeTestCaseClassesr0r)r"rs  &&&&&&......))))))))========:::::::: 766666QQQQQQQQ>>>>>>""""""////////44444444++++++" 9;;   M2M2M2M2M2"2M2M2M2`U4U4U4U4U4U4U4U4p     %'B   86F6F6F6F6F%'B6F6F6Fr &::<<=== (<<>>?????r)