Method _fixture

test/engine/test_reconnect.py:920–977

(self, explode_on_exec, initialize)

    # a generic "recovery".   maybe this test suite should have been
    # named "test_error_recovery".
    def _fixture(self, explode_on_exec, initialize):
        class DBAPIError(Exception):
            pass

        def MockDBAPI():
            def cursor():
                while True:
                    if explode_on_exec:
                        yield Mock(
                            description=[],
                            close=Mock(side_effect=DBAPIError("explode")),
                            execute=Mock(side_effect=DBAPIError("explode")),
                        )
                    else:
                        yield Mock(
                            description=[],
                            close=Mock(side_effect=Exception("explode")),
                        )

            def connect():
                while True:
                    yield Mock(
                        spec=["cursor", "commit", "rollback", "close"],
                        cursor=Mock(side_effect=cursor()),
                    )

            return Mock(
                Error=DBAPIError,
                paramstyle="qmark",
                connect=Mock(side_effect=connect()),
            )

        dbapi = MockDBAPI()

        from sqlalchemy.engine import default

        url = Mock(
            get_dialect=lambda: default.DefaultDialect,
            _get_entrypoint=lambda: default.DefaultDialect,
            _instantiate_plugins=lambda kwargs: (url, [], kwargs),
            translate_connect_args=lambda: {},
            query={},
        )
        eng = testing_engine(
            url, options=dict(module=dbapi, _initialize=initialize)
        )
        eng.pool.logger = Mock()

        def get_default_schema_name(connection):
            try:
                cursor = connection.connection.cursor()
                connection._cursor_execute(cursor, "statement", {})
                cursor.close()
            except exc.DBAPIError:
                util.warn("Exception attempting to detect")

        eng.dialect._get_default_schema_name = get_default_schema_name
        return eng