U gZ@s|UddlmZddlZddlmZmZmZmZddlm Z m Z ddl m Z ddl mZmZerjddlmZegeefZd ed <egefZd ed <ed ed Zeded ZGdddZe jGdddeZe jGdddeZdddddddddZddddZdd d!d"d#Zd$dd%d&ZGd'd(d(Z Gd)d*d*eZ!Gd+d,d,eZ"d-dd.d/Z#d0dd1d2Z$dS)3) annotationsN) TYPE_CHECKING AwaitableCallableTypeVar)_core_util StapledStream) ReceiveStream SendStream) TypeAliasr AsyncHookSyncHook SendStreamT)boundReceiveStreamTc@seZdZddddZddddZddddZd dd d d Zd ddddZd ddddZdd ddddZ dd ddddZ dS)_UnboundedByteQueueNonereturncCs(t|_d|_t|_td|_dS)NFz%another task is already fetching data) bytearray_data_closedr ParkingLot_lotr ConflictDetector _fetch_lockselfr!@/tmp/pip-unpacked-wheel-ks04xdmi/trio/testing/_memory_streams.py__init__s  z_UnboundedByteQueue.__init__cCsd|_|jdSNT)rr unpark_allrr!r!r"close'sz_UnboundedByteQueue.closecCst|_|dSN)rrr&rr!r!r"close_and_wipe+sz"_UnboundedByteQueue.close_and_wipebytes | bytearray | memoryviewdatarcCs,|jrtd|j|7_|jdS)Nzvirtual connection closed)rrClosedResourceErrorrrr%r r+r!r!r"put/s z_UnboundedByteQueue.put int | None max_bytesrcCs*|dkr dSt|}|dkr&tddS)Nmax_bytes must be >= 1)operatorindex ValueErrorr r1r!r!r"_check_max_bytes5s  z$_UnboundedByteQueue._check_max_bytesrcCsX|js|jst|dkr"t|j}|jrN|jd|}|jd|=|sJt|StSdSr')rrAssertionErrorlenr)r r1chunkr!r!r" _get_impl<s  z_UnboundedByteQueue._get_implNc CsD|j4|||js$|js$tj||W5QRSQRXdSr')rr8rrr WouldBlockr<r7r!r!r" get_nowaitHs   z_UnboundedByteQueue.get_nowaitc s^|jN|||js0|js0|jIdHntIdH||W5QRSQRXdSr') rr8rrrparkr checkpointr<r7r!r!r"getOs   z_UnboundedByteQueue.get)N)N) __name__ __module__ __qualname__r#r&r(r.r8r<r>rAr!r!r!r"rs  rc@seZdZdZdddddddZdd d d d Zd d ddZd d ddZd d ddZddddddZ ddddddZ dS)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. NAsyncHook | NoneSyncHook | None) send_all_hook"wait_send_all_might_not_block_hook close_hookcCs*td|_t|_||_||_||_dS)N!another task is using this stream)r r_conflict_detectorr _outgoingrHrIrJ)r rHrIrJr!r!r"r#oszMemorySendStream.__init__r)rr*c sV|jFtIdHtIdH|j||jdk rH|IdHW5QRXdS)z}Places the given data into the object's internal buffer, and then calls the :attr:`send_all_hook` (if any). N)rLrr@rMr.rHr-r!r!r"send_all}s   zMemorySendStream.send_allrc sV|jFtIdHtIdH|jd|jdk rH|IdHW5QRXdS)znCalls the :attr:`wait_send_all_might_not_block_hook` (if any), and then returns immediately. N)rLrr@rMr.rIrr!r!r"wait_send_all_might_not_blocks   z.MemorySendStream.wait_send_all_might_not_blockcCs |j|jdk r|dS)z^Marks this stream as closed, and then calls the :attr:`close_hook` (if any). N)rMr&rJrr!r!r"r&s  zMemorySendStream.closecs|tIdHdSz!Same as :meth:`close`, but async.Nr&rr@rr!r!r"acloseszMemorySendStream.acloser/rr0cs|j|IdHS)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)rMrAr7r!r!r"get_dataszMemorySendStream.get_datacCs |j|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. )rMr>r7r!r!r"get_data_nowaits z MemorySendStream.get_data_nowait)NNN)N)N) rBrCrD__doc__r#rNrPr&rSrTrUr!r!r!r"rEYsrEc@sneZdZdZddddddZddd d d d Zd dddZd dddZdd dddZd dddZ dS)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. NrFrG)receive_some_hookrJcCs*td|_t|_d|_||_||_dS)NrKF)r rrLr _incomingrrXrJ)r rXrJr!r!r"r#szMemoryReceiveStream.__init__r/rr0c s|jptIdHtIdH|jr0tj|jdk rH|IdH|j|IdH}|jrftj|W5QRSQRXdS)zCalls the :attr:`receive_some_hook` (if any), and then retrieves data from the internal buffer, blocking if necessary. N)rLrr@rr,rXrYrA)r r1r+r!r!r" receive_somes z MemoryReceiveStream.receive_somerrcCs&d|_|j|jdk r"|dS)zfDiscards any pending data from the internal buffer, and marks this stream as closed. TN)rrYr(rJrr!r!r"r&s  zMemoryReceiveStream.closecs|tIdHdSrQrRrr!r!r"rS szMemoryReceiveStream.acloser)r*cCs|j|dS)z.Appends the given data to the internal buffer.N)rYr.r-r!r!r"put_dataszMemoryReceiveStream.put_datacCs|jdS)z2Adds an end-of-file marker to the internal buffer.N)rYr&rr!r!r"put_eofszMemoryReceiveStream.put_eof)NN)N) rBrCrDrVr#rZr&rSr[r\r!r!r!r"rWs  rW)r1r/bool)memory_send_streammemory_receive_streamr1rcCslz||}Wntjk r&YdSXz|s8|n ||Wn"tjk rftddYnXdS)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)rUrr=r\r[r,BrokenResourceError)r^r_r1r+r!r!r"memory_stream_pumps raz,tuple[MemorySendStream, MemoryReceiveStream]rcsFttddfdd ddfdd }|__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. rrcstdSr')rar!) recv_stream send_streamr!r"$pump_from_send_stream_to_recv_stream[szHmemory_stream_one_way_pair..pump_from_send_stream_to_recv_streamcs dSr'r!r!)rdr!r"*async_pump_from_send_stream_to_recv_stream^szNmemory_stream_one_way_pair..async_pump_from_send_stream_to_recv_stream)rErWrHrJ)rer!)rdrbrcr"memory_stream_one_way_pair@srfz0Callable[[], tuple[SendStreamT, ReceiveStreamT]]z]tuple[StapledStream[SendStreamT, ReceiveStreamT], StapledStream[SendStreamT, ReceiveStreamT]]) one_way_pairrcCs0|\}}|\}}t||}t||}||fSr'r )rgZ pipe1_sendZ pipe1_recvZ pipe2_sendZ pipe2_recvZstream1Zstream2r!r!r"_make_stapled_pairfs     rhzqtuple[StapledStream[MemorySendStream, MemoryReceiveStream], StapledStream[MemorySendStream, MemoryReceiveStream]]cCsttS)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. )rhrfr!r!r!r"memory_stream_pairssMric@seZdZddddZddddZdddd d Zddd d Zddd dZdddddZddddZ ddddddZ dS)_LockstepByteQueuerrcCs@t|_d|_d|_d|_t|_t d|_ t d|_ dS)NFzanother task is already sendingz!another task is already receiving) rr_sender_closed_receiver_closed_receiver_waitingrr_waitersr r_send_conflict_detector_receive_conflict_detectorrr!r!r"r#s z_LockstepByteQueue.__init__cCs|jdSr')rnr%rr!r!r"_something_happenedsz&_LockstepByteQueue._something_happenedzCallable[[], bool])fnrcs:|rq(|js(|jrq(|jIdHqtIdHdSr')rkrlrnr?rr@)r rrr!r!r" _wait_fors  z_LockstepByteQueue._wait_forcCsd|_|dSr$)rkrqrr!r!r" close_sendersz_LockstepByteQueue.close_sendercCsd|_|dSr$)rlrqrr!r!r"close_receiversz!_LockstepByteQueue.close_receiverr)r*c sjtjrtjjr tjjr*tj|7_ fddIdHjrdtjjrvjrvtjW5QRXdS)Ncs jdkSNrOrr!rr!r"rOz-_LockstepByteQueue.send_all..) rorkrr,rlr`rr9rqrsr-r!rr"rNs  z_LockstepByteQueue.send_allc shjXjrtjjr6tIdHW5QRdSfddIdHjrZtjW5QRXdS)NcsjSr')rmr!rr!r"rxrOzB_LockstepByteQueue.wait_send_all_might_not_block..)rorkrr,rlr@rsrr!rr"rPsz0_LockstepByteQueue.wait_send_all_might_not_blockNr/bytes | bytearrayr0c sj|dk r*t|}|dkr*tdjr6tjd_z fddIdHW5d_Xjrvtjj rj d|}j d|=|W5QRSj st W5QRdSW5QRXdS)Nr2r3TFcs jdkSrvrwr!rr!r"rxrOz1_LockstepByteQueue.receive_some..rO) rpr4r5r6rlrr,rmrqrsrrkr9)r r1gotr!rr"rZs*   z_LockstepByteQueue.receive_some)N) rBrCrDr#rqrsrtrurNrPrZr!r!r!r"rjs   rjc@sTeZdZddddZddddZddd d Zd dd d dZddddZdS)_LockstepSendStreamrjlbqcCs ||_dSr'_lbqr r}r!r!r"r#'sz_LockstepSendStream.__init__rrcCs|jdSr')rrtrr!r!r"r&*sz_LockstepSendStream.closecs|tIdHdSr'rRrr!r!r"rS-sz_LockstepSendStream.acloser)r*cs|j|IdHdSr')rrNr-r!r!r"rN1sz_LockstepSendStream.send_allcs|jIdHdSr')rrPrr!r!r"rP4sz1_LockstepSendStream.wait_send_all_might_not_blockN)rBrCrDr#r&rSrNrPr!r!r!r"r{&s r{c@sHeZdZddddZddddZddd d Zdd d dddZd S)_LockstepReceiveStreamrjr|cCs ||_dSr'r~rr!r!r"r#9sz_LockstepReceiveStream.__init__rrcCs|jdSr')rrurr!r!r"r&<sz_LockstepReceiveStream.closecs|tIdHdSr'rRrr!r!r"rS?sz_LockstepReceiveStream.acloseNr/ryr0cs|j|IdHSr')rrZr7r!r!r"rZCsz#_LockstepReceiveStream.receive_some)N)rBrCrDr#r&rSrZr!r!r!r"r8srz tuple[SendStream, ReceiveStream]cCst}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. )rjr{rr|r!r!r"lockstep_stream_one_way_pairGsrzYtuple[StapledStream[SendStream, ReceiveStream], StapledStream[SendStream, ReceiveStream]]cCsttS)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. )rhrr!r!r!r"lockstep_stream_pairbsr)% __future__rr4typingrrrrrr Z_highlevel_genericr abcr r Ztyping_extensionsrobjectr__annotations__rrrrfinalrErWrarfrhrirjr{rrrr!r!r!r"s6     ?rQ&& U^