Method _first_connect_event_fixture

test/engine/test_pool.py:439–448

(self)

class PoolEventsTest(PoolTestBase):
    def _first_connect_event_fixture(self):
        p = self._queuepool_fixture()
        canary = []

        def first_connect(*arg, **kw):
            canary.append("first_connect")

        event.listen(p, "first_connect", first_connect)

        return p, canary

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