o 'hE@sBddlZddlZddlZddlmZddlmZGdddZdS)N)guarantee_single_callable)timeoutc@steZdZdZddZeddZeddZedd Zdd d Z dddZ ddZ ddZ dddZ dddZdS)ApplicationCommunicatorz} Runs an ASGI application in a test mode, allowing sending of messages to it and retrieval of messages it sends. cCs&d|_t||_||_d|_d|_dSN)_futurer applicationscope _input_queue _output_queue)selfrr r C/var/www/html/myenv/lib/python3.10/site-packages/asgiref/testing.py__init__s   z ApplicationCommunicator.__init__cC|jdur t|_|jSr)r asyncioQueuer r r r input_queue  z#ApplicationCommunicator.input_queuecCrr)r rrrr r r output_queuerz$ApplicationCommunicator.output_queuecCs8|jdurttj||j|jj |j j |_|jSr) r contextvarsContextrunr create_taskrr rgetrputrr r rfuture$s zApplicationCommunicator.futurercszXt|4IdH#z |jIdH|jWn tjy"YnwWdIdHn 1IdHs3wYW|jsX|jz |jIdHWdStjyWYdSwdS|jsw|jz|jIdHWwtjyvYwww)zV Waits for the application to stop itself and returns any exceptions. N) async_timeoutrresultrCancelledErrordonecancel)r rr r rwait3s6 (   zApplicationCommunicator.waitTcCs:|j}|dur dS|s|dS|r|dSdSr)rr!r"r)r exceptionsrr r rstopFs  zApplicationCommunicator.stopcCs(z |jddWdStyYdSw)NF)r$)r% RuntimeErrorrr r r__del__Qs  zApplicationCommunicator.__del__cs,|jr |j|j|IdHdS)z; Sends a single message to the application N)rr!rrr)r messager r r send_inputYs  z"ApplicationCommunicator.send_inputc s|jr |jz*t|4IdH|jIdHWdIdHWS1IdHs.wYWdStjyh}z&|jrJ|j||jz |jIdHW|tj ycY|wd}~ww)zX Receives a single message from the application, with optional timeout. N) rr!rrrrr TimeoutErrorr"r )r rer r rreceive_outputds*  4   z&ApplicationCommunicator.receive_output皙?{Gz?csf|jr |jt}t||kr.|jsdSt|IdHt||ks|jS)zO Checks that there is no message to receive in the given time. FN) rr!rtime monotonicremptyrsleep)r rintervalstartr r rreceive_nothing{s    z'ApplicationCommunicator.receive_nothingN)r)T)r-r.)__name__ __module__ __qualname____doc__rpropertyrrrr#r%r'r)r,r5r r r rr s       r)rrr/ compatibilityrrrrr r r rs