File: //usr/local/lib/python3.9/site-packages/celery/contrib/testing/__pycache__/app.cpython-39.pyc
a
X>h( � @ s� d Z ddlZddlmZ ddlmZ ddlmZ ddlm Z m
Z
dddhd d
d d dd
�Z G dd� d�Z G dd� dee j
��Zddddeddfdd�Zedd� �Zeddd��ZdS )z-Create Celery app instances used for testing.� N)�contextmanager)�deepcopy)�symbol_by_name)�Celery�_stateF�jsonT�UTCz memory://zcache+memory://)Zworker_hijack_root_loggerZworker_log_colorZaccept_contentZ
enable_utc�timezone�
broker_url�result_backendZbroker_heartbeatc @ s e Zd ZdZdd� ZdS )�Trapz�Trap that pretends to be an app but raises an exception instead.
This to protect from code that does not properly pass app instances,
then falls back to the current_app.
c C s( |dks|dkrd S t |� td��d S )N�
_is_coroutine�__func__zTest depends on current_app)�print� RuntimeError)�self�name� r �D/usr/local/lib/python3.9/site-packages/celery/contrib/testing/app.py� __getattr__ s zTrap.__getattr__N)�__name__�
__module__� __qualname__�__doc__r r r r r r s r c s e Zd ZdZ� fdd�Z� ZS )� UnitLoggingz)Sets up logging for the test application.c s t � j|i |�� d| _d S )NT)�super�__init__Z
already_setup)r �args�kwargs�� __class__r r r * s zUnitLogging.__init__)r r r r r �
__classcell__r r r r r ' s r c
K s� ddl m} ttt�fi |p i ��}|dur8|�dd� |durL|�dd� |rTdn|}t| p`df||||d�|��} | �|� | S )zApp used for testing.� )�tasksNr
r z celery.tests)�set_as_current�log�broker�backend)� r# �dictr �DEFAULT_TEST_CONFIG�popr Z add_defaults)
r �configZenable_loggingr$ r% r' r& r r# Ztest_appr r r �TestApp/ s& ���
r- c # sN t � � tj}t�� � G � fdd�d�}|� t_zdV W |t_n|t_0 dS )z�Contextmanager that installs the trap app.
The trap means that anything trying to use the current or default app
will raise an exception.
c s e Zd Z� ZdS )zset_trap.<locals>.NonTLSN)r r r � current_appr �Ztrapr r �NonTLSO s r0 N)r r �_tls�set_default_app)�appZprev_tlsr0 r r/ r �set_trapD s
r4 c c s� t �� }t j}tt j�}t�t j�}zl|rZt| �� dV W d � q`1 sN0 Y ndV W t � |� |t j
_ | |ur�| � � |t _|t _n0t � |� |t j
_ | |ur�| � � |t _|t _0 dS )zWSetup default app for testing.
Ensures state is clean after the test returns.
N)
r Zget_current_appZ default_app�setZ_on_app_finalizers�weakref�WeakSetZ_appsr4 r2 r1 r. �close)r3 Zuse_trapZprev_current_appZprev_default_appZprev_finalizersZ prev_appsr r r �setup_default_appY s*
&
�
r9 )F)r r6 �
contextlibr �copyr Zkombu.utils.importsr Zceleryr r r* r Zlog_clsr r- r4 r9 r r r r �<module> s. � �