Method _checkin_event_fixture

test/engine/test_pool.py:472–497

(self, _is_asyncio=False, _has_terminate=False)

        return p, canary

    def _checkin_event_fixture(self, _is_asyncio=False, _has_terminate=False):
        p = self._queuepool_fixture(
            _is_asyncio=_is_asyncio, _has_terminate=_has_terminate
        )
        canary = []

        @event.listens_for(p, "reset")
        def reset(conn, rec, state):
            canary.append(
                "reset_"
                f"{'rollback_ok' if state.asyncio_safe else 'no_rollback'}"
            )

        @event.listens_for(p, "checkin")
        def checkin(*arg, **kw):
            canary.append("checkin")

        @event.listens_for(p, "close_detached")
        def close_detached(*arg, **kw):
            canary.append("close_detached")

        @event.listens_for(p, "detach")
        def detach(*arg, **kw):
            canary.append("detach")

        return p, canary

    def _reset_event_fixture(self):
        p = self._queuepool_fixture()