File: //usr/local/lib/python3.9/site-packages/celery/contrib/testing/__pycache__/manager.cpython-39.pyc
a
X>h�! � @ s� d Z ddlZddlZddlmZ ddlmZ ddlmZ ddl m
Z
m Z m Z m
Z
mZmZ ddlmZ ddlmZ dd lmZ dd
lmZ dd lmZmZ dd lm Z dd
l m Z dZ ee dd�Z G dd� de!�Z"G dd� d�Z#G dd� de#�Z$dS )z Integration testing utilities.� N)� defaultdict)�partial)�count)�Any�Callable�Dict�Sequence�TextIO�Tuple)�ContentDisallowed��retry_over_time��states)� TimeoutError)� AsyncResult� ResultSet)�truncate)�humanize_secondsz4Still waiting for {0}. Trying again {when}: {exc!r}T)� microsecondsc @ s e Zd ZdZdS )�Sentinelz Signifies the end of something.N)�__name__�
__module__� __qualname__�__doc__� r r �H/usr/local/lib/python3.9/site-packages/celery/contrib/testing/manager.pyr s r c @ s� e Zd ZdZd=dd�Zd>dd �Zd
d � Zd?dd�Zd@dd�Zdd� Z dAdd �Z
dBd d �Z dCd d!�Z dDd"d#�Z
dEd%d&�ZdFd(d)�ZdGd+d,�ZdHd-d.�Zed/d0� �ZdId1d2�Zd3d4� Zd5d6� ZdJd7d8�Zd9d:� Zd;d<� ZdS )K� ManagerMixinz.Mixin that adds :class:`Manager` capabilities.� �@FNc C sF |d u rt jn|| _|d u r"t jn|| _| j�� j| _|| _|| _d S �N) �sys�stdout�stderr�app�
connectionZ recoverable_connection_errors�
connerrors�
block_timeout�no_join)�selfr& r' r! r" r r r �
_init_manager s
zManagerMixin._init_manager�-c C s t |� |� �| jd� d S )N)�file)�printr! )r( �s�sepr r r �remark( s zManagerMixin.remarkc C s dd� |D �S )Nc S s g | ]}|j |jjvr|j �qS r )�id�backend�_cache)�.0�resr r r �
<listcomp>. � z0ManagerMixin.missing_results.<locals>.<listcomp>r )r( �rr r r �missing_results, s z ManagerMixin.missing_results�thingr �
皙�����?� �?� @c s@ |si n|}� ���fdd�}
�j ||f|||
||| d�| ��S )z�Wait for event to happen.
The `catch` argument specifies the exception that means the event
has not happened yet.
c s> t |�}�r*��tj� t|dd�| d�� �r:�| ||� |S )N�in� )�when�exc)�next�warn�E_STILL_WAITING�formatr )rA Z intervals�retries�interval��desc� emit_warning�errbackr( r r �on_errorG s � z'ManagerMixin.wait_for.<locals>.on_error)�args�kwargsrK � max_retries�interval_start�
interval_stepr )r( �fun�catchrI rM rN rK rO rP rQ � interval_maxrJ �optionsrL r rH r �wait_for0 s
���zManagerMixin.wait_for� �{�G�z�?� �?c
K sB z | j ||||||||d�W S |y. Y n0 td|� ���dS )z;Make sure something does not happen (at least for a while).)rI rO rP rQ rT rJ zShould not have happened: N)rV �AssertionError)
r( rR rS rI rO rP rQ rT rJ rU r r r �ensure_not_for_a_whileY s � z#ManagerMixin.ensure_not_for_a_whilec O s t |i |��S r r )r( rM rN r r r r
j s z ManagerMixin.retry_over_timec s | j r
d S t|t�s"| j�|g�}g � � fdd�}|r>t|�ntd�D ]�}g � d d �< z |jf ||d�|��W S tjt fy� } zH| �
|�}| � d� t
|�t
� � t
|�td�|��|�d� W Y d }~qFd }~0 | j�y } z | � d|��d� W Y d }~qFd }~0 0 qFtd ��d S )
Nc s � � | � d S r )�append)�task_id�value�Zreceivedr r � on_resultt s z$ManagerMixin.join.<locals>.on_resultr )�callback� propagatez#Still waiting for {}/{}: [{}]: {!r}z, �!zjoin: connection lost: z!Test failed: Missing task results)r' �
isinstancer r# �ranger �get�socket�timeoutr r8 r/ rE �lenr �joinr% rZ ) r( r7 rb rO rN r` �irA Z waiting_forr r_ r rj m s,
��*zManagerMixin.join� @c C s | j jj|d�S �N�rh )r# �control�inspect)r( rh r r r rp � s zManagerMixin.inspectc c s& | � |�j|� pi }|�� E d H d S r )rp Z
query_task�items)r( �idsrh �tasksr r r � query_tasks� s zManagerMixin.query_tasksc C sH t t�}| j||d�D ],\}}|�� D ]\}\}}|| �|� q&q|S rm )r �setrt rq �add) r( rr rh r �hostnameZreplyr] �state�_r r r �query_task_states� s
z ManagerMixin.query_task_states� waiting for tasks to be acceptedc K s | j | j|f||d�|��S �N)rG rI )�assert_task_worker_state� is_accepted�r( rr rG rI �policyr r r �assert_accepted� s ���z ManagerMixin.assert_accepted� waiting for tasks to be receivedc K s | j | j|f||d�|��S r| )r} � is_receivedr r r r �assert_received� s ���z ManagerMixin.assert_received�,waiting for tasks to be started or completedc K s | j | j|f||d�|��S r| )� assert_task_state_from_result�is_result_task_in_progress)r( Z
async_resultsrG rI r� r r r �,assert_result_tasks_in_progress_or_completed� s ���z9ManagerMixin.assert_result_tasks_in_progress_or_completedc K s$ | j t| j|||d�tffi |��S rm �rV r �
true_or_raiser )r( rR �resultsrG r� r r r r� � s ��z*ManagerMixin.assert_task_state_from_resultc s" t jt jf� t� fdd�| D ��S )Nc 3 s | ]}|j � v V qd S r )rx )r3 �result�Zpossible_statesr r � <genexpr>� r6 z:ManagerMixin.is_result_task_in_progress.<locals>.<genexpr>)r ZSTARTED�SUCCESS�all)r� rN r r� r r� � s z'ManagerMixin.is_result_task_in_progressc K s$ | j t| j|||d�tffi |��S rm r� )r( rR rr rG r� r r r r} � s ��z%ManagerMixin.assert_task_worker_statec K s | j g d�|fi |��S )N)�reserved�active�ready��_ids_matches_state�r( rr rN r r r r� � s
��zManagerMixin.is_receivedc K s | j ddg|fi |��S )Nr� r� r� r� r r r r~ � s zManagerMixin.is_acceptedc s& | j ||d��t� �fdd�|D ��S )Nrn c 3 s2 | ]*� t � fd d��fdd��D �D ��V qdS )c 3 s | ]}� |v V qd S r r )r3 r- ��tr r r� � r6 z<ManagerMixin._ids_matches_state.<locals>.<genexpr>.<genexpr>c s g | ] }� | �qS r r )r3 �kr r r r5 � r6 z=ManagerMixin._ids_matches_state.<locals>.<genexpr>.<listcomp>N)�any)r3 ��expected_statesr r� r r� � s �z2ManagerMixin._ids_matches_state.<locals>.<genexpr>)rz r� )r( r� rr rh r r� r r� � s �z ManagerMixin._ids_matches_statec O s ||i |��}|st � �|S r )r )r( rR rM rN r4 r r r r� � s zManagerMixin.true_or_raisec C s� | j j}| j �� �x}|j|d�}|dkrq,q|�� }||_z tdd� |�� �� D ��}W n tyn Y q|Y n0 |dkr:q|q:W d � n1 s�0 Y d S )N)r$ r c s s | ]}t |�V qd S r )ri )r3 r� r r r r� � r6 z/ManagerMixin.wait_until_idle.<locals>.<genexpr>) r# ro r$ �purgerp �sumr� �valuesr )r( ro r$ r rp r r r �wait_until_idle� s
z ManagerMixin.wait_until_idle)r FNN)r* ) r9 r NNr: r; r< r= F)r9 rW r; rX rY F)Fr: )rl )r< )r< )r< r{ )r< r� )r< r� )r< )r< )r< )r r r r r) r/ r8 rV r[ r
rj rp rt rz r� r� r� r� � staticmethodr� r} r� r~ r� r� r� r r r r r sX �
�
* �
�
�
�
�
r c @ s e Zd ZdZdd� ZdS )�Managerz(Test helpers for task integration tests.c K s || _ | jf i |�� d S r )r# r) )r( r# rN r r r �__init__� s zManager.__init__N)r r r r r� r r r r r� � s r� )%r rg r � collectionsr � functoolsr � itertoolsr �typingr r r r r r
Zkombu.exceptionsr Zkombu.utils.functionalr
Zceleryr Zcelery.exceptionsr Z
celery.resultr r Zcelery.utils.textr Zcelery.utils.timer Z_humanize_secondsrD � Exceptionr r r� r r r r �<module> s& P