Ki0 dZddlmZddlmZddlmZddlmZdede edefzd e edfd zd e e efd zd e d zd e f dZ de edefzd e edfd zd e e efd zd e d zd e f dZd d d ddede edefzd e edfd zd e e efd zd e d zd d f dZd d d ddede edefzd e edfd zd e e efd zd e d zd d f dZ ddede edefzd zded d fdZded d fdZy )aTesting utilities for making assertions about scheduled tasks. Example usage: from docket import Docket, testing docket = Docket("redis://localhost:6379/0") # Schedule a task await docket.add(my_task)("arg1", kwarg1="value1") # Assert it's scheduled await testing.assert_task_scheduled(docket, my_task, args=("arg1",)) # After completion await worker.run_until_finished() await testing.assert_no_tasks(docket) )Callable)Any)Docket) Execution executionfunction.argsNkwargskeyreturnc t|tr|n |j}|j|k7ry||j|k7ry||j |k7ry|:|j D]'\}}||jvs|j||k7s'yy)z1Check if an execution matches the given criteria.FT) isinstancestr__name__ function_namer r itemsr )rrr r r rkvs Z/home/jay/workspace/scripts/.codegraph-venv/lib/python3.12/site-packages/docket/testing.py_matches_criteriars!+8S 9Hx?P?PM-/ 9==C/ INNd2LLN DAq (((I,<,C0(C50C5cXK|jd{}t|j}|D]r}t|||||st |t r|n |j }td|d|jd|jd|jd|jd y7w)a6Assert that no task matching the criteria is scheduled. Args: docket: The Docket instance to check function: The task function or function name (string) args: Optional tuple of positional arguments to match kwargs: Optional dict of keyword arguments to match (subset matching) key: Optional task key to match Raises: AssertionError: If a matching task is found Example: await assert_task_not_scheduled(docket, my_task) await assert_task_not_scheduled(docket, my_task, args=("foo",)) Nrz* found but should not be scheduled Found: r!r"r#r$) r&r'r(rrrrr)rr r r ) rrr r r r&r*rrs rassert_task_not_scheduledr1s0__&&HX__%I Y$ D(28S(AHxGXGXM  '#112&8JK#**-VIMM3DAG  'sB*B(-B*A$B*countc,K|jd{}t|j}|t|}d}n7t |t r|n |j tfd|D}d}||k7rtd|d|d|y7zw)aAssert the number of scheduled tasks matches the expected count. Args: docket: The Docket instance to check function: Optional task function or name to count (if None, counts all tasks) count: Expected number of tasks Raises: AssertionError: If the count doesn't match Example: await assert_task_count(docket, count=5) # All tasks await assert_task_count(docket, my_task, count=2) # Specific function Nz all tasksc3BK|]}|jk(sdyw)N)r).0r,rs r z$assert_task_count..sT1??m3S1Tsz tasks for z Expected  z, found ) r&r'r(lenrrrsumr))rrr2r&r* actual_count function_descrs @rassert_task_countr=s(__&&HX__%I9~ # $.x$=8CTCT TiTT $]O4 uyqx ~VWW'sBBA;Bc VK|jd{}t|j}|ro|Dcgc]9}d|jd|jd|j d|j d ;}}dj|}tdt|d |y7cc}ww) zAssert that no tasks are scheduled on the docket. Args: docket: The Docket instance to check Raises: AssertionError: If any tasks are scheduled Example: await assert_no_tasks(docket) Nr r!r"r#r$r%zExpected no tasks, found z task(s) scheduled: ) r&r'r(rr r r rr)r9)rr&r*r,r-r.s rassert_no_tasksr?s__&&HX__%I 1??#6!&&9QXXLquuiWX Y  IIk* 'I'77LYK X   '  s B)B"B)>B$4/B)$B))N)__doc__collections.abcrtypingr docket.docketrdocket.executionrrtupledictboolrrr/r1intr=r?rrrJs$$% &HS#X&& S/D  cNT !  t   <HS#X&& S/D  cNT ! t   4$($( 5 5HS#X&&5 S/D 5 cNT ! 5 t 5 5x$($( $ $HS#X&&$ S/D $ cNT ! $ t $ $R15"X "XHS#X&&-"X  "X  "XJ & T r