o wvXj/ @sdZddlmZmZddlZddlmZmZzddlmZWn!e y?zddl mZWne y<ddl mZYnwYnw  dddZ  dd d Z d e fd d ZdS)zUseful mocks for unit testing.)absolute_importunicode_literalsN)datetime timedelta)Mockc Ks|sin|}ddlm} ddlm} |p| }td|d} |||d| _|||d} | j| | ||| fdd \| _| _ | _ ||| f| _ | S) z)Create task message in protocol 2 format.ruuiddumpsTaskMessage-{0}name)idtaskshadow) callbackserrbackschainjson) serializer) celeryr kombu.serializationr rformatheadersupdate content_typecontent_encodingbodypayload)rrargskwargsrrrrutcoptionsr r messageembedrrY/var/www/html/myproject/venv/lib/python3.10/site-packages/celery/contrib/testing/mocks.py TaskMessages        r'c Ks~|sin|}ddlm}ddlm} |p|}td|d} i| _||||||d| _| j|| | j\| _ | _ | _ | S)z)Create task message in protocol 1 format.rrr r r )rrr r!rr) rr rr rrrrrrrr) rrr r!rrrr#r r r$rrr& TaskMessage1)s$     r(Tc Cs||jdd}|jdd}|jdd}|r%|t|d}n|jdd}|r7t|tr7|}|jdd}|rOt|tj rO|t|d}|rZt|trZ|}||j f|j |j |j |rmdd |Dnd|rwd d |Dnd|||d |jS) 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') linkN link_error countdown)secondsetaexpirescSg|]}t|qSrdict.0srrr& ]z)task_message_from_sig..cSr/rr0r2rrr&r5^r6)rr r!rrr-r.r")freezer#popnowr isinstancer isoformatnumbersRealrrr r!) appsigr"r'rrr+r-r.rrr&task_message_from_sigCs6r@)NrNNNNNN)NrNNNN)__doc__ __future__rrr<rrcaser ImportError unittest.mockmockr'r(r@rrrr&s*