#)fZUddlmZddlZddlmZmZmZmZddlm Z m Z ddl m Z ddl mZmZerddlmZegeefZd ed <egefZd ed <ed e Zede ZGddZe j2GddeZe j2GddeZdd d"dZd#dZ d$dZd%dZGddZ GddeZ!GddeZ"d&d Z#d'd!Z$y)() annotationsN) TYPE_CHECKING AwaitableCallableTypeVar)_core_util StapledStream) ReceiveStream SendStream) TypeAliasr AsyncHookSyncHook SendStreamT)boundReceiveStreamTcPeZdZd dZd dZd dZd dZd dZd dZdd dZ dd d Z y)_UnboundedByteQueuect|_d|_tj|_t jd|_y)NFz%another task is already fetching data) bytearray_data_closedr ParkingLot_lotr ConflictDetector _fetch_lockselfs Y/var/www/html/flask-app/venv/lib/python3.12/site-packages/trio/testing/_memory_streams.py__init__z_UnboundedByteQueue.__init__s8[  $$&  11 3 cFd|_|jjyNT)rr unpark_allrs r!closez_UnboundedByteQueue.close's  r#cBt|_|jyN)rrr'rs r!close_and_wipez"_UnboundedByteQueue.close_and_wipe+s[  r#c|jrtjd|xj|z c_|jj y)Nzvirtual connection closed)rr ClosedResourceErrorrrr&r datas r!putz_UnboundedByteQueue.put/s9 <<++,GH H d  r#cT|ytj|}|dkr tdy)Nmax_bytes must be >= 1)operatorindex ValueErrorr max_bytess r!_check_max_bytesz$_UnboundedByteQueue._check_max_bytes5s1   NN9- q=56 6 r#c|js|jsJ|t|j}|jr$|jd|}|jd|=|sJ|StSr))rrlenr)r r7chunks r! _get_implz_UnboundedByteQueue._get_impl<s^||tzz)  DJJI ::JJz *E :I:&LL; r#Nc|j5|j||js|jstj |j |cdddS#1swYyxYwr))rr8rrr WouldBlockr<r6s r! get_nowaitz_UnboundedByteQueue.get_nowaitHsS    -  ! !) ,<< &&&>>),  - - -s A A!!A*cHK|j5|j||js/|js#|jj d{nt jd{|j|cdddS7;7 #1swYyxYwwr)) rr8rrrparkr checkpointr<r6s r!getz_UnboundedByteQueue.getOs    -  ! !) ,<< iinn&&&&&(((>>),  - -'(  - -sA B"ABBB3B4B B"BBBB"returnNoner.zbytes | bytearray | memoryviewrErF)r7 int | NonerErFr7rHrErr)) __name__ __module__ __qualname__r"r'r*r/r8r<r?rCr#r!rrs*  7 --r#rc^eZdZdZ d d dZd dZd dZd dZd dZdddZ ddd Z y)MemorySendStreamaAn in-memory :class:`~trio.abc.SendStream`. Args: send_all_hook: An async function, or None. Called from :meth:`send_all`. Can do whatever you like. wait_send_all_might_not_block_hook: An async function, or None. Called from :meth:`wait_send_all_might_not_block`. Can do whatever you like. close_hook: A synchronous function, or None. Called from :meth:`close` and :meth:`aclose`. Can do whatever you like. .. attribute:: send_all_hook wait_send_all_might_not_block_hook close_hook All of these hooks are also exposed as attributes on the object, and you can change them at any time. Nctjd|_t|_||_||_||_y)N!another task is using this stream)r r_conflict_detectorr _outgoing send_all_hook"wait_send_all_might_not_block_hook close_hook)r rTrUrVs r!r"zMemorySendStream.__init__os= #("8"8 /# -.*2T/$r#cHK|j5tjd{tjd{|jj ||j |j d{dddy7h7N7#1swYyxYww)z}Places the given data into the object's internal buffer, and then calls the :attr:`send_all_hook` (if any). N)rRr rBrSr/rTr-s r!send_allzMemorySendStream.send_all}s $ $ +""$ $ $""$ $ $ NN  t $!!-((***  + + $ $+  + +P B"BBBB>BBB B"BBBBB"cHK|j5tjd{tjd{|jj d|j |j d{dddy7h7N7#1swYyxYww)znCalls the :attr:`wait_send_all_might_not_block_hook` (if any), and then returns immediately. Nr#)rRr rBrSr/rUrs r!wait_send_all_might_not_blockz.MemorySendStream.wait_send_all_might_not_blocks $ $ @""$ $ $""$ $ $ NN  s #66B==???  @ @ $ $@  @ @rYcr|jj|j|jyy)z^Marks this stream as closed, and then calls the :attr:`close_hook` (if any). N)rSr'rVrs r!r'zMemorySendStream.closes-  ?? & OO  'r#chK|jtjd{y7wz!Same as :meth:`close`, but async.Nr'r rBrs r!aclosezMemorySendStream.aclose!    (202cTK|jj|d{S7w)aRetrieves data from the internal buffer, blocking if necessary. Args: max_bytes (int or None): The maximum amount of data to retrieve. None (the default) means to retrieve all the data that's present (but still blocks until at least one byte is available). Returns: If this stream has been closed, an empty bytearray. Otherwise, the requested data. N)rSrCr6s r!get_datazMemorySendStream.get_datas#^^'' 2222 (&(c8|jj|S)zRetrieves data from the internal buffer, but doesn't block. See :meth:`get_data` for details. Raises: trio.WouldBlock: if no data is available to retrieve. )rSr?r6s r!get_data_nowaitz MemorySendStream.get_data_nowaits~~((33r#)NNN)rTAsyncHook | NonerUrhrVSyncHook | NonerGrDr)rI) rJrKrL__doc__r"rXr[r'r`rdrgrMr#r!rOrOYsR,+/?C&* %' %-= %$ % + @"! 3 4r#rOcNeZdZdZ d d dZd d dZd dZd dZddZd dZ y)MemoryReceiveStreamaAn in-memory :class:`~trio.abc.ReceiveStream`. Args: receive_some_hook: An async function, or None. Called from :meth:`receive_some`. Can do whatever you like. close_hook: A synchronous function, or None. Called from :meth:`close` and :meth:`aclose`. Can do whatever you like. .. attribute:: receive_some_hook close_hook Both hooks are also exposed as attributes on the object, and you can change them at any time. Nctjd|_t|_d|_||_||_y)NrQF)r rrRr _incomingrreceive_some_hookrV)r rorVs r!r"zMemoryReceiveStream.__init__s< #("8"8 /# -. !2$r#cK|j5tjd{tjd{|jrtj|j |j d{|j j|d{}|jrtj|cdddS777R71#1swYyxYww)zCalls the :attr:`receive_some_hook` (if any), and then retrieves data from the internal buffer, blocking if necessary. N)rRr rBrr,rornrC)r r7r.s r! receive_somez MemoryReceiveStream.receive_somes $ $ ""$ $ $""$ $ $||///%%1,,... ++I66D||///   $ $/ 7  sb C&CCCC?CC"C&C'!C C&CCCCC#C&cd|_|jj|j|jyy)zfDiscards any pending data from the internal buffer, and marks this stream as closed. TN)rrnr*rVrs r!r'zMemoryReceiveStream.closes4   %%' ?? & OO  'r#chK|jtjd{y7wr^r_rs r!r`zMemoryReceiveStream.aclose rarbc:|jj|y)z.Appends the given data to the internal buffer.N)rnr/r-s r!put_datazMemoryReceiveStream.put_datas 4 r#c8|jjy)z2Adds an end-of-file marker to the internal buffer.N)rnr'rs r!put_eofzMemoryReceiveStream.put_eofs r#)NN)rorhrVrir)rIrDrG) rJrKrLrjr"rqr'r`rurwrMr#r!rlrls?$/3&* %+ %$ %.! !r#rl)r7c |j|} |s|jy|j | y#tj$rYywxYw#tj $rtj ddwxYw)aTake data out of the given :class:`MemorySendStream`'s internal buffer, and put it into the given :class:`MemoryReceiveStream`'s internal buffer. Args: memory_send_stream (MemorySendStream): The stream to get data from. memory_receive_stream (MemoryReceiveStream): The stream to put data into. max_bytes (int or None): The maximum amount of data to transfer in this call, or None to transfer all available data. Returns: True if it successfully transferred some data, or False if there was no data to transfer. This is used to implement :func:`memory_stream_one_way_pair` and :func:`memory_stream_pair`; see the latter's docstring for an example of how you might use it yourself. FzMemoryReceiveStream was closedNT)rgr r>rwrur,BrokenResourceError)memory_send_streammemory_receive_streamr7r.s r!memory_stream_pumpr|s0!11)<T ! ) ) +  " * *4 0      $ $T''(HItSTs:AAAA*A=cnttdfd dfd }|__fS)uQCreate a connected, pure-Python, unidirectional stream with infinite buffering and flexible configuration options. You can think of this as being a no-operating-system-involved Trio-streamsified version of :func:`os.pipe` (except that :func:`os.pipe` returns the streams in the wrong order – we follow the superior convention that data flows from left to right). Returns: A tuple (:class:`MemorySendStream`, :class:`MemoryReceiveStream`), where the :class:`MemorySendStream` has its hooks set up so that it calls :func:`memory_stream_pump` from its :attr:`~MemorySendStream.send_all_hook` and :attr:`~MemorySendStream.close_hook`. The end result is that data automatically flows from the :class:`MemorySendStream` to the :class:`MemoryReceiveStream`. But you're also free to rearrange things however you like. For example, you can temporarily set the :attr:`~MemorySendStream.send_all_hook` to None if you want to simulate a stall in data transmission. Or see :func:`memory_stream_pair` for a more elaborate example. ctyr))r|) recv_stream send_streamsr!$pump_from_send_stream_to_recv_streamzHmemory_stream_one_way_pair..pump_from_send_stream_to_recv_stream[s ; 4r#cKywr)rM)rsr!*async_pump_from_send_stream_to_recv_streamzNmemory_stream_one_way_pair..async_pump_from_send_stream_to_recv_stream^s ,.s rD)rOrlrTrV)rrrrs @@@r!memory_stream_one_way_pairr@s=0#$K%'K5/!KKAK  ##r#cb|\}}|\}}t||}t||}||fSr)r ) one_way_pair pipe1_send pipe1_recv pipe2_send pipe2_recvstream1stream2s r!_make_stapled_pairrfs? *^J )^J J 3GJ 3G G r#c ttS)a Create a connected, pure-Python, bidirectional stream with infinite buffering and flexible configuration options. This is a convenience function that creates two one-way streams using :func:`memory_stream_one_way_pair`, and then uses :class:`~trio.StapledStream` to combine them into a single bidirectional stream. This is like a no-operating-system-involved, Trio-streamsified version of :func:`socket.socketpair`. Returns: A pair of :class:`~trio.StapledStream` objects that are connected so that data automatically flows from one to the other in both directions. After creating a stream pair, you can send data back and forth, which is enough for simple tests:: left, right = memory_stream_pair() await left.send_all(b"123") assert await right.receive_some() == b"123" await right.send_all(b"456") assert await left.receive_some() == b"456" But if you read the docs for :class:`~trio.StapledStream` and :func:`memory_stream_one_way_pair`, you'll see that all the pieces involved in wiring this up are public APIs, so you can adjust to suit the requirements of your tests. For example, here's how to tweak a stream so that data flowing from left to right trickles in one byte at a time (but data flowing from right to left proceeds at full speed):: left, right = memory_stream_pair() async def trickle(): # left is a StapledStream, and left.send_stream is a MemorySendStream # right is a StapledStream, and right.recv_stream is a MemoryReceiveStream while memory_stream_pump(left.send_stream, right.recv_stream, max_bytes=1): # Pause between each byte await trio.sleep(1) # Normally this send_all_hook calls memory_stream_pump directly without # passing in a max_bytes. We replace it with our custom version: left.send_stream.send_all_hook = trickle And here's a simple test using our modified stream objects:: async def sender(): await left.send_all(b"12345") await left.send_eof() async def receiver(): async for data in right: print(data) async with trio.open_nursery() as nursery: nursery.start_soon(sender) nursery.start_soon(receiver) By default, this will print ``b"12345"`` and then immediately exit; with our trickle stream it instead sleeps 1 second, then prints ``b"1"``, then sleeps 1 second, then prints ``b"2"``, etc. Pro-tip: you can insert sleep calls (like in our example above) to manipulate the flow of data across tasks... and then use :class:`MockClock` and its :attr:`~MockClock.autojump_threshold` functionality to keep your test suite running quickly. If you want to stress test a protocol implementation, one nice trick is to use the :mod:`random` module (preferably with a fixed seed) to move random numbers of bytes at a time, and insert random sleeps in between them. You can also set up a custom :attr:`~MemoryReceiveStream.receive_some_hook` if you want to manipulate things on the receiving side, and not just the sending side. )rrrMr#r!memory_stream_pairrssZ 8 99r#cNeZdZd dZd dZd dZd dZd dZd dZd dZ d dd Z y)_LockstepByteQueuect|_d|_d|_d|_t j |_tjd|_ tjd|_ y)NFzanother task is already sendingz!another task is already receiving) rr_sender_closed_receiver_closed_receiver_waitingr r_waitersr r_send_conflict_detector_receive_conflict_detectorrs r!r"z_LockstepByteQueue.__init__sa[ # %!&((* ','='= -( $+0*@*@ /+ 'r#c8|jjyr))rr&rs r!_something_happenedz&_LockstepByteQueue._something_happeneds   "r#cK |rn<|js |jrn#|jjd{Dt j d{y7"7wr))rrrrAr rB)r fns r! _wait_forz_LockstepByteQueue._wait_fors]t""d&;&;--$$& & &      ' s$AA)A%A)A' A)'A)c2d|_|jyr%)rrrs r! close_senderz_LockstepByteQueue.close_senders"   "r#c2d|_|jyr%)rrrs r!close_receiverz!_LockstepByteQueue.close_receivers $   "r#cKj5jrtjjrtj j rJxj |z c_jjfdd{jrtjj rjrtj dddy7Q#1swYyxYww)Nc"jdk(SNr#rrsr!z-_LockstepByteQueue.send_all..ss):r#) rrr r,rryrrrr-s` r!rXz_LockstepByteQueue.send_alls  ) ) 0""///$$///zz ! JJ$ J  $ $ &..!:; ; ;""///zzd33/// 0 0 < 0 0s0 C3BC'C%AC' C3%C''C0,C3cfKj5jrtjjr&tj d{ dddyj fdd{jrtj dddy7P7,#1swYyxYww)NcjSr))rrsr!rzB_LockstepByteQueue.wait_send_all_might_not_block..s )?)?r#)rrr r,rrBrrs`r!r[z0_LockstepByteQueue.wait_send_all_might_not_blocks  ) ) 0""///$$&&(((  0 0 ..!?@ @ @""///# 0 0) @  0 0sM B1AB%B!B% B1B%6B#7 B% B1!B%#B%%B.*B1NcHKj5|%tj|}|dkr tdjrt j d_j jfdd{d_jrt j jr9jd|}jd|=j|cdddSjsJ dddy7#d_wxYw#1swYyxYww)Nr1r2Tc"jdk7Srrrsr!rz1_LockstepByteQueue.receive_some..sTZZ3->r#Fr#) rr3r4r5rr r,rrrrr)r r7gots` r!rqz_LockstepByteQueue.receive_somes  , , $$NN95 q=$%=>>$$///%)D "  $ $ & /nn%>???).&$$///zzjj),JJz z*((*3  6***9  @).&  sT D"AD,D DD AD& D"0D? D"D  DDDD"rD)rzCallable[[], bool]rErFrGr)r7rHrEzbytes | bytearray) rJrKrLr"rrrrrXr[rqrMr#r!rrs*  # !## 0 0r#rc4eZdZddZddZddZd dZddZy) _LockstepSendStreamc||_yr)_lbqr lbqs r!r"z_LockstepSendStream.__init__'  r#c8|jjyr))rrrs r!r'z_LockstepSendStream.close*s  r#chK|jtjd{y7wr)r_rs r!r`z_LockstepSendStream.aclose-!    rbcVK|jj|d{y7wr))rrXr-s r!rXz_LockstepSendStream.send_all1sii  &&&s )')cTK|jjd{y7wr))rr[rs r!r[z1_LockstepSendStream.wait_send_all_might_not_block4sii55777s (&(NrrrDrG)rJrKrLr"r'r`rXr[rMr#r!rr&s!!'8r#rc.eZdZddZddZddZdd dZy) _LockstepReceiveStreamc||_yr)rrs r!r"z_LockstepReceiveStream.__init__9rr#c8|jjyr))rrrs r!r'z_LockstepReceiveStream.close<s   "r#chK|jtjd{y7wr)r_rs r!r`z_LockstepReceiveStream.aclose?rrbNcTK|jj|d{S7wr))rrqr6s r!rqz#_LockstepReceiveStream.receive_someCs!YY++I6666rerrDr)r)rJrKrLr"r'r`rqrMr#r!rr8s#!7r#rcBt}t|t|fS)a Create a connected, pure Python, unidirectional stream where data flows in lockstep. Returns: A tuple (:class:`~trio.abc.SendStream`, :class:`~trio.abc.ReceiveStream`). This stream has *absolutely no* buffering. Each call to :meth:`~trio.abc.SendStream.send_all` will block until all the given data has been returned by a call to :meth:`~trio.abc.ReceiveStream.receive_some`. This can be useful for testing flow control mechanisms in an extreme case, or for setting up "clogged" streams to use with :func:`check_one_way_stream` and friends. In addition to fulfilling the :class:`~trio.abc.SendStream` and :class:`~trio.abc.ReceiveStream` interfaces, the return objects also have a synchronous ``close`` method. )rrr)rs r!lockstep_stream_one_way_pairrGs".  C s #%;C%@ @@r#c ttS)aCreate a connected, pure-Python, bidirectional stream where data flows in lockstep. Returns: A tuple (:class:`~trio.StapledStream`, :class:`~trio.StapledStream`). This is a convenience function that creates two one-way streams using :func:`lockstep_stream_one_way_pair`, and then uses :class:`~trio.StapledStream` to combine them into a single bidirectional stream. )rrrMr#r!lockstep_stream_pairrbs : ;;r#)rzrOr{rlr7rHrEbool)rEz,tuple[MemorySendStream, MemoryReceiveStream])rz0Callable[[], tuple[SendStreamT, ReceiveStreamT]]rEz]tuple[StapledStream[SendStreamT, ReceiveStreamT], StapledStream[SendStreamT, ReceiveStreamT]])rEzqtuple[StapledStream[MemorySendStream, MemoryReceiveStream], StapledStream[MemorySendStream, MemoryReceiveStream]])rEz tuple[SendStream, ReceiveStream])rEzYtuple[StapledStream[SendStream, ReceiveStream], StapledStream[SendStream, ReceiveStream]])% __future__rr3typingrrrrr r _highlevel_genericr abcr rtyping_extensionsrobjectr__annotations__rrrrfinalrOrlr|rrrrrrrrrMr#r!rs;">>.++ If$5 56 96r6z*)*m:6 )?<-<-~o4zo4 o4dJ-J Jb! #(#.# #  #L#$L B  M:j[[|8*8$ 7] 7A6