a QIhG@sUddlmZddlZddlZddlZddlZddlZddlZddlm Z m Z m Z m Z m Z ddlZddlZddlmZmZe rddlZddlmZddlmZmZddlmZddlmZmZmZdd lmZe ej ej!fZ"d e#d <d d d ddZ$d d dddZ%d d dddZ&d ddddZ'ddd dddZ(ej)GdddZ*ej)Gd d!d!Z+ej)Gd"d#d#Z,ej)Gd$d%d%ej-j.Z/ej)Gd&d'd'ej-j0Z1eGd(d)d)Z2eGd*d+d+ejj3ed,Z4dS)-) annotationsN) TYPE_CHECKINGAnyNoReturnUnionoverload)NoPublicConstructorfinal)Iterable) AddressFamily SocketKind) TracebackType)BufferSelf TypeAlias) AddressFormatr IPAddressint)ipreturncCs4t|tjrtjjSt|tjr(tjjStddS)Nz!Unhandled IPAddress instance type) isinstance ipaddress IPv4AddresstriosocketAF_INET IPv6AddressAF_INET6NotImplementedError)rrP/var/www/html/venv_bot_3.9/lib/python3.9/site-packages/trio/testing/_fake_net.py _family_for+s   r!)familyrcCs8|tjjkrtdS|tjjkr,tdStddS)N0.0.0.0::Unhandled ip address familyrrrr ip_addressrrr"rrr _wildcard_ip_for3s     r)cCs8|tjjkrtdS|tjjkr,tdStddS)N 127.0.0.1::1r%r&r(rrr _localhost_ip_for<s     r,r)codercCst|t|dSN)OSErrorosstrerror)r-rrr _fake_errDsr2bytesIterable[Buffer])databuffersrc Cs|d}|D]n}|||t|j}t| }||dt|<Wdn1sP0Y|t|7}|t|krqxq|S)Nr) memoryviewnbyteslen)r5r6writtenbufZ next_pieceZmbufrrr _scatterHs .  r<c@s@eZdZUded<ded<ddddZeddd d d Zd S) UDPEndpointrrrport+tuple[str, int] | tuple[str, int, int, int]rcCs(|jj|jf}t|jtjr$|d7}|S)N)rr)r compressedr>rrr)selfsockaddrrrr as_python_sockaddrYs zUDPEndpoint.as_python_sockaddr)rCrcCs"|dd\}}|t||dS)N)rr>)rr')clsrCrr>rrr from_python_sockaddrbsz UDPEndpoint.from_python_sockaddrN)__name__ __module__ __qualname____annotations__rD classmethodrGrrrr r=Ts  r=c@seZdZUded<dS) UDPBindingr=localN)rHrIrJrKrrrr rMks rMc@sFeZdZUded<ded<ejdddZded<ddd d d Zd S) UDPPacketr=source destinationcCs|Sr.)hex)prrr uzUDPPacket.)reprr3payload)rWrcCst|j|j|dS)NrPrQrW)rOrQrP)rBrWrrr replyxs zUDPPacket.replyN)rHrIrJrKattrsfieldrWrYrrrr rOqs rOc@s*eZdZUded<dddddddZdS) FakeSocketFactoryFakeNetfake_netr FakeSocket)r"type_protorcCst|j|||Sr.)r__creater^)rBr"r`rarrr rszFakeSocketFactory.socketN)rHrIrJrKrrrrr r\s r\c @sDeZdZUded<dddddddddd d Zd dd d ddZdS)FakeHostnameResolverr]r^rz bytes | Nonezbytes | str | int | Nonerzqlist[tuple[AddressFamily, SocketKind, int, str, tuple[str, int] | tuple[str, int, int, int] | tuple[int, bytes]]])hostr>r"typeraflagsrcs tddSNzFakeNet doesn't do fake DNS yetr)rBrdr>r"rerarfrrr getaddrinfosz FakeHostnameResolver.getaddrinfor?ztuple[str, str])rCrfrcs tddSrgrh)rBrCrfrrr getnameinfosz FakeHostnameResolver.getnameinfoN)rrrr)rHrIrJrKrirjrrrr rcs rcc@sZeZdZddddZdddddd Zddd d Zd dd ddZd dd ddZdS)r]Noner@cCs@td|_td|_ttdd|_i|_ d|_ dS)Nz 1.0.0.0/8z1::/16iPi) r IPv4NetworkhostsZ_auto_ipv4_iter IPv6NetworkZ_auto_ipv6_iteriterrange_auto_port_iter_bound route_packetrBrrr __init__s zFakeNet.__init__rMr_)bindingrrcCs"||jvrttj||j|<dSr.)rrr2errno EADDRINUSE)rBrvrrrr _binds  z FakeNet._bindcCs$tjt|tjt|dSr.)rrZset_custom_socket_factoryr\Zset_custom_hostname_resolverrcrtrrr enableszFakeNet.enablerOpacketrcCs$|jdur||n ||dSr.)rsdeliver_packetrBr|rrr send_packets  zFakeNet.send_packetcCs,t|jd}||jvr(|j||ndS)NrN)rMrQrr_deliver_packet)rBr|rvrrr r}s  zFakeNet.deliver_packetN)rHrIrJruryrzrr}rrrr r]s  r]c@s<eZdZddddddddZedd d d Zedd d d Zedd ddZdd ddZdd ddZ ddddddZ dddddZ ddddd Z dd!d"d#d$Z d~d(d)dd*dd+d,d-Zejd.kseseejd/reZdd(ddd0d1d2d3Zejd.kses"eejd/r"eZd4d d5d6Zd4d d7d8Zedddd9d:d;Zeddddd;Zdddd?d@d=dAd;ZedddBddCdDdEZeddddddFdGdEZddddHd?ddFdIdEZdJd dKdLZdMdNdOddPdQdRZddSdddTdUdVZedSdWddXdYdZZedSdd[dd\d]dZZd^dd_d`dZZdddd_rrvrrr binds"      zFakeSocket.bindr)peerrcs tddS)Nz0FakeNet does not (yet) support connected socketsrh)rBrrrr connect2szFakeSocket.connectrrNr4z!Iterable[tuple[int, int, Buffer]]zAddressFormat | None)r6ancdatarfrrcs|tjIdH|dur4|j|ddIdH}|r@td|rRtd||durdttjt |}|j dur| t |jjdfIdHd|}|j dusJt|j j||d}|j|t|S)NFr&FakeNet doesn't support ancillary dataz"FakeNet send flags must be 0, not rrUrX)rrrrrrr2rwENOTCONNr=rGrrr)r"rAjoinrOrNrrr9)rBr6rrfrrQrWr|rrr _sendmsg5s,     zFakeSocket._sendmsgwin32sendmsgzZtuple[int, list[tuple[int, int, bytes]], int, tuple[str, int] | tuple[str, int, int, int]])r6 ancbufsizerfrc s|dkrtd|dkr td|jdur2td|g}d}|jIdH}|j}t|j|}|t |jkr|t j j O}||||fS)Nrrz&FakeNet doesn't support any recv flagszThe code will most likely hang if you try to receive on a fakesocket without a binding. If that is not the case, or you explicitly want to test that, remove this warning.) rrrrZreceiverPrDr<rWr9rr MSG_TRUNC) rBr6rrfr msg_flagsr|rr:rrr _recvmsg_intoas"     zFakeSocket._recvmsg_intor?cCsJ||jdur|jjS|jtjjkr0dS|jtjjksBJdSdS)N)r#r)r$r) rrrNrDr"rrrrrtrrr getsocknames  zFakeSocket.getsocknamecCsb||jdurTt|jds&Jd|jjdurTt|jjtsHJd|jjSttj dS)NremotezGThis method seems to assume that self._binding has a remote UDPEndpointz,Self._binding.remote should be a UDPEndpoint) rrhasattrrrr=rDr2rwrrtrrr getpeernames"   zFakeSocket.getpeername)leveloptnamercCsdSr.r)rBrrrrr getsockoptszFakeSocket.getsockoptr3)rrbuflenrcCsdSr.rrBrrrrrr rsz int | Nonez int | bytescCs"|td|d|ddS)Nz%FakeNet doesn't implement getsockopt(, ))rr/rrrr rsz int | Buffer)rrvaluercCsdSr.r)rBrrrrrr setsockoptszFakeSocket.setsockopt)rrroptlenrcCsdSr.rrBrrrrrrr rszint | Buffer | NonecCsF|||ftjjtjjfkr,|s,tdtd|d|ddS)Nz#FakeNet always has IPV6_V6ONLY=Truez%FakeNet doesn't implement setsockopt(rz, ...))rrr IPPROTO_IPV6 IPV6_V6ONLYrr/rrrr rsrcCs|Sr.rrtrrr __enter__szFakeSocket.__enter__z#builtins.type[BaseException] | NonezBaseException | NonezTracebackType | None)exc_type exc_value tracebackrcCs |dSr.)r)rBrrrrrr __exit__szFakeSocket.__exit__r)r5rfrcs|||dIdHSr.)sendto)rBr5rfrrr sendszFakeSocket.sendz!tuple[object, ...] | str | Buffer)_FakeSocket__data_FakeSocket__addressrcsdSr.r)rBrrrrr rszFakeSocket.sendtoz(tuple[object, ...] | str | Buffer | None)r_FakeSocket__flagsrrcsdSr.r)rBrrrrrr rsr)argsrcsRt|dkr|\}}d}n t|dkr2|\}}}ntd||gg||IdHS)NrErzwrong number of arguments)r9 TypeErrorr)rBrr5rrfrrr rs   )bufsizerfrcs|||IdH\}}|Sr.)recvfrom)rBrrfr5_addressrrr recv szFakeSocket.recv)r;r8rfrcs||||IdH\}}|Sr.) recvfrom_into)rBr;r8rfZ got_bytesrrrr recv_intoszFakeSocket.recv_intoztuple[bytes, AddressFormat]cs"|||IdH\}}}}||fSr.)_recvmsg)rBrrfr5_ancdata _msg_flagsrrrr rszFakeSocket.recvfromztuple[int, AddressFormat]csD|dkr|t|jkrtd||gd|IdH\}}}}||fS)Nrzpartial recvfrom_into)r7r8rr)rBr;r8rf got_nbytesrrrrrr rszFakeSocket.recvfrom_intoz>tuple[bytes, list[tuple[int, int, bytes]], int, AddressFormat])rrrfrc s>t|}||g||IdH\}}}}t|d||||fSr.) bytearrayrr3) rBrrrfr;rrrrrrr r)szFakeSocket._recvmsgcCs tddS)Nz&can't get fileno() for FakeNet socketsrhrtrrr fileno<szFakeSocket.filenocCs tddS)Nzcan't detach() a FakeNet socketrhrtrrr detach?szFakeSocket.detachcCsdS)NFrrtrrr get_inheritableBszFakeSocket.get_inheritable) inheritablercCs|r tddS)Nz&FakeNet can't make inheritable socketsrh)rBrrrr set_inheritableEszFakeSocket.set_inheritableshare) process_idrcCs tddS)NzFakeNet can't share socketsrh)rBrrrr rMszFakeSocket.share)rrN)rr)N)N)r)r)rr)r)rr)rr)+rHrIrJrupropertyrer"rarrrrrrrsysplatformrrrrr recvmsg_intorrrrrrrrrrrrrrrecvmsgrrrrrrrrr r_s  '  $         r_) metaclass)5 __future__rrrwrr0rrtypingrrrrrrZrZ trio._utilrr builtinscollections.abcr r r typesr Ztyping_extensionsrrrZ trio._socketrrrrrKr!r)r,r2r<frozenr=rMrOabcZ SocketFactoryr\ZHostnameResolverrcr] SocketTyper_rrrr  sJ     #