o I<V@sddlmZmZddlZddlmZddlmZddlm Z ddl m Z ddl m ZddlmZdd lmZdd lmZdd lmZdd lmZdd lmZmZddlmZddlmZddZGddde Z!Gddde!Z"eedZ#eedZ$ee dZ%ee dZ&iZ'eddZ(e fddZ)ddZ*Gd d!d!eZ+Gd"d#d#e+Z,Gd$d%d%e+Z-Gd&d'd'e+Z.Gd(d)d)e+Z/Gd*d+d+e+Z0dS),)absolute_importunicode_literalsN)partial) ExceptionInforeverse) HttpResponse)TestCase)TemplateDoesNotExist) deserialize) current_app)statestask) gen_unique_idget_full_cls_name) task_webhook) MockRequestcKs t||dS)Nkwargsr)namerrK/var/www/html/env/lib/python3.10/site-packages/djcelery/tests/test_views.py reversestars rc@eZdZddZdS)MyErrorcCs d|S)Nz$<{0.__class__.__name__}: {0.args!r}>)formatselfrrr__repr__ s zMyError.__repr__N)__name__ __module__ __qualname__rrrrrrs rc@s eZdZdS)MyRetryTaskErrorN)r r!r"rrrrr#$sr#zcelery-is_task_successfulzcelery-task_statusz celery-applyz celery-taskscCst|t|}td<|S)Nresult)intscratch)xyretrrrmytask/sr*cCst||fiSN)type)rbaserrrcreate_exception5sr.c CsJz||jy$}ztj|}|ttjfWYd}~Sd}~wwr+) __class__r backendprepare_exceptionrsysexc_info traceback) exceptionexcrrrcatch_exception9s r7c@s$eZdZddZddZddZdS) ViewTestCasec CsZt|tr|jp |}z|t|d|WdSty,}ztd||d}~ww)Nutf-8z{0}: {1}) isinstancercontent assertEqualr decode TypeErrorr)rjsonpyr6rrrassertJSONEqualCszViewTestCase.assertJSONEqualcGs@ztj|||g|RWdSty|||vYdSwr+)DjangoTestCaseassertInAttributeError assertTrue)rexpectedsourceargsrrrrCJs  zViewTestCase.assertIncGsbztj|||g|RWdSty0|D]\}}|||v||||qYdSwr+)rBassertDictContainsSubsetrDitemsrEr<)rabrHkeyvaluerrrrIPs z%ViewTestCase.assertDictContainsSubsetN)r r!r"rArCrIrrrrr8As r8c@eZdZddZddZdS)test_task_applycCsNdtj_z|jtdtjidd|t ddWdtj_dSdtj_w)NT task_namer?x=4&y=4r$F) r confCELERY_ALWAYS_EAGERclientget task_applyr*rr<r&rrrr test_apply[sztest_task_apply.test_applycCs|dtj_z4d}t|jjtd|idd}z|}Wn ty#Ynw||j dWdtj_dSWdtj_dSdtj_w)NTzxxx.does.not.existrQrrRiF) r rTrUrrVrWrXr r< status_code)rractionresrrr*test_apply_raises_404_on_unregistered_taskes    z:test_task_apply.test_apply_raises_404_on_unregistered_taskN)r r!r"rYr]rrrrrPYs rPc@r)test_registered_taskscCs2|jt}t|jd}|d|ddS)Nr9zcelery.backend_cleanupregular)rVrWregistered_tasksr r;r=rC)rr?tasksrrrtest_list_registered_tasksysz0test_registered_tasks.test_list_registered_tasksN)r r!r"rbrrrrr^ws r^c@rO)test_webhook_taskcCLtdd}tdtddd}||}|dddt|jd dS) NcSs$t|jd}t|jd}||SNr'r()r%GETrequestr'r(rrr add_webhooksz>test_webhook_task.test_successful_request..add_webhookz /tasks/add r'r(success)statusretvalr9rrrWdictrIr r;r=)rrirhresponserrrtest_successful_requests z)test_webhook_task.test_successful_requestcCrd) NcSs(t|jd}t|jd}t||re)r%rfrrgrrr error_webhooks z.error_webhookz /tasks/errorrjrkfailurez)rnreasonr9rp)rrtrhrrrrrtest_failed_requests z%test_webhook_task.test_failed_requestN)r r!r"rsrwrrrrrcs rcc@s.eZdZd ddZddZddZdd ZdS) test_task_statusNcCst}tjj||||d|jt|d}t|||d}|tjjvrJtj |}| t |j dt |j dt ||d<t|j|d<||d<||t|ddS) N)r4task_id)idrnr$rr$r6r4r)rr r0 store_resultrVrW task_statusrqEXCEPTION_STATES get_resultr<strrHreprrr/rA)rrnr\r4uuidr?expectinstorerrrassertStatusForIss     z"test_task_status.assertStatusForIscC|tjddS)NzThe quick brown foxrr SUCCESSrrrr test_successztest_task_status.test_successcCs$ttd\}}|tj||dS)Nfoo)r7rrr FAILURE)rr6tbrrr test_failuresztest_task_status.test_failurecCs:ttd\}}ttt||\}}|tj||dS)NzResource not available)r7rr#rrr RETRY)roexc_r6rrrr test_retrysztest_task_status.test_retryr+)r r!r"rrrrrrrrrxs   rxc@s4eZdZddZddZddZddZd d Zd S) test_task_is_successfulcCsHt}t}tj||||jt|d}||d||didS)Nryr)r{executed)rr r0r|rVrWtask_is_successfulrA)rrnoutcomerr$r?rrrrs z)test_task_is_successful.assertStatusForIscCr)NTrrrrrrrz$test_task_is_successful.test_successcCrNF)rr PENDINGrrrr test_pendingrz$test_task_is_successful.test_pendingcCrr)rr rrrrrrrz$test_task_is_successful.test_failurecCrr)rr rrrrrrrz"test_task_is_successful.test_retryN)r r!r"rrrrrrrrrrs  r)1 __future__rrr2 functoolsrbilliard.einfordjango.core.urlresolversr django.httprdjango.test.testcasesr rBdjango.templater anyjsonr celeryr r celery.taskr celery.utilsrrdjcelery.viewsrdjcelery.tests.reqrr Exceptionrr#rr}rXr`r&r*r.r7r8rPr^rcrxrrrrrsB