o
    ›¨ÊhV  ã                   @   s®   d Z ddlZddlmZmZ ddlmZmZmZ ddlm	Z	 ddl
mZ ddlmZ 								dd	d
„Z						ddd„Zdefdd„ZG dd„ de	ƒZdd„ ZdS )zUseful mocks for unit testing.é    N)ÚdatetimeÚ	timedelta)ÚAnyÚMappingÚSequence)ÚMock)ÚCelery)Ú	Signature© c	                 K   s’   |si n|}ddl m}
 ddlm} |p|ƒ }td|› �d�}|| |dœ|_|||dœ}|j |	¡ |
|||fdd	�\|_|_|_	|||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
   úN/var/www/html/env/lib/python3.10/site-packages/celery/contrib/testing/mocks.pyÚTaskMessage   s    
ý
ÿr)   c                 K   s~   |si n|}ddl m} ddlm}	 |p|	ƒ }td|› �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 dd¡}|j dd¡}|j dd¡}|r%|  ¡ t|d� }n|j dd¡}|r7t|tƒr7| ¡ }|j dd¡}|rOt|tj	ƒrO|  ¡ t|d� }|rZt|tƒrZ| ¡ }||j
f|j|j|j|rmdd	„ |D ƒnd|rwd
d	„ |D ƒnd|||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')
    ÚlinkNÚ
link_errorÚ	countdown)ÚsecondsÚetaÚexpiresc                 S   ó   g | ]}t |ƒ‘qS r
   ©Údict©Ú.0Úsr
   r
   r(   Ú
<listcomp>i   ó    z)task_message_from_sig.<locals>.<listcomp>c                 S   r1   r
   r2   r4   r
   r
   r(   r7   j   r8   )r   r"   r#   r   r   r/   r0   r$   )Úfreezer%   ÚpopÚnowr   Ú
isinstancer   Ú	isoformatÚnumbersÚRealr   r   r"   r#   )	ÚappÚsigr$   r)   r   r   r-   r/   r0   r
   r
   r(   Útask_message_from_sigO   s6   ÿùørB   c                   @   s    e Zd ZdZdd„ Zdd„ ZdS )Ú_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   ó   z_ContextMock.__enter__c                 G   s   d S rD   r
   )rE   Úexc_infor
   r
   r(   Ú__exit__|   rG   z_ContextMock.__exit__N)Ú__name__Ú
__module__Ú__qualname__Ú__doc__rF   rI   r
   r
   r
   r(   rC   r   s    rC   c                  O   s>   t | i |¤Ž}| t ƒ d¡ | t ƒ d¡ ||j_d|j_|S )z3Mock that mocks :keyword:`with` statement contexts.rF   rI   N)rC   Úattach_mockrF   Úreturn_valuerI   )r"   r#   Úobjr
   r
   r(   ÚContextMock€   s   rQ   )Nr
   NNNNNN)Nr
   NNNN)rM   r>   r   r   Útypingr   r   r   Úunittest.mockr   r   r   Úcelery.canvasr	   r)   r*   rB   rC   rQ   r
   r
   r
   r(   Ú<module>   s4    
÷$
ù"#