o 4@HhW@sjddlmZddlmZmZddlmZddlmZer"ddl m Z edd d Z dddZ dddZ dS)) annotations)AbstractContextManagercontextmanager) TYPE_CHECKING)_core) GeneratorexpectedboolreturnGenerator[None, None, None]ccsd}t}|j}|j}z(dV|r!|j|ks|j|kr!tdW|s2|j|ks.|j|kr4tddSdS|sF|j|ksB|j|krFtdw)z5Check if checkpoints are executed in a block of code.TNz'assert_checkpoints block did not yield!z$assert_no_checkpoints block yielded!)r current_task_cancel_points_schedule_pointsAssertionError)r __tracebackhide__task orig_cancel orig_schedulerY/home/air/sanwanet/gpt-api/venv/lib/python3.10/site-packages/trio/testing/_checkpoints.py_assert_yields_or_not s& rAbstractContextManager[None]cCs d}tdS)aUse as a context manager to check that the code inside the ``with`` block either exits with an exception or executes at least one :ref:`checkpoint `. Raises: AssertionError: if no checkpoint was executed. Example: Check that :func:`trio.sleep` is a checkpoint, even if it doesn't block:: with trio.testing.assert_checkpoints(): await trio.sleep(0) Trrrrrassert_checkpoints rcCs d}tdS)aUse as a context manager to check that the code inside the ``with`` block does not execute any :ref:`checkpoints `. Raises: AssertionError: if a checkpoint was executed. Example: Synchronous code never contains any checkpoints, but we can double-check that:: send_channel, receive_channel = trio.open_memory_channel(10) with trio.testing.assert_no_checkpoints(): send_channel.send_nowait(None) TFrrrrrassert_no_checkpoints4rrN)r r r r )r r) __future__r contextlibrrtypingrrcollections.abcrrrrrrrrs