File: //usr/local/lib/python3.9/site-packages/celery/contrib/testing/__pycache__/mocks.cpython-39.pyc
a
X>hV � @ s� d Z ddlZddlmZmZ ddlmZmZmZ ddlm Z ddl
m Z ddl m
Z
dd d
�Zdd d �Zd
efdd�ZG dd� de �Zdd� ZdS )z Useful mocks for unit testing.� N)�datetime� timedelta)�Any�Mapping�Sequence)�Mock)�Celery)� Signature� c K s� |si n|}ddl m}
ddlm} |p,| � }td|� �d�} || |d�| _|||d�}
| j�| � |
|||
fdd �\| _| _| _ |||
f| _
| S )
z)Create task message in protocol 2 format.r ��dumps��uuid� TaskMessage-��name)�id�task�shadow)� callbacks�errbacks�chain�json)�
serializer) �kombu.serializationr �celeryr r �headers�update� content_type�content_encoding�body�payload)r r �args�kwargsr r r r �utc�optionsr r �message�embedr
r
�F/usr/local/lib/python3.9/site-packages/celery/contrib/testing/mocks.py� TaskMessage s
�
� r) c K s~ |si n|}ddl m} ddlm} |p,| � }td|� �d�}
i |
_| |||||d�|
_|
j�|� ||
j�\|
_|
_ |
_
|
S )z)Create task message in protocol 1 format.r r r
r r )r r r" r# r r ) r r r r r r r! r r r r ) r r r" r# r r r r% r r r& r
r
r( � TaskMessage1- s$
� �r* Tc C s |� � |j�dd�}|j�dd�}|j�dd�}|rJ| �� t|d� }n|j�dd�}|rnt|t�rn|�� }|j�dd�}|r�t|tj �r�| �� t|d� }|r�t|t�r�|�� }||j
f|j |j |j
|r�dd � |D �nd|r�d
d � |D �nd|||d �|j��S ) z�Create task message from :class:`celery.Signature`.
Example:
>>> m = task_message_from_sig(app, add.s(2, 2))
>>> amqp_client.basic_publish(m, exchange='ex', routing_key='rkey')
�linkNZ
link_error� countdown)�seconds�eta�expiresc S s g | ] }t |��qS r
��dict��.0�sr
r
r( �
<listcomp>i � z)task_message_from_sig.<locals>.<listcomp>c S s g | ] }t |��qS r
r0 r2 r
r
r( r5 j r6 )r r" r# r r r. r/ r$ )�freezer% �pop�nowr �
isinstancer � isoformat�numbers�Realr r r" r# ) Zapp�sigr$ r) r r r, r. r/ r
r
r( �task_message_from_sigO s6 ���r? c @ s e Zd ZdZdd� Zdd� ZdS )� _ContextMockz�Dummy class implementing __enter__ and __exit__.
The :keyword:`with` statement requires these to be implemented
in the class, not just the instance.
c C s | S �Nr
)�selfr
r
r( � __enter__y s z_ContextMock.__enter__c G s d S rA r
)rB �exc_infor
r
r( �__exit__| s z_ContextMock.__exit__N)�__name__�
__module__� __qualname__�__doc__rC rE r
r
r
r( r@ r s r@ c O s> t | i |��}|�t � d� |�t � d� ||j_d|j_|S )z3Mock that mocks :keyword:`with` statement contexts.rC rE N)r@ Z attach_mockrC Z return_valuerE )r" r# �objr
r
r( � ContextMock� s rK )Nr
NNNNNN)Nr
NNNN)rI r< r r �typingr r r Z
unittest.mockr r r Z
celery.canvasr r) r* r? r@ rK r
r
r
r( �<module> s2 �
$ �
"#