File: //usr/local/CyberPanel/lib64/python3.10/site-packages/asgiref/__pycache__/testing.cpython-310.pyc
o
��hE � @ sB d dl Z d dlZd dlZddlmZ ddlmZ G dd� d�ZdS )� N� )�guarantee_single_callable)�timeoutc @ st e Zd ZdZdd� Zedd� �Zedd� �Zedd � �Zdd d �Z d dd�Z
dd� Z dd� Z ddd�Z
d dd�ZdS ) �ApplicationCommunicatorz}
Runs an ASGI application in a test mode, allowing sending of
messages to it and retrieval of messages it sends.
c C s& d | _ t|�| _|| _d | _d | _d S �N)�_futurer � application�scope� _input_queue�
_output_queue)�selfr r � r
�E/usr/local/CyberPanel/lib/python3.10/site-packages/asgiref/testing.py�__init__ s
z ApplicationCommunicator.__init__c C � | j d u r
t�� | _ | j S r )r
�asyncio�Queue�r r
r
r � input_queue �
z#ApplicationCommunicator.input_queuec C r r )r r r r r
r
r � output_queue r z$ApplicationCommunicator.output_queuec C s8 | j d u rt�� �tj| �| j| jj | j
j ��| _ | j S r ) r � contextvars�Context�runr � create_taskr r r �getr �putr r
r
r �future$ s
��z ApplicationCommunicator.futurer c � s� �zXt |�4 I dH �# z
| jI dH | j�� W n
tjy" Y nw W d �I dH n 1 I dH s3w Y W | j�� sX| j�� z | jI dH W dS tjyW Y dS w dS | j�� sw| j�� z| jI dH W w tjyv Y w w w )zV
Waits for the application to stop itself and returns any exceptions.
N)�
async_timeoutr �resultr �CancelledError�done�cancel)r r r
r
r �wait3 s6 � ��(��
��
��z ApplicationCommunicator.waitTc C s: | j }|d u r d S |�� s|�� d S |r|�� d S d S r )r r! r" r )r �
exceptionsr r
r
r �stopF s �z ApplicationCommunicator.stopc C s( z | j dd� W d S ty Y d S w )NF)r$ )r% � RuntimeErrorr r
r
r �__del__Q s
�z ApplicationCommunicator.__del__c � s, �| j �� r | j �� | j�|�I dH dS )z;
Sends a single message to the application
N)r r! r r r )r �messager
r
r �
send_inputY s �
z"ApplicationCommunicator.send_inputc � s� �| j �� r | j �� z*t|�4 I dH � | j�� I dH W d �I dH W S 1 I dH s.w Y W dS tjyh } z&| j �� rJ| j �� |�| j �� z | j I dH W |� tj yc Y |�w d}~ww )zX
Receives a single message from the application, with optional timeout.
N)
r r! r r r r r � TimeoutErrorr" r )r r �er
r
r �receive_outputd s* �
4�
�����z&ApplicationCommunicator.receive_output皙�����?�{�G�z�?c � sf �| j �� r | j �� t�� }t�� | |k r.| j�� s dS t�|�I dH t�� | |k s| j�� S )zO
Checks that there is no message to receive in the given time.
FN) r r! r �time� monotonicr �emptyr �sleep)r r �interval�startr
r
r �receive_nothing{ s �
�
z'ApplicationCommunicator.receive_nothingN)r )T)r- r. )�__name__�
__module__� __qualname__�__doc__r �propertyr r r r# r% r' r) r, r5 r
r
r
r r s
r )r r r/ �
compatibilityr r r r r
r
r
r �<module> s