o Ye@sZddlmZddlmZddlmZmZddlmZedd d Z dd dZ dddZ dS)) annotations) Generator)AbstractContextManagercontextmanager)_coreexpectedboolreturnGenerator[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!)rZ current_taskZ_cancel_pointsZ_schedule_pointsAssertionError)r__tracebackhide__ZtaskZ orig_cancelZ orig_scheduler=usr/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) Trr rrrassert_checkpointsrcCs 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_checkpoints1rrN)rr r r )r r) __future__rcollections.abcr contextlibrrrrrrrrrrs