File: //usr/local/lib/python3.9/site-packages/celery/contrib/__pycache__/pytest.cpython-39.pyc
a
X>hb � @ s� d Z ddlZddlmZ ddlmZmZmZmZm Z ddl
Z
erZddl m Z ddl
mZ ne Z Zej�d�Zd d
� Zed*d d
��Ze
jdd�dd� �Ze
jdd�dd� �Ze
jdd�dd� �Ze
jdd�dd� �Ze
jdd�dd� �Ze
jdd�dd� �Ze
jdd�d d � �Z e
jdd�d d � �Z e
jdd�d d!� �Z e
�� d"d#� �Z e
jdd�d$d%� �Z e
�� d&d'� �Z!e
�� d(d)� �Z"dS )+z;Fixtures and testing utilities for :pypi:`pytest <pytest>`.� N)�contextmanager)�
TYPE_CHECKING�Any�Mapping�Sequence�Union)�Celery� )�WorkController� NO_WORKERc C s | � dd� dS )z)Register additional pytest configuration.�markerszBcelery(**overrides): override celery configuration for a test caseN)Zaddinivalue_line)�config� r �?/usr/local/lib/python3.9/site-packages/celery/contrib/pytest.py�pytest_configure s �r Fc k sh ddl m}m} |si n|}|f d| |d�|��}|||d�� |V W d � n1 sZ0 Y dS )z=Utility context used to setup Celery app for pytest fixtures.� )�TestApp�setup_default_appF)Zset_as_current�enable_loggingr
)�use_trapN)Z testing.appr r )r r �
parametersr
r r Ztest_appr r r � _create_app s ��r �session)Zscopec C s dS )z�You can override this fixture to enable the app trap.
The app trap raises an exception whenever something attempts
to use the current or default apps.
Fr r r r r �use_celery_app_trap5 s r c c sz | j �d�}t|fi |r |jni ��}tf |||d�|���*}|sR|�� |�� |V W d � n1 sl0 Y dS )z1Session Fixture: Return app for session fixtures.�celery�r r r N)�node�get_closest_marker�dict�kwargsr � set_default� set_current��request�
celery_config�celery_parameters�celery_enable_loggingr �markr
Zappr r r �celery_session_app@ s ��r( c
c sz ddl m} tsv|D ]}|j�|� q|D ]}|�|� q*|j|fd|i|���} | V W d � n1 sl0 Y dS )z?Session Fixture: Start worker that lives throughout test suite.r ��worker�poolN)�testingr* r �loader�import_task_moduleZ
register_task� start_worker)
r# r( �celery_includes�celery_class_tasks�celery_worker_pool�celery_worker_parametersr* �moduleZ
class_task�wr r r �celery_session_workerT s
��r6 c C s dS )z0You can override this fixture to enable logging.Fr r r r r r& l s r& c C s dS )z�You can override this include modules when a worker start.
You can have this return a list of module names to import,
these can be task modules, modules registering signals, and so on.
r r r r r r r0 s s r0 c C s dS )z�You can override this fixture to set the worker pool.
The "solo" pool is used by default, but you can set this to
return e.g. "prefork".
Zsolor r r r r r2 ~ s r2 c C s i S )z�Redefine this fixture to configure the test Celery app.
The config returned by your fixture will then be used
to configure the :func:`celery_app` fixture.
r r r r r r$ � s r$ c C s i S )z�Redefine this fixture to change the init parameters of test Celery app.
The dict returned by your fixture will then be used
as parameters when instantiating :class:`~celery.Celery`.
r r r r r r% � s r% c C s i S )a" Redefine this fixture to change the init parameters of Celery workers.
This can be used e. g. to define queues the worker will consume tasks from.
The dict returned by your fixture will then be used
as parameters when instantiating :class:`~celery.worker.WorkController`.
r r r r r r3 � s
r3 c c sf | j �d�}t|fi |r |jni ��}tf |||d�|���}|V W d � n1 sX0 Y dS )z/Fixture creating a Celery application instance.r r N)r r r r r r" r r r �
celery_app� s ��r7 c C s g S )zARedefine this fixture to register tasks with the test Celery app.r r r r r r1 � s r1 c c sf ddl m} tsb|D ]}|j�|� q|j|fd|i|���}|V W d � n1 sX0 Y dS )zAFixture: Start worker in a thread, stop it when the test returns.r r) r+ N)r, r* r r- r. r/ )r# r7 r0 r2 r3 r* r4 r5 r r r �
celery_worker� s
��r8 c C s | � � dS )z!Fixture that sets app as current.N)r! )r7 r r r �depends_on_current_app� s r9 )FFN)#�__doc__�os�
contextlibr �typingr r r r r Zpytestr r r* r
�object�environ�getr r r Zfixturer r( r6 r& r0 r2 r$ r% r3 r7 r1 r8 r9 r r r r �<module> sR
�