Method test_custom

test/engine/test_pool.py:2066–2097

(self, extra_step)

    @testing.combinations("detach", "invalidate", "return")
    def test_custom(self, extra_step):
        dbapi, p = self._fixture(reset_on_return=None)

        @event.listens_for(p, "reset")
        def custom_reset(dbapi_conn, record, reset_state):
            dbapi_conn.special_reset_method(reset_state)

        c1 = p.connect()

        if extra_step == "detach":
            c1.detach()
        elif extra_step == "invalidate":
            c1.invalidate()
        c1.close()

        special_event = mock.call.special_reset_method(
            PoolResetState(
                transaction_was_reset=False,
                terminate_only=extra_step == "detach",
                asyncio_safe=True,
            )
        )
        if extra_step == "detach":
            expected = [special_event, mock.call.close()]
        elif extra_step == "invalidate":
            expected = [mock.call.close()]
        else:
            expected = [special_event]
        eq_(dbapi.connect().mock_calls, expected)

        assert not dbapi.connect().rollback.called
        assert not dbapi.connect().commit.called

    @testing.combinations(True, False, argnames="assert_w_event")
    @testing.combinations(True, False, argnames="use_engine_transaction")