Method test_per_connection

test/engine/test_transaction.py:1368–1409

(self)

            eq_(c2.get_isolation_level(), self._default_isolation_level())

    def test_per_connection(self):

        eng = testing_engine(
            options=dict(
                poolclass=(
                    QueuePool
                    if not testing.db.dialect.is_async
                    else AsyncAdaptedQueuePool
                ),
                pool_size=2,
                max_overflow=0,
            )
        )

        c1 = eng.connect()
        c1 = c1.execution_options(
            isolation_level=self._non_default_isolation_level()
        )
        c2 = eng.connect()
        eq_(
            eng.dialect.get_isolation_level(c1.connection.dbapi_connection),
            self._non_default_isolation_level(),
        )
        eq_(
            eng.dialect.get_isolation_level(c2.connection.dbapi_connection),
            self._default_isolation_level(),
        )
        c1.close()
        c2.close()
        c3 = eng.connect()
        eq_(
            eng.dialect.get_isolation_level(c3.connection.dbapi_connection),
            self._default_isolation_level(),
        )
        c4 = eng.connect()
        eq_(
            eng.dialect.get_isolation_level(c4.connection.dbapi_connection),
            self._default_isolation_level(),
        )

        c3.close()
        c4.close()

    def test_exception_in_transaction(self):
        eng = testing_engine()