Method _pool_fixture

test/engine/test_reconnect.py:173–187

(self, pre_ping, setup_disconnect=True, pool_kw=None)

        self.dbapi = MockDBAPI()

    def _pool_fixture(self, pre_ping, setup_disconnect=True, pool_kw=None):
        dialect = default.DefaultDialect()
        dialect.dbapi = self.dbapi
        _pool = pool.QueuePool(
            creator=lambda: self.dbapi.connect("foo.db"),
            pre_ping=pre_ping,
            dialect=dialect,
            **(pool_kw if pool_kw else {}),
        )

        if setup_disconnect:
            dialect.is_disconnect = lambda e, conn, cursor: isinstance(
                e, MockDisconnect
            )
        return _pool

    def teardown_test(self):
        self.dbapi.dispose()