U
    d>                     @   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
   @/tmp/pip-unpacked-wheel-mu1yl971/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}|rJ|  t|d }n|jdd}|rnt|trn| }|jdd}|rt|tj	r|  t|d }|rt|tr| }||j
f|j|j|j|rdd	 |D nd|rd
d	 |D nd|||d|jS )zCreate 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 |qS r
   dict.0sr
   r
   r(   
<listcomp>i   s     z)task_message_from_sig.<locals>.<listcomp>c                 S   s   g | ]}t |qS r
   r0   r2   r
   r
   r(   r5   j   s     )r   r"   r#   r   r   r.   r/   r$   )freezer%   popnowr   
isinstancer   	isoformatnumbersRealr   r   r"   r#   )	Zappsigr$   r)   r   r   r,   r.   r/   r
   r
   r(   task_message_from_sigO   s8     r>   c                   @   s    e Zd ZdZdd Zdd ZdS )_ContextMockzDummy 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 r@   r
   )rA   exc_infor
   r
   r(   __exit__|   s    z_ContextMock.__exit__N)__name__
__module____qualname____doc__rB   rD   r
   r
   r
   r(   r?   r   s   r?   c                  O   s:   t | |}|t  d |t  d ||j_d|j_|S )z3Mock that mocks :keyword:`with` statement contexts.rB   rD   N)r?   Zattach_mockrB   Zreturn_valuerD   )r"   r#   objr
   r
   r(   ContextMock   s    
rJ   )Nr
   NNNNNN)Nr
   NNNN)rH   r;   r   r   typingr   r   r   Zunittest.mockr   r   r   Zcelery.canvasr	   r)   r*   r>   r?   rJ   r
   r
   r
   r(   <module>   s2           
$      
"#